Java 类javax.jms.JMSContext 实例源码

@Override
public JMSContext createContext(String username, String password, int sessionMode) {
    if (stopped.get()) {
        LOG.debug("JmsPoolConnectionFactory is stopped, skip create new connection.");
        return null;
    }

    if (!jmsContextSupported) {
        throw new JMSRuntimeException("Configured ConnectionFactory is not JMS 2+ capable");
    }

    if (isUseProviderJMSContext()) {
        return createProviderContext(username, password, sessionMode);
    } else {
        try {
            return newPooledConnectionContext(createJmsPoolConnection(username, password), sessionMode);
        } catch (JMSException e) {
            throw JMSExceptionSupport.createRuntimeException(e);
        }
    }
}

@Test(timeout = 60000)
public void testSetClientIDTwiceWithSameID() throws Exception {
    JMSContext context = cf.createContext();

    // test: call setClientID("newID") twice
    // this should be tolerated and not result in an exception
    context.setClientID("newID");

    try {
        context.setClientID("newID");
        context.start();
        context.close();
    } catch (IllegalStateRuntimeException ise) {
        LOG.error("Repeated calls to newID2.setClientID(\"newID\") caused " + ise.getMessage());
        fail("Repeated calls to newID2.setClientID(\"newID\") caused " + ise.getMessage());
    } finally {
        cf.stop();
    }

    LOG.debug("Test finished.");
}

@Test(timeout = 60000)
public void testSetClientIDTwiceWithDifferentID() throws Exception {
    JMSContext context = cf.createContext();

    // test: call setClientID() twice with different IDs
    // this should result in an IllegalStateException
    context.setClientID("newID1");
    try {
        context.setClientID("newID2");
        fail("calling Connection.setClientID() twice with different clientID must raise an IllegalStateException");
    } catch (IllegalStateRuntimeException ise) {
        LOG.debug("Correctly received " + ise);
    } finally {
        context.close();
        cf.stop();
    }

    LOG.debug("Test finished.");
}

@Test(timeout = 60000)
public void testSetClientIDAfterConnectionStart() throws Exception {
    JMSContext context = cf.createContext();

    // test: try to call setClientID() after start()
    // should result in an exception
    try {
        context.start();
        context.setClientID("newID3");
        fail("Calling setClientID() after start() mut raise a JMSException.");
    } catch (IllegalStateRuntimeException ise) {
        LOG.debug("Correctly received " + ise);
    } finally {
        context.close();
        cf.stop();
    }

    LOG.debug("Test finished.");
}

/**
 * Internal method to connect to MQ.
 *
 * @throws RetriableException Operation failed, but connector should continue to retry.
 * @throws ConnectException   Operation failed and connector should stop.
 */
private void connectInternal() throws ConnectException, RetriableException {
    if (connected) {
        return;
    }

    try {
        if (userName != null) {
            jmsCtxt = mqConnFactory.createContext(userName, password, JMSContext.SESSION_TRANSACTED);
        }
        else {
            jmsCtxt = mqConnFactory.createContext(JMSContext.SESSION_TRANSACTED);
        }            

        jmsProd = jmsCtxt.createProducer();
        jmsProd.setDeliveryMode(deliveryMode);
        jmsProd.setTimeToLive(timeToLive);
        connected = true;
    }
    catch (JMSRuntimeException jmse) {
        log.debug("JMS exception {}", jmse);
        handleException(jmse);
    }
}

/**
 * Internal method to connect to MQ.
 *
 * @throws RetriableException Operation failed, but connector should continue to retry.
 * @throws ConnectException   Operation failed and connector should stop.
 */
private void connectInternal() throws ConnectException, RetriableException {
    if (connected) {
        return;
    }

    if (closeNow.get()) {
        throw new ConnectException("Connection closing");
    }

    try {
        if (userName != null) {
            jmsCtxt = mqConnFactory.createContext(userName, password, JMSContext.SESSION_TRANSACTED);
        }
        else {
            jmsCtxt = mqConnFactory.createContext(JMSContext.SESSION_TRANSACTED);
        }            

        jmsCons = jmsCtxt.createConsumer(queue);
        connected = true;
    }
    catch (JMSRuntimeException jmse) {
        log.debug("JMS exception {}", jmse);
        handleException(jmse);
    }
}

/**
 * Convert a message into a Kafka Connect SourceRecord.
 * 
 * @param context            the JMS context to use for building messages
 * @param topic              the Kafka topic
 * @param messageBodyJms     whether to interpret MQ messages as JMS messages
 * @param message            the message
 * 
 * @return the Kafka Connect SourceRecord
 * 
 * @throws JMSException      Message could not be converted
 */
@Override public SourceRecord toSourceRecord(JMSContext context, String topic, boolean messageBodyJms, Message message) throws JMSException {
    byte[] payload;
    if (message instanceof BytesMessage) {
        payload = message.getBody(byte[].class);
    }
    else if (message instanceof TextMessage) {
        String s = message.getBody(String.class);
        payload = s.getBytes(UTF_8);
    }
    else {
        log.error("Unsupported JMS message type {}", message.getClass());
        throw new ConnectException("Unsupported JMS message type");
    }

    SchemaAndValue sv = converter.toConnectData(topic, payload);
    return new SourceRecord(null, null, topic, sv.schema(), sv.value());
}

private void connectInternal(Runnable command) {
    final JMSContext jmsCtx = mock(JMSContext.class);
    when(this.jmsCtxProvider.get()).thenReturn(jmsCtx);

    when(this.artemisConfig.getManagementNotificationAddress()).thenReturn(new SimpleString("notif"));
    final JMSConsumer consumer = mock(JMSConsumer.class);
    when(jmsCtx.createConsumer(any())).thenReturn(consumer);

    command.run();

    verify(this.jmsCtxProvider).get();
    verify(this.artemisConfig).getManagementNotificationAddress();
    verify(this.log).info("Connecting to broker for sourcing destination events.");
    verify(jmsCtx).createConsumer(any());
    verify(consumer).setMessageListener(this.eventProducer);
    verifyNoMoreInteractions(consumer);
}

/**
* Creates and sets the JMS connection and session variables.
* @throws Exception
*/
  protected void buildJMSResources() throws Exception {
    destroyJMSResources(true);
    if (!connectionInitialised) buildConnectionResources();

    //Build any JMS 2.0 thread resources here
      //Create the first JMSContext here, which can be used to create other JMSContexts for each thread
      if (transacted) {
        Log.logger.log(Level.FINE, "Using Transacted Mode");
        context = masterContext.createContext(JMSContext.SESSION_TRANSACTED);
      } else {
        int ackMode = Config.parms.getInt("am");
        Log.logger.log(Level.FINE, "Using Acknowledge Mode: {0}", ackMode);
        context = masterContext.createContext(ackMode);
      }
  }

public DestinationWrapper<Topic> lookupTopic(String topic, JMSContext context) throws JMSException, NamingException {

    if (usingJNDI || context == null) {
        if (autoCreateTopics) {
            Topic t = configureMQTopic((MQTopic)context.createTopic(topic));
            try {
                getInitialContext().bind(topic, t);
                Log.logger.fine( "Auto-created JNDI entry for: " + topic );
            } catch ( NameAlreadyBoundException e ) {
                // No op - already exists
            }
        } // end if
        return lookupTopicFromJNDI(topic);
    } else {
        return new DestinationWrapper<Topic>(topic, configureMQTopic((MQTopic) context.createTopic(topic)));            
    }
}

public QuoteDataBean pingTwoPhase(String symbol) throws Exception {

    if (Log.doTrace()) {
        Log.trace("TradeSLSBBean:pingTwoPhase", symbol);
    }

    QuoteDataBean quoteData = null;

    try (JMSContext queueContext = queueConnectionFactory.createContext();) {
        // Get a Quote and send a JMS message in a 2-phase commit
        quoteData = entityManager.find(QuoteDataBean.class, symbol);

        TextMessage message = queueContext.createTextMessage();

        message.setStringProperty("command", "ping");
        message.setLongProperty("publishTime", System.currentTimeMillis());
        message.setText("Ping message for queue java:comp/env/jms/TradeBrokerQueue sent from TradeSLSBBean:pingTwoPhase at " + new java.util.Date());
        queueContext.createProducer().send(tradeBrokerQueue, message);
    } catch (Exception e) {
        Log.error("TradeSLSBBean:pingTwoPhase -- exception caught", e);
    }

    return quoteData;
}

public static void main(final String[] args) throws Exception {
   // Step 2. Perfom a lookup on the queue
   Queue queue = ActiveMQJMSClient.createQueue("exampleQueue");

   // Step 4.Create a JMS Context using the try-with-resources statement
   try
      (
         // Even though ConnectionFactory is not closeable it would be nice to close an ActiveMQConnectionFactory
         ActiveMQConnectionFactory cf = new ActiveMQConnectionFactory();
         JMSContext jmsContext = cf.createContext()
      ) {
      // Step 5. create a jms producer
      JMSProducer jmsProducer = jmsContext.createProducer();

      // Step 6. Try sending a message, we don't have the appropriate privileges to do this so this will throw an exception
      jmsProducer.send(queue, "A Message from JMS2!");

      System.out.println("Received:" + jmsContext.createConsumer(queue).receiveBody(String.class));
   }
}

public static void main(final String[] args) throws Exception {
   // Instantiate the queue
   Queue queue = ActiveMQJMSClient.createQueue("exampleQueue");

   // Instantiate the ConnectionFactory (Using the default URI on this case)
   // Also instantiate the jmsContext
   // Using closeable interface
   try (ActiveMQConnectionFactory cf = new ActiveMQConnectionFactory();
        JMSContext jmsContext = cf.createContext()) {
      // Create a message producer, note that we can chain all this into one statement
      jmsContext.createProducer().setDeliveryMode(DeliveryMode.PERSISTENT).send(queue, "this is a string");

      // Create a Consumer and receive the payload of the message direct.
      String payLoad = jmsContext.createConsumer(queue).receiveBody(String.class);

      System.out.println("payLoad = " + payLoad);

   }

}

@Test
public void testSimpleSendNoXAJMSContext() throws Exception {
   Queue q = ActiveMQJMSClient.createQueue(MDBQUEUE);

   try (ClientSessionFactory sf = locator.createSessionFactory();
        ClientSession session = sf.createSession();
        ClientConsumer consVerify = session.createConsumer(MDBQUEUE);
        JMSContext jmsctx = qraConnectionFactory.createContext();
   ) {
      session.start();
      // These next 4 lines could be written in a single line however it makes difficult for debugging
      JMSProducer producer = jmsctx.createProducer();
      producer.setProperty("strvalue", "hello");
      TextMessage msgsend = jmsctx.createTextMessage("hello");
      producer.send(q, msgsend);

      ClientMessage msg = consVerify.receive(1000);
      assertNotNull(msg);
      assertEquals("hello", msg.getStringProperty("strvalue"));
   }
}

@Test
public void testSimpleSendNoXAJMSContext() throws Exception {
   Queue q = ActiveMQJMSClient.createQueue(MDBQUEUE);

   try (ClientSessionFactory sf = locator.createSessionFactory();
        ClientSession session = sf.createSession();
        ClientConsumer consVerify = session.createConsumer(MDBQUEUE);
        JMSContext jmsctx = qraConnectionFactory.createContext();
   ) {
      session.start();
      // These next 4 lines could be written in a single line however it makes difficult for debugging
      JMSProducer producer = jmsctx.createProducer();
      producer.setProperty("strvalue", "hello");
      TextMessage msgsend = jmsctx.createTextMessage("hello");
      producer.send(q, msgsend);

      ClientMessage msg = consVerify.receive(1000);
      assertNotNull(msg);
      assertEquals("hello", msg.getStringProperty("strvalue"));
   }
}

@Test
public void testDupsOK() {
   JMSContext ctx = addContext(cf.createContext(JMSContext.DUPS_OK_ACKNOWLEDGE));
   assertEquals(JMSContext.DUPS_OK_ACKNOWLEDGE, ctx.getSessionMode());

   ctx.close();
   ctx = addContext(cf.createContext(JMSContext.SESSION_TRANSACTED));
   assertEquals(JMSContext.SESSION_TRANSACTED, ctx.getSessionMode());

   ctx.close();
   ctx = addContext(cf.createContext(JMSContext.CLIENT_ACKNOWLEDGE));
   assertEquals(JMSContext.CLIENT_ACKNOWLEDGE, ctx.getSessionMode());

   ctx.close();
   ctx = addContext(cf.createContext(JMSContext.AUTO_ACKNOWLEDGE));
   assertEquals(JMSContext.AUTO_ACKNOWLEDGE, ctx.getSessionMode());

}

@Test
public void testGetAnotherContextFromIt() {
   JMSContext c2 = context.createContext(Session.DUPS_OK_ACKNOWLEDGE);
   Assert.assertNotNull(c2);
   Assert.assertEquals(Session.DUPS_OK_ACKNOWLEDGE, c2.getSessionMode());
   Message m2 = c2.createMessage();
   Assert.assertNotNull(m2);
   c2.close(); // should close its session, but not its (shared) connection
   try {
      c2.createMessage();
      Assert.fail("session should be closed...");
   } catch (JMSRuntimeException expected) {
      // expected
   }
   Message m1 = context.createMessage();
   Assert.assertNotNull("connection must be open", m1);
}

@Override
public JMSContext createContext(int sessionMode) {
   switch (sessionMode) {
      case Session.AUTO_ACKNOWLEDGE:
      case Session.CLIENT_ACKNOWLEDGE:
      case Session.DUPS_OK_ACKNOWLEDGE:
      case Session.SESSION_TRANSACTED:
      case ActiveMQJMSConstants.INDIVIDUAL_ACKNOWLEDGE:
      case ActiveMQJMSConstants.PRE_ACKNOWLEDGE:
         break;
      default:
         throw new JMSRuntimeException("Invalid ackmode: " + sessionMode);
   }
   refCounter.increment();

   return new ActiveMQJMSContext(this, sessionMode, threadAwareContext);
}

@Test
public void testRuntimeExceptionOnUnsubscribeFailure() throws JMSException {
    JmsConnection connection = Mockito.mock(JmsConnection.class);
    JmsSession session = Mockito.mock(JmsSession.class);
    Mockito.when(connection.createSession(Mockito.anyInt())).thenReturn(session);
    JmsContext context = new JmsContext(connection, JMSContext.CLIENT_ACKNOWLEDGE);

    Mockito.doThrow(IllegalStateException.class).when(session).unsubscribe(anyString());

    try {
        context.unsubscribe("subscription");
        fail("Should throw ISRE");
    } catch (IllegalStateRuntimeException isre) {
    } finally {
        context.close();
    }
}

@Test
public void testRuntimeExceptionOnCommitFailure() throws JMSException {
    JmsConnection connection = Mockito.mock(JmsConnection.class);
    JmsSession session = Mockito.mock(JmsSession.class);
    Mockito.when(connection.createSession(Mockito.anyInt())).thenReturn(session);
    JmsContext context = new JmsContext(connection, JMSContext.SESSION_TRANSACTED);

    Mockito.doThrow(IllegalStateException.class).when(session).commit();

    try {
        context.commit();
        fail("Should throw ISRE");
    } catch (IllegalStateRuntimeException isre) {
    } finally {
        context.close();
    }
}

JMSContext createJMSContext(TestAmqpPeer testPeer, boolean ssl, String optionsString, Symbol[] serverCapabilities, Map<Symbol, Object> serverProperties, boolean setClientId, int sessionMode) throws JMSException {
    testPeer.expectSaslPlain("guest", "guest");
    testPeer.expectOpen(serverProperties, serverCapabilities);

    // Each connection creates a session for managing temporary destinations etc
    testPeer.expectBegin();

    String remoteURI = buildURI(testPeer, ssl, optionsString);

    ConnectionFactory factory = new JmsConnectionFactory(remoteURI);
    JMSContext context = factory.createContext("guest", "guest", sessionMode);

    if (setClientId) {
        // Set a clientId to provoke the actual AMQP connection process to occur.
        context.setClientID("clientName");
    }

    assertNull(testPeer.getThrowable());

    return context;
}

@Test(timeout = 20000)
public void testCreateConsumer() throws Exception {
    try (TestAmqpPeer testPeer = new TestAmqpPeer();) {
        JMSContext context = testFixture.createJMSContext(testPeer);
        testPeer.expectBegin();
        testPeer.expectReceiverAttach();
        testPeer.expectLinkFlow();

        Queue queue = context.createQueue("test");
        JMSConsumer consumer = context.createConsumer(queue);
        assertNotNull(consumer);

        testPeer.expectEnd();
        testPeer.expectClose();
        context.close();

        testPeer.waitForAllHandlersToComplete(1000);
    }
}

@Test
public void testRuntimeExceptionOnCreateQueueFailure() throws JMSException {
    JmsConnection connection = Mockito.mock(JmsConnection.class);
    JmsSession session = Mockito.mock(JmsSession.class);
    Mockito.when(connection.createSession(Mockito.anyInt())).thenReturn(session);
    JmsContext context = new JmsContext(connection, JMSContext.CLIENT_ACKNOWLEDGE);

    Mockito.doThrow(IllegalStateException.class).when(session).createQueue(anyString());

    try {
        context.createQueue("test");
        fail("Should throw ISRE");
    } catch (IllegalStateRuntimeException isre) {
    } finally {
        context.close();
    }
}

@Test
public void testRuntimeExceptionOnCreateObjectMessageWithBody() throws JMSException {
    JmsConnection connection = Mockito.mock(JmsConnection.class);
    JmsSession session = Mockito.mock(JmsSession.class);
    Mockito.when(connection.createSession(Mockito.anyInt())).thenReturn(session);
    JmsContext context = new JmsContext(connection, JMSContext.CLIENT_ACKNOWLEDGE);

    Mockito.doThrow(IllegalStateException.class).when(session).createObjectMessage(any(Serializable.class));

    try {
        context.createObjectMessage(UUID.randomUUID());
        fail("Should throw ISRE");
    } catch (IllegalStateRuntimeException isre) {
    } finally {
        context.close();
    }
}

@Test
public void testRuntimeExceptionOnCreateQueueBrowserWithSelector() throws JMSException {
    JmsConnection connection = Mockito.mock(JmsConnection.class);
    JmsSession session = Mockito.mock(JmsSession.class);
    Mockito.when(connection.createSession(Mockito.anyInt())).thenReturn(session);
    Mockito.when(session.createTemporaryQueue()).thenReturn(new JmsTemporaryQueue());
    JmsContext context = new JmsContext(connection, JMSContext.AUTO_ACKNOWLEDGE);

    Mockito.doThrow(IllegalStateException.class).when(session).createBrowser(any(Queue.class), anyString());

    try {
        context.createBrowser(context.createTemporaryQueue(), "a == b");
        fail("Should throw ISRE");
    } catch (IllegalStateRuntimeException isre) {
    } finally {
        context.close();
    }

    Mockito.verify(session, Mockito.times(1)).createBrowser(any(Queue.class), anyString());
}

@Test
public void testRuntimeExceptionOnCreateQueueBrowser() throws JMSException {
    JmsConnection connection = Mockito.mock(JmsConnection.class);
    JmsSession session = Mockito.mock(JmsSession.class);
    Mockito.when(session.createTemporaryQueue()).thenReturn(new JmsTemporaryQueue());
    Mockito.when(connection.createSession(Mockito.anyInt())).thenReturn(session);
    JmsContext context = new JmsContext(connection, JMSContext.AUTO_ACKNOWLEDGE);

    Mockito.doThrow(IllegalStateException.class).when(session).createBrowser(any(Queue.class));

    try {
        context.createBrowser(context.createTemporaryQueue());
        fail("Should throw ISRE");
    } catch (IllegalStateRuntimeException isre) {
    } finally {
        context.close();
    }

    Mockito.verify(session, Mockito.times(1)).createBrowser(any(Queue.class));
}

@Test
public void testRuntimeExceptionOnCreateTemporaryQueueFailure() throws JMSException {
    JmsConnection connection = Mockito.mock(JmsConnection.class);
    JmsSession session = Mockito.mock(JmsSession.class);
    Mockito.when(connection.createSession(Mockito.anyInt())).thenReturn(session);
    JmsContext context = new JmsContext(connection, JMSContext.CLIENT_ACKNOWLEDGE);

    Mockito.doThrow(IllegalStateException.class).when(session).createTemporaryQueue();

    try {
        context.createTemporaryQueue();
        fail("Should throw ISRE");
    } catch (IllegalStateRuntimeException isre) {
    } finally {
        context.close();
    }
}

@Test
public void testRuntimeExceptionOnAcknowledgeFailure() throws JMSException {
    JmsConnection connection = Mockito.mock(JmsConnection.class);
    JmsSession session = Mockito.mock(JmsSession.class);
    Mockito.when(connection.createSession(Mockito.anyInt())).thenReturn(session);
    JmsContext context = new JmsContext(connection, JMSContext.CLIENT_ACKNOWLEDGE);

    Mockito.doThrow(IllegalStateException.class).when(session).acknowledge(ACK_TYPE.ACCEPTED);

    try {
        context.acknowledge();
        fail("Should throw ISRE");
    } catch (IllegalStateRuntimeException isre) {
    } finally {
        context.close();
    }
}

@Test(timeout = 20000)
public void testCreateProducer() throws Exception {
    try (TestAmqpPeer testPeer = new TestAmqpPeer();) {
        JMSContext context = testFixture.createJMSContext(testPeer, SERVER_ANONYMOUS_RELAY);
        testPeer.expectBegin();
        testPeer.expectSenderAttach();

        JMSProducer producer = context.createProducer();
        assertNotNull(producer);

        testPeer.expectEnd();
        testPeer.expectClose();
        context.close();

        testPeer.waitForAllHandlersToComplete(1000);
    }
}

@Test(timeout = 20000)
public void testJMSProducerHasDefaultConfiguration() throws Exception {
    try (TestAmqpPeer testPeer = new TestAmqpPeer();) {
        JMSContext context = testFixture.createJMSContext(testPeer, SERVER_ANONYMOUS_RELAY);
        testPeer.expectBegin();
        testPeer.expectSenderAttach();

        JMSProducer producer = context.createProducer();
        assertNotNull(producer);

        assertEquals(Message.DEFAULT_DELIVERY_DELAY, producer.getDeliveryDelay());
        assertEquals(Message.DEFAULT_DELIVERY_MODE, producer.getDeliveryMode());
        assertEquals(Message.DEFAULT_PRIORITY, producer.getPriority());
        assertEquals(Message.DEFAULT_TIME_TO_LIVE, producer.getTimeToLive());

        testPeer.expectEnd();
        testPeer.expectClose();
        context.close();

        testPeer.waitForAllHandlersToComplete(1000);
    }
}

@Test
public void testRuntimeExceptionOnCreateMessage() throws JMSException {
    JmsConnection connection = Mockito.mock(JmsConnection.class);
    JmsSession session = Mockito.mock(JmsSession.class);
    Mockito.when(connection.createSession(Mockito.anyInt())).thenReturn(session);
    JmsContext context = new JmsContext(connection, JMSContext.CLIENT_ACKNOWLEDGE);

    Mockito.doThrow(IllegalStateException.class).when(session).createMessage();

    try {
        context.createMessage();
        fail("Should throw ISRE");
    } catch (IllegalStateRuntimeException isre) {
    } finally {
        context.close();
    }
}

@Test(timeout = 20000)
public void testCreateContextWithTransactedSessionMode() throws Exception {
    Binary txnId = new Binary(new byte[]{ (byte) 5, (byte) 6, (byte) 7, (byte) 8});

    try (TestAmqpPeer testPeer = new TestAmqpPeer();) {
        JMSContext context = testFixture.createJMSContext(testPeer, JMSContext.SESSION_TRANSACTED);
        assertEquals(JMSContext.SESSION_TRANSACTED, context.getSessionMode());

        // Session should be created and a coordinator should be attached since this
        // should be a TX session, then a new TX is declared, once closed the TX should
        // be discharged as a roll back.
        testPeer.expectBegin();
        testPeer.expectCoordinatorAttach();
        testPeer.expectDeclare(txnId);
        testPeer.expectDischarge(txnId, true);
        testPeer.expectEnd();
        testPeer.expectClose();

        context.createTopic("TopicName");

        context.close();

        testPeer.waitForAllHandlersToComplete(1000);
    }
}

@Test(timeout = 20000)
public void testOnlyOneProducerCreatedInSingleContext() throws Exception {
    try (TestAmqpPeer testPeer = new TestAmqpPeer();) {
        JMSContext context = testFixture.createJMSContext(testPeer, SERVER_ANONYMOUS_RELAY);
        assertEquals(JMSContext.AUTO_ACKNOWLEDGE, context.getSessionMode());
        testPeer.expectBegin();
        testPeer.expectSenderAttach();

        // One producer created should send an attach.
        JMSProducer producer1 = context.createProducer();
        assertNotNull(producer1);

        // An additional one should not result in an attach
        JMSProducer producer2 = context.createProducer();
        assertNotNull(producer2);

        testPeer.expectEnd();
        testPeer.expectClose();
        context.close();

        testPeer.waitForAllHandlersToComplete(1000);
    }
}

@Test
public void testAutoStartOnDoesStartTheConnectionMessageConsumer() throws JMSException {
    JmsConnection connection = Mockito.mock(JmsConnection.class);
    JmsSession session = Mockito.mock(JmsSession.class);
    JmsMessageConsumer consumer = Mockito.mock(JmsMessageConsumer.class);

    Mockito.when(session.createTemporaryQueue()).thenReturn(new JmsTemporaryQueue());
    Mockito.when(connection.createSession(anyInt())).thenReturn(session);
    Mockito.when(session.createConsumer(any(Destination.class))).thenReturn(consumer);

    JmsContext context = new JmsContext(connection, JMSContext.AUTO_ACKNOWLEDGE);
    context.setAutoStart(true);

    try {
        context.createConsumer(context.createTemporaryQueue());
    } finally {
        context.close();
    }

    Mockito.verify(session, Mockito.times(1)).createConsumer(any(Destination.class));
    Mockito.verify(connection, Mockito.times(1)).start();
}

@Test
public void testAutoStartOffDoesNotStartTheConnectionMessageConsumer() throws JMSException {
    JmsConnection connection = Mockito.mock(JmsConnection.class);
    JmsSession session = Mockito.mock(JmsSession.class);
    JmsMessageConsumer consumer = Mockito.mock(JmsMessageConsumer.class);

    Mockito.when(connection.createSession(anyInt())).thenReturn(session);
    Mockito.when(session.createConsumer(any(Destination.class))).thenReturn(consumer);
    Mockito.when(session.createTemporaryQueue()).thenReturn(new JmsTemporaryQueue());

    JmsContext context = new JmsContext(connection, JMSContext.AUTO_ACKNOWLEDGE);
    context.setAutoStart(false);

    try {
        context.createConsumer(context.createTemporaryQueue());
    } finally {
        context.close();
    }

    Mockito.verify(session, Mockito.times(1)).createConsumer(any(Destination.class));
    Mockito.verify(connection, Mockito.times(0)).start();
}

@Test
public void testRuntimeExceptionOnCreateSharedDurableConsumerSelectorNoLocal() throws JMSException {
    JmsConnection connection = Mockito.mock(JmsConnection.class);
    JmsSession session = Mockito.mock(JmsSession.class);
    Mockito.when(session.createTemporaryTopic()).thenReturn(new JmsTemporaryTopic());
    Mockito.when(connection.createSession(Mockito.anyInt())).thenReturn(session);
    JmsContext context = new JmsContext(connection, JMSContext.AUTO_ACKNOWLEDGE);

    Mockito.doThrow(IllegalStateException.class).when(session).
    createSharedDurableConsumer(any(Topic.class), anyString(), anyString());

    try {
        context.createSharedDurableConsumer(context.createTemporaryTopic(), "name", "a = b");
        fail("Should throw ISRE");
    } catch (IllegalStateRuntimeException isre) {
    } finally {
        context.close();
    }

    Mockito.verify(session, Mockito.times(1)).createSharedDurableConsumer(any(Topic.class), anyString(), anyString());
}

@Test
public void testAutoStartOnDoesStartTheConnectionMessageConsumerSelectorNoLocal() throws JMSException {
    JmsConnection connection = Mockito.mock(JmsConnection.class);
    JmsSession session = Mockito.mock(JmsSession.class);
    JmsMessageConsumer consumer = Mockito.mock(JmsMessageConsumer.class);

    Mockito.when(connection.createSession(Mockito.anyInt())).thenReturn(session);
    Mockito.when(session.createTemporaryTopic()).thenReturn(new JmsTemporaryTopic());
    Mockito.when(session.createConsumer(any(Destination.class), anyString(), anyBoolean())).thenReturn(consumer);

    JmsContext context = new JmsContext(connection, JMSContext.AUTO_ACKNOWLEDGE);
    context.setAutoStart(true);

    try {
        context.createConsumer(context.createTemporaryTopic(), "a = b", true);
    } finally {
        context.close();
    }

    Mockito.verify(session, Mockito.times(1)).createConsumer(any(Topic.class), anyString(), anyBoolean());
    Mockito.verify(connection, Mockito.times(1)).start();
}

@Test
public void testAutoStartOffDoesNotStartTheConnectionMessageConsumerSelectorNoLocal() throws JMSException {
    JmsConnection connection = Mockito.mock(JmsConnection.class);
    JmsSession session = Mockito.mock(JmsSession.class);
    JmsMessageConsumer consumer = Mockito.mock(JmsMessageConsumer.class);

    Mockito.when(connection.createSession(Mockito.anyInt())).thenReturn(session);
    Mockito.when(session.createTemporaryTopic()).thenReturn(new JmsTemporaryTopic());
    Mockito.when(session.createConsumer(any(Destination.class), anyString(), anyBoolean())).thenReturn(consumer);

    JmsContext context = new JmsContext(connection, JMSContext.AUTO_ACKNOWLEDGE);
    context.setAutoStart(false);

    try {
        context.createConsumer(context.createTemporaryTopic(), "a = b", true);
    } finally {
        context.close();
    }

    Mockito.verify(session, Mockito.times(1)).createConsumer(any(Topic.class), anyString(), anyBoolean());
    Mockito.verify(connection, Mockito.times(0)).start();
}

@Test
public void testAutoStartOnDoesStartTheConnectionDurableMessageConsumer() throws JMSException {
    JmsConnection connection = Mockito.mock(JmsConnection.class);
    JmsSession session = Mockito.mock(JmsSession.class);
    JmsMessageConsumer consumer = Mockito.mock(JmsMessageConsumer.class);

    Mockito.when(connection.createSession(Mockito.anyInt())).thenReturn(session);
    Mockito.when(session.createDurableConsumer(any(Topic.class), anyString())).thenReturn(consumer);
    Mockito.when(session.createTemporaryTopic()).thenReturn(new JmsTemporaryTopic());

    JmsContext context = new JmsContext(connection, JMSContext.AUTO_ACKNOWLEDGE);
    context.setAutoStart(true);

    try {
        context.createDurableConsumer(context.createTemporaryTopic(), "name");
    } finally {
        context.close();
    }

    Mockito.verify(session, Mockito.times(1)).createDurableConsumer(any(Topic.class), anyString());
    Mockito.verify(connection, Mockito.times(1)).start();
}

@Test
public void testRuntimeExceptionOnCreateTopicFailure() throws JMSException {
    JmsConnection connection = Mockito.mock(JmsConnection.class);
    JmsSession session = Mockito.mock(JmsSession.class);
    Mockito.when(connection.createSession(Mockito.anyInt())).thenReturn(session);
    JmsContext context = new JmsContext(connection, JMSContext.CLIENT_ACKNOWLEDGE);

    Mockito.doThrow(IllegalStateException.class).when(session).createTopic(anyString());

    try {
        context.createTopic("test");
        fail("Should throw ISRE");
    } catch (IllegalStateRuntimeException isre) {
    } finally {
        context.close();
    }
}