下面列出了怎么用javax.jms.ConnectionConsumer的API类实例代码及写法，或者点击链接到github查看源代码。
/**
* Create a connection consumer -- throws IllegalStateException
*
* @param topic The topic
* @param messageSelector The message selector
* @param sessionPool The session pool
* @param maxMessages The number of max messages
* @return The connection consumer
* @throws JMSException Thrown if an error occurs
*/
@Override
public ConnectionConsumer createConnectionConsumer(final Topic topic,
final String messageSelector,
final ServerSessionPool sessionPool,
final int maxMessages) throws JMSException {
if (ActiveMQRALogger.LOGGER.isTraceEnabled()) {
ActiveMQRALogger.LOGGER.trace("createConnectionConsumer(" + topic +
", " +
messageSelector +
", " +
sessionPool +
", " +
maxMessages +
")");
}
throw new IllegalStateException(ISE);
}
/**
* Create a durable connection consumer -- throws IllegalStateException
*
* @param topic The topic
* @param subscriptionName The subscription name
* @param messageSelector The message selector
* @param sessionPool The session pool
* @param maxMessages The number of max messages
* @return The connection consumer
* @throws JMSException Thrown if an error occurs
*/
@Override
public ConnectionConsumer createDurableConnectionConsumer(final Topic topic,
final String subscriptionName,
final String messageSelector,
final ServerSessionPool sessionPool,
final int maxMessages) throws JMSException {
if (ActiveMQRALogger.LOGGER.isTraceEnabled()) {
ActiveMQRALogger.LOGGER.trace("createConnectionConsumer(" + topic +
", " +
subscriptionName +
", " +
messageSelector +
", " +
sessionPool +
", " +
maxMessages +
")");
}
throw new IllegalStateException(ISE);
}
/**
* Create a connection consumer -- throws IllegalStateException
*
* @param destination The destination
* @param name The name
* @param pool The session pool
* @param maxMessages The number of max messages
* @return The connection consumer
* @throws JMSException Thrown if an error occurs
*/
@Override
public ConnectionConsumer createConnectionConsumer(final Destination destination,
final String name,
final ServerSessionPool pool,
final int maxMessages) throws JMSException {
if (ActiveMQRALogger.LOGGER.isTraceEnabled()) {
ActiveMQRALogger.LOGGER.trace("createConnectionConsumer(" + destination +
", " +
name +
", " +
pool +
", " +
maxMessages +
")");
}
throw new IllegalStateException(ISE);
}
@Override
public ConnectionConsumer createDurableConnectionConsumer(final Topic topic,
final String subscriptionName,
final String messageSelector,
final ServerSessionPool sessionPool,
final int maxMessages) throws JMSException {
checkClosed();
// As spec. section 4.11
if (connectionType == ActiveMQConnection.TYPE_QUEUE_CONNECTION) {
String msg = "Cannot create a durable connection consumer on a QueueConnection";
throw new javax.jms.IllegalStateException(msg);
}
checkTempQueues(topic);
// We offer RA, so no need for this
return null;
}
@Test(timeout = 20000)
public void testCreateConnectionConsumer() throws Exception {
try (TestAmqpPeer testPeer = new TestAmqpPeer();) {
JmsServerSessionPool sessionPool = new JmsServerSessionPool();
Connection connection = testFixture.establishConnecton(testPeer);
// No additional Begin calls as there's no Session created for a Connection Consumer
testPeer.expectReceiverAttach();
testPeer.expectLinkFlow();
Queue queue = new JmsQueue("myQueue");
ConnectionConsumer consumer = connection.createConnectionConsumer(queue, null, sessionPool, 100);
testPeer.expectDetach(true, true, true);
consumer.close();
testPeer.expectClose();
connection.close();
testPeer.waitForAllHandlersToComplete(1000);
}
}
@JMS2_0
public ConnectionConsumer createSharedDurableConnectionConsumer(Topic topic,
String subscriptionName, String messageSelector, ServerSessionPool sessionPool,
int maxMessages) throws JMSException {
ConnectionConsumer cc =
delegate.createSharedDurableConnectionConsumer(topic, subscriptionName, messageSelector,
sessionPool, maxMessages);
return TracingConnectionConsumer.create(cc, jmsTracing);
}
public ConnectionConsumer createDurableConnectionConsumer(Topic topic,
String subscriptionName, String messageSelector,
ServerSessionPool sessionPool, int maxMessages) throws JMSException {
this.checkStatus();
return null;
}
@Override
public ConnectionConsumer createConnectionConsumer(
Topic topic,
String messageSelector,
ServerSessionPool sessionPool,
int maxMessages ) throws JMSException {
return addConnectionConsumer(topicConnection.createConnectionConsumer(topic,
messageSelector,
sessionPool,
maxMessages));
}
@Override
public ConnectionConsumer createConnectionConsumer(
Destination destination,
String messageSelector,
ServerSessionPool sessionPool,
int maxMessages ) throws JMSException {
return connection.createConnectionConsumer(destination, messageSelector, sessionPool, maxMessages);
}
@Override
public ConnectionConsumer createDurableConnectionConsumer(
Topic topic,
String subscriptionName,
String messageSelector,
ServerSessionPool sessionPool,
int maxMessages ) throws JMSException {
return connection.createDurableConnectionConsumer(topic,
subscriptionName,
messageSelector,
sessionPool,
maxMessages);
}
@Override
public ConnectionConsumer createConnectionConsumer(
Queue queue,
String messageSelector,
ServerSessionPool sessionPool,
int maxMessages ) throws JMSException {
return addConnectionConsumer(queueConnection.createConnectionConsumer(queue,
messageSelector,
sessionPool,
maxMessages));
}
@Override
public ConnectionConsumer createConnectionConsumer(
Queue queue,
String messageSelector,
ServerSessionPool sessionPool,
int maxMessages ) throws JMSException {
return addConnectionConsumer( ((QueueConnection) connection).createConnectionConsumer(queue,
messageSelector,
sessionPool,
maxMessages));
}
@Override
public ConnectionConsumer createConnectionConsumer(
Topic topic,
String messageSelector,
ServerSessionPool sessionPool,
int maxMessages ) throws JMSException {
return addConnectionConsumer( ((TopicConnection) connection).createConnectionConsumer(topic,
messageSelector,
sessionPool,
maxMessages));
}
@Override
public ConnectionConsumer createSharedConnectionConsumer(Topic arg0,
String arg1, String arg2, ServerSessionPool arg3, int arg4)
throws JMSException {
// TODO Auto-generated method stub
return null;
}
@Override
public ConnectionConsumer createSharedDurableConnectionConsumer(Topic arg0,
String arg1, String arg2, ServerSessionPool arg3, int arg4)
throws JMSException {
// TODO Auto-generated method stub
return null;
}
@Test(timeout = 20000)
public void testOnExceptionFiredOnSessionPoolFailure() throws Exception {
final CountDownLatch exceptionFired = new CountDownLatch(1);
try (TestAmqpPeer testPeer = new TestAmqpPeer();) {
Connection connection = testFixture.establishConnecton(testPeer);
connection.setExceptionListener(new ExceptionListener() {
@Override
public void onException(JMSException exception) {
exceptionFired.countDown();
}
});
connection.start();
JmsFailingServerSessionPool sessionPool = new JmsFailingServerSessionPool();
// Now the Connection consumer arrives and we give it a message
// to be dispatched to the server session.
DescribedType amqpValueNullContent = new AmqpValueDescribedType(null);
testPeer.expectReceiverAttach();
testPeer.expectLinkFlowRespondWithTransfer(null, null, null, null, amqpValueNullContent);
Queue queue = new JmsQueue("myQueue");
ConnectionConsumer consumer = connection.createConnectionConsumer(queue, null, sessionPool, 100);
assertTrue("Exception should have been fired", exceptionFired.await(5, TimeUnit.SECONDS));
testPeer.expectDetach(true, true, true);
testPeer.expectDispositionThatIsReleasedAndSettled();
consumer.close();
testPeer.expectClose();
connection.close();
testPeer.waitForAllHandlersToComplete(1000);
}
}
/**
* Create a connection consumer -- throws IllegalStateException
*
* @param destination The destination
* @param pool The session pool
* @param maxMessages The number of max messages
* @return The connection consumer
* @throws JMSException Thrown if an error occurs
*/
public ConnectionConsumer createConnectionConsumer(final Destination destination,
final ServerSessionPool pool,
final int maxMessages) throws JMSException {
if (ActiveMQRALogger.LOGGER.isTraceEnabled()) {
ActiveMQRALogger.LOGGER.trace("createConnectionConsumer(" + destination +
", " +
pool +
", " +
maxMessages +
")");
}
throw new IllegalStateException(ISE);
}
@Override
public ConnectionConsumer createSharedConnectionConsumer(Topic topic,
String subscriptionName,
String messageSelector,
ServerSessionPool sessionPool,
int maxMessages) throws JMSException {
if (ActiveMQRALogger.LOGGER.isTraceEnabled()) {
ActiveMQRALogger.LOGGER.trace("createSharedConnectionConsumer(" + topic + ", " + subscriptionName + ", " +
messageSelector + ", " + sessionPool + ", " + maxMessages + ")");
}
throw new IllegalStateException(ISE);
}
@Override
public ConnectionConsumer createSharedDurableConnectionConsumer(Topic topic,
String subscriptionName,
String messageSelector,
ServerSessionPool sessionPool,
int maxMessages) throws JMSException {
if (ActiveMQRALogger.LOGGER.isTraceEnabled()) {
ActiveMQRALogger.LOGGER.trace("createSharedDurableConnectionConsumer(" + topic + ", " + subscriptionName +
", " + messageSelector + ", " + sessionPool + ", " + maxMessages + ")");
}
throw new IllegalStateException(ISE);
}
@Test(timeout = 20000)
public void testOnExceptionFiredOnServerSessionFailure() throws Exception {
final CountDownLatch exceptionFired = new CountDownLatch(1);
try (TestAmqpPeer testPeer = new TestAmqpPeer();) {
Connection connection = testFixture.establishConnecton(testPeer);
connection.setExceptionListener(new ExceptionListener() {
@Override
public void onException(JMSException exception) {
exceptionFired.countDown();
}
});
connection.start();
JmsServerSessionPool sessionPool = new JmsServerSessionPool(new JmsFailingServerSession());
// Now the Connection consumer arrives and we give it a message
// to be dispatched to the server session.
DescribedType amqpValueNullContent = new AmqpValueDescribedType(null);
testPeer.expectReceiverAttach();
testPeer.expectLinkFlowRespondWithTransfer(null, null, null, null, amqpValueNullContent);
Queue queue = new JmsQueue("myQueue");
ConnectionConsumer consumer = connection.createConnectionConsumer(queue, null, sessionPool, 100);
assertTrue("Exception should have been fired", exceptionFired.await(5, TimeUnit.SECONDS));
testPeer.expectDetach(true, true, true);
testPeer.expectDispositionThatIsReleasedAndSettled();
consumer.close();
testPeer.expectClose();
connection.close();
testPeer.waitForAllHandlersToComplete(1000);
}
}
@Override
public ConnectionConsumer createConnectionConsumer(final Queue queue,
final String messageSelector,
final ServerSessionPool sessionPool,
final int maxMessages) throws JMSException {
checkClosed();
checkTempQueues(queue);
return null;
}
@Override
public ConnectionConsumer createSharedConnectionConsumer(Topic topic,
String subscriptionName,
String messageSelector,
ServerSessionPool sessionPool,
int maxMessages) throws JMSException {
return null; // we offer RA
}
@Override
public ConnectionConsumer createSharedDurableConnectionConsumer(Topic topic,
String subscriptionName,
String messageSelector,
ServerSessionPool sessionPool,
int maxMessages) throws JMSException {
return null; // we offer RA
}
public ConnectionConsumer createConnectionConsumer(Destination destination,
String messageSelector, ServerSessionPool sessionPool,
int maxMessages) throws JMSException {
this.checkStatus();
return null;
}
@Override
public ConnectionConsumer createSharedDurableConnectionConsumer(Topic topic, String subscriptionName, String messageSelector, ServerSessionPool sessionPool, int maxMessages) throws JMSException {
checkClosedOrFailed();
createJmsConnection();
return createConnectionConsumer(topic, messageSelector, sessionPool, maxMessages, subscriptionName, true, true);
}
@Override
public ConnectionConsumer createDurableConnectionConsumer(Topic topic, String subscriptionName, String messageSelector, ServerSessionPool sessionPool, int maxMessages) throws JMSException {
checkClosedOrFailed();
createJmsConnection();
return createConnectionConsumer(topic, messageSelector, sessionPool, maxMessages, subscriptionName, true, false);
}
@Override
public ConnectionConsumer createConnectionConsumer(Destination destination, String messageSelector, ServerSessionPool sessionPool, int maxMessages) throws JMSException {
checkClosedOrFailed();
createJmsConnection();
return createConnectionConsumer(destination, messageSelector, sessionPool, maxMessages, null, false, false);
}
@Override
public ConnectionConsumer createConnectionConsumer(Queue queue, String messageSelector, ServerSessionPool sessionPool, int maxMessages) throws JMSException {
checkClosedOrFailed();
createJmsConnection();
return createConnectionConsumer(queue, messageSelector, sessionPool, maxMessages, null, false, false);
}
@JMS2_0
public ConnectionConsumer createSharedConnectionConsumer(Topic topic, String subscriptionName,
String messageSelector, ServerSessionPool sessionPool, int maxMessages) throws JMSException {
ConnectionConsumer cc =
delegate.createSharedConnectionConsumer(topic, subscriptionName, messageSelector,
sessionPool, maxMessages);
return TracingConnectionConsumer.create(cc, jmsTracing);
}
@Override
public ConnectionConsumer createConnectionConsumer(Destination destination, String paramName, ServerSessionPool serverSessionPool, int i) throws JMSException {
return null;
}