Java 类javax.jms.TopicConnectionFactory 实例源码

/**
 * @param args
 * @throws JMSException
 * @throws IOException
 */
public static void main(String[] args) throws JMSException, IOException {

    if (args.length != 1) {
        System.out.println("User Name is required....");
    } else {
        userId = args[0];
        ApplicationContext ctx = new ClassPathXmlApplicationContext(
                "com/springtraining/jms/spring-config.xml");

        BasicJMSChat basicJMSChat = (BasicJMSChat) ctx
                .getBean("basicJMSChat");
        TopicConnectionFactory topicConnectionFactory = (TopicConnectionFactory) basicJMSChat.chatJMSTemplate
                .getConnectionFactory();
        TopicConnection tc = topicConnectionFactory.createTopicConnection();

        basicJMSChat.publish(tc,  basicJMSChat.chatTopic, userId);
        basicJMSChat.subscribe(tc,  basicJMSChat.chatTopic, basicJMSChat);
    }
}

@Before
public void setUp() throws Exception {
  eventBus = new SimpleEventBus();
  cut = new JmsPublisher(eventBus);
  connectionFactory = mock(TopicConnectionFactory.class);
  publisher = mock(TopicPublisher.class);
  topic = mock(Topic.class);
  converter = mock(JmsMessageConverter.class);
  cut.setConnectionFactory(connectionFactory);
  cut.setTopic(topic);
  cut.setTransacted(true);
  cut.setMessageConverter(converter);
  cut.setPersistent(false);
  cut.postConstruct();
  cut.start();
}

@Test
public void testWithTopicConnectionFactoryAndJms11Usage() throws JMSException {
    TopicConnectionFactory cf = mock(TopicConnectionFactory.class);
    TopicConnection con = mock(TopicConnection.class);

    given(cf.createConnection()).willReturn(con);

    SingleConnectionFactory scf = new SingleConnectionFactory(cf);
    Connection con1 = scf.createConnection();
    Connection con2 = scf.createConnection();
    con1.start();
    con2.start();
    con1.close();
    con2.close();
    scf.destroy();  // should trigger actual close

    verify(con).start();
    verify(con).stop();
    verify(con).close();
    verifyNoMoreInteractions(con);
}

@Test
public void testWithTopicConnectionFactoryAndJms102Usage() throws JMSException {
    TopicConnectionFactory cf = mock(TopicConnectionFactory.class);
    TopicConnection con = mock(TopicConnection.class);

    given(cf.createTopicConnection()).willReturn(con);

    SingleConnectionFactory scf = new SingleConnectionFactory(cf);
    Connection con1 = scf.createTopicConnection();
    Connection con2 = scf.createTopicConnection();
    con1.start();
    con2.start();
    con1.close();
    con2.close();
    scf.destroy();  // should trigger actual close

    verify(con).start();
    verify(con).stop();
    verify(con).close();
    verifyNoMoreInteractions(con);
}

public TopicConnection getTopicConnection(TopicConnectionFactory tcf, String uniqueID )
        throws JMSException {

    final TopicConnection tc;
    final String username = Config.parms.getString("us");
    if (username != null && username.length() != 0) {
        Log.logger.log(Level.INFO, "getTopicConnection(): authenticating as \"" + username + "\"");
        final String password = Config.parms.getString("pw");
        tc = tcf.createTopicConnection(username, password);
    } else {
        tc = tcf.createTopicConnection();
    }

    if (durable) {
        // Note: change signature to match getConnection
        setDurableConnectionId( tc, ((WorkerThread)Thread.currentThread()), uniqueID );
    } // end if durable

    return tc;

}

public TopicConnectionFactory getTopicConnectionFactory(String contextUrl)
        throws ServiceLocatorException {
    TopicConnectionFactory factory = (TopicConnectionFactory) topicConnFactoryCache
            .get(contextUrl == null ? THIS_SERVER : contextUrl);
    if (factory == null) {
        try {
            factory = jmsProvider.getTopicConnectionFactory(namingProvider, contextUrl);
            if (contextUrl == null)
                topicConnFactoryCache.put(THIS_SERVER, factory);
            else
                topicConnFactoryCache.put(contextUrl, factory);
        }
        catch (Exception ex) {
            throw new ServiceLocatorException(-1, ex.getMessage(), ex);
        }
    }
    return factory;
}

public JMSSink(final String tcfBindingName, final String topicBindingName, final String username,
               final String password) {

   try {
      final Context ctx = new InitialContext();
      final TopicConnectionFactory topicConnectionFactory = (TopicConnectionFactory) lookup(ctx,
              tcfBindingName);

      final TopicConnection topicConnection =
              topicConnectionFactory.createTopicConnection(username,
                      password);
      topicConnection.start();

      final TopicSession topicSession = topicConnection.createTopicSession(false,
              Session.AUTO_ACKNOWLEDGE);

      final Topic topic = (Topic) ctx.lookup(topicBindingName);

      final TopicSubscriber topicSubscriber = topicSession.createSubscriber(topic);

      topicSubscriber.setMessageListener(this);

   } catch (final Exception e) {
      logger.error("Could not read JMS message.", e);
   }
}

protected void fireAndForget(HttpServletRequest request, HttpServletResponse response, String userMessage) {
    try {
        InitialContext ctx = new InitialContext();
        TopicConnectionFactory qconFactory = (TopicConnectionFactory) ctx.lookup(CONN_FACTORY);

        ConnectionContextFactory ccf = new ConnectionContextFactory(qconFactory);
        ProducerConnectionContext pcc = ccf.createProducerConnectionContext(new Endpoint(Endpoint.Type.TOPIC,
                TOPIC_NAME));

        SimpleBasicMessage msg = new SimpleBasicMessage(userMessage);
        MessageId mid = new MessageProcessor().send(pcc, msg, FNF_HEADER);

        PrintWriter out = response.getWriter();
        out.println("<h1>Fire and Forget</h1>");
        out.println("<p>BasicMessage Sent [" + msg + "]</p>");
        out.println("<p>(messageId=" + mid + ")</p>");
    } catch (Exception e) {
        e.printStackTrace();
    }
}

/**
 * In addition to checking if the connection factory is set, make sure
 * that the supplied connection factory is of the appropriate type for
 * the specified destination type: QueueConnectionFactory for queues,
 * and TopicConnectionFactory for topics.
 */
public void afterPropertiesSet() {
    super.afterPropertiesSet();

    // Make sure that the ConnectionFactory passed is consistent.
    // Some provider implementations of the ConnectionFactory interface
    // implement both domain interfaces under the cover, so just check if
    // the selected domain is consistent with the type of connection factory.
    if (isPubSubDomain()) {
        if (!(getConnectionFactory() instanceof TopicConnectionFactory)) {
            throw new IllegalArgumentException(
                    "Specified a Spring JMS 1.0.2 transaction manager for topics " +
                    "but did not supply an instance of TopicConnectionFactory");
        }
    }
    else {
        if (!(getConnectionFactory() instanceof QueueConnectionFactory)) {
            throw new IllegalArgumentException(
                    "Specified a Spring JMS 1.0.2 transaction manager for queues " +
                    "but did not supply an instance of QueueConnectionFactory");
        }
    }
}

/**
 * In addition to checking whether the target ConnectionFactory is set,
 * make sure that the supplied factory is of the appropriate type for
 * the specified destination type: QueueConnectionFactory for queues,
 * TopicConnectionFactory for topics.
 */
public void afterPropertiesSet() {
    super.afterPropertiesSet();

    // Make sure that the ConnectionFactory passed is consistent.
    // Some provider implementations of the ConnectionFactory interface
    // implement both domain interfaces under the cover, so just check if
    // the selected domain is consistent with the type of connection factory.
    if (isPubSubDomain()) {
        if (!(getTargetConnectionFactory() instanceof TopicConnectionFactory)) {
            throw new IllegalArgumentException(
                    "Specified a Spring JMS 1.0.2 SingleConnectionFactory for topics " +
                    "but did not supply an instance of TopicConnectionFactory");
        }
    }
    else {
        if (!(getTargetConnectionFactory() instanceof QueueConnectionFactory)) {
            throw new IllegalArgumentException(
                    "Specified a Spring JMS 1.0.2 SingleConnectionFactory for queues " +
                    "but did not supply an instance of QueueConnectionFactory");
        }
    }
}

/**
 * In addition to checking if the connection factory is set, make sure
 * that the supplied connection factory is of the appropriate type for
 * the specified destination type: QueueConnectionFactory for queues,
 * and TopicConnectionFactory for topics.
 */
public void afterPropertiesSet() {
    super.afterPropertiesSet();

    // Make sure that the ConnectionFactory passed is consistent.
    // Some provider implementations of the ConnectionFactory interface
    // implement both domain interfaces under the cover, so just check if
    // the selected domain is consistent with the type of connection factory.
    if (isPubSubDomain()) {
        if (!(getConnectionFactory() instanceof TopicConnectionFactory)) {
            throw new IllegalArgumentException(
                    "Specified a Spring JMS 1.0.2 template for topics " +
                    "but did not supply an instance of TopicConnectionFactory");
        }
    }
    else {
        if (!(getConnectionFactory() instanceof QueueConnectionFactory)) {
            throw new IllegalArgumentException(
                    "Specified a Spring JMS 1.0.2 template for queues " +
                    "but did not supply an instance of QueueConnectionFactory");
        }
    }
}

@Test
@Deprecated
public void testTransactionCommit102WithTopic() throws JMSException {
    TopicConnectionFactory cf = mock(TopicConnectionFactory.class);
    TopicConnection con = mock(TopicConnection.class);
    final TopicSession session = mock(TopicSession.class);

    given(cf.createTopicConnection()).willReturn(con);
    given(con.createTopicSession(true, Session.AUTO_ACKNOWLEDGE)).willReturn(session);

    JmsTransactionManager tm = new JmsTransactionManager102(cf, true);
    TransactionStatus ts = tm.getTransaction(new DefaultTransactionDefinition());
    JmsTemplate jt = new JmsTemplate102(cf, true);
    jt.execute(new SessionCallback() {
        @Override
        public Object doInJms(Session sess) {
            assertTrue(sess == session);
            return null;
        }
    });
    tm.commit(ts);

    verify(session).commit();
    verify(session).close();
    verify(con).close();
}

@Test
public void testWithTopicConnectionFactoryAndJms11Usage() throws JMSException {
    TopicConnectionFactory cf = mock(TopicConnectionFactory.class);
    TopicConnection con = mock(TopicConnection.class);

    given(cf.createConnection()).willReturn(con);

    SingleConnectionFactory scf = new SingleConnectionFactory(cf);
    Connection con1 = scf.createConnection();
    con1.start();
    con1.close();  // should be ignored
    Connection con2 = scf.createConnection();
    con2.start();
    con2.close();  // should be ignored
    scf.destroy();  // should trigger actual close

    verify(con).start();
    verify(con).stop();
    verify(con).close();
    verifyNoMoreInteractions(con);
}

@Test
public void testWithTopicConnectionFactoryAndJms102Usage() throws JMSException {
    TopicConnectionFactory cf = mock(TopicConnectionFactory.class);
    TopicConnection con = mock(TopicConnection.class);

    given(cf.createTopicConnection()).willReturn(con);

    SingleConnectionFactory scf = new SingleConnectionFactory(cf);
    Connection con1 = scf.createTopicConnection();
    con1.start();
    con1.close();  // should be ignored
    Connection con2 = scf.createTopicConnection();
    con2.start();
    con2.close();  // should be ignored
    scf.destroy();  // should trigger actual close

    verify(con).start();
    verify(con).stop();
    verify(con).close();
    verifyNoMoreInteractions(con);
}

@Test
public void testConnectionFactory102WithTopic() throws JMSException {
    TopicConnectionFactory cf = mock(TopicConnectionFactory.class);
    TopicConnection con = mock(TopicConnection.class);

    given(cf.createTopicConnection()).willReturn(con);

    SingleConnectionFactory scf = new SingleConnectionFactory102(cf, true);
    TopicConnection con1 = scf.createTopicConnection();
    con1.start();
    con1.close();  // should be ignored
    TopicConnection con2 = scf.createTopicConnection();
    con2.start();
    con2.close();  // should be ignored
    scf.destroy();  // should trigger actual close

    verify(con).start();
    verify(con).stop();
    verify(con).close();
    verifyNoMoreInteractions(con);
}

public static Topic createTopic(String uri, String topicName) throws JMSException {
    TopicConnectionFactory connectionFactory = null;
    TopicConnection connection = null;
    TopicSession session = null;
    Topic topic = null;
    try {
        connectionFactory = new ActiveMQConnectionFactory(uri);
        connection = connectionFactory.createTopicConnection();
        connection.start();
        session = connection.createTopicSession(Boolean.TRUE, Session.AUTO_ACKNOWLEDGE);
        topic = session.createTopic(topicName);
        session.commit();
    } finally {
        closeQuietly(session);
        closeQuietly(connection);
    }
    return topic;
}

/**
 * Product message for assigned topic.
 * 
 * @param uri
 *            e.g.: tcp://3CNL12096:61616
 * @param queueName
 *            name of queue
 * @throws JMSException
 */
public static void publishTextMsg2Topic(String uri, String topicName, String text) throws JMSException {
    TopicConnectionFactory connectionFactory = null;
    TopicConnection connection = null;
    TopicSession session = null;
    TopicPublisher tp = null;
    try {
        connectionFactory = new ActiveMQConnectionFactory(uri);
        connection = connectionFactory.createTopicConnection();
        connection.start();
        session = connection.createTopicSession(Boolean.TRUE, Session.AUTO_ACKNOWLEDGE);
        tp = session.createPublisher(session.createTopic(topicName));
        tp.setDeliveryMode(DeliveryMode.PERSISTENT);
        tp.publish(session.createTextMessage(text));
        session.commit();
    } finally {
        closeQuietly(tp);
        closeQuietly(session);
        closeQuietly(connection);
    }
}

protected Connection createConnection() throws JMSException {
    if (StringUtils.isNotEmpty(authAlias)) {
        CredentialFactory cf = new CredentialFactory(authAlias,null,null);
        if (log.isDebugEnabled()) log.debug("using userId ["+cf.getUsername()+"] to create Connection");
        if (useJms102()) {
            if (connectionFactory instanceof QueueConnectionFactory) {
                return ((QueueConnectionFactory)connectionFactory).createQueueConnection(cf.getUsername(),cf.getPassword());
            } else {
                return ((TopicConnectionFactory)connectionFactory).createTopicConnection(cf.getUsername(),cf.getPassword());
            }
        } else {
            return connectionFactory.createConnection(cf.getUsername(),cf.getPassword());
        }
    }
    if (useJms102()) {
        if (connectionFactory instanceof QueueConnectionFactory) {
            return ((QueueConnectionFactory)connectionFactory).createQueueConnection();
        } else {
            return ((TopicConnectionFactory)connectionFactory).createTopicConnection();
        }
    } else {
        return connectionFactory.createConnection();
    }
}

/**
 * {@inheritDoc}
 */
public void sendEnqueueMessage() {
    ServiceLocator locator = ServiceLocatorFactory.getLocator();
    final TopicConnectionFactory factory = (TopicConnectionFactory) locator.lookup(DEFAULT_QUEUE_CONN_FACTORY);
    final Topic topic = (Topic) locator.lookup(FileManagementMDB.QUEUE_JNDI_NAME);
    TopicConnection connection = null;
    TopicSession session = null;
    TopicPublisher publisher = null;
    try {
        connection = factory.createTopicConnection();
        session = connection.createTopicSession(true, 0);
        publisher = session.createPublisher(topic);

        final Message message = session.createTextMessage("enqueue");
        publisher.send(message);
    } catch (JMSException e) {
        LOG.error("Couldn't submit job to JMS", e);
    } finally {
        close(publisher);
        close(session);
        close(connection);
    }
}

@Test(timeout = 60000)
public void testInstanceOf() throws  Exception {
    cf = new JmsPoolConnectionFactory();
    assertTrue(cf instanceof QueueConnectionFactory);
    assertTrue(cf instanceof TopicConnectionFactory);
    cf.stop();
}

@Test(timeout = 60000)
public void testInstanceOf() throws  Exception {
    JmsPoolConnectionFactory pcf = new JmsPoolConnectionFactory();
    assertTrue(pcf instanceof QueueConnectionFactory);
    assertTrue(pcf instanceof TopicConnectionFactory);
    pcf.stop();
}

@Test(timeout = 60000)
public void testInstanceOf() throws  Exception {
    JmsPoolConnectionFactory pcf = new JmsPoolConnectionFactory();
    assertTrue(pcf instanceof QueueConnectionFactory);
    assertTrue(pcf instanceof TopicConnectionFactory);
    pcf.stop();
}

@Test(timeout = 60000)
public void testInstanceOf() throws  Exception {
    JmsPoolConnectionFactory pcf = new JmsPoolConnectionFactory();
    assertTrue(pcf instanceof QueueConnectionFactory);
    assertTrue(pcf instanceof TopicConnectionFactory);
    pcf.stop();
}

@Test(timeout = 60000)
public void testInstanceOf() throws  Exception {
    JmsPoolXAConnectionFactory pcf = new JmsPoolXAConnectionFactory();
    assertTrue(pcf instanceof QueueConnectionFactory);
    assertTrue(pcf instanceof TopicConnectionFactory);
    pcf.stop();
}

@Parameters({ "broker-port"})
@Test
public void testSubscriberPublisher(String port) throws Exception {
    String topicName = "MyTopic1";
    int numberOfMessages = 100;

    InitialContext initialContext = ClientHelper
            .getInitialContextBuilder("admin", "admin", "localhost", port)
            .withTopic(topicName)
            .build();

    TopicConnectionFactory connectionFactory
            = (TopicConnectionFactory) initialContext.lookup(ClientHelper.CONNECTION_FACTORY);
    TopicConnection connection = connectionFactory.createTopicConnection();
    connection.start();

    // Initialize subscriber
    TopicSession subscriberSession = connection.createTopicSession(false, Session.AUTO_ACKNOWLEDGE);
    Topic subscriberDestination = (Topic) initialContext.lookup(topicName);
    TopicSubscriber subscriber = subscriberSession.createSubscriber(subscriberDestination);

    // publish 100 messages
    TopicSession producerSession = connection.createTopicSession(false, Session.AUTO_ACKNOWLEDGE);
    TopicPublisher producer = producerSession.createPublisher(subscriberDestination);

    for (int i = 0; i < numberOfMessages; i++) {
        producer.publish(producerSession.createTextMessage("Test message " + i));
    }

    producerSession.close();

    for (int i = 0; i < numberOfMessages; i++) {
        Message message = subscriber.receive(1000);
        Assert.assertNotNull(message, "Message #" + i + " was not received");
    }

    connection.close();
}

@Parameters({ "broker-port", "admin-username", "admin-password" })
@Test
public void testValidClientConnection(String port, String adminUsername, String adminPassword) throws Exception {
    String topicName = "MyTopic1";
    InitialContext initialContext = ClientHelper
            .getInitialContextBuilder(adminUsername, adminPassword, "localhost", port).withTopic(topicName).build();
    TopicConnectionFactory connectionFactory = (TopicConnectionFactory) initialContext
            .lookup(ClientHelper.CONNECTION_FACTORY);
    TopicConnection connection = connectionFactory.createTopicConnection();
    connection.start();
    connection.close();
}

@Parameters({ "broker-port", "admin-username" })
@Test(expectedExceptions = JMSException.class)
public void testInvalidClientConnection(String port, String adminUsername) throws Exception {
    String topicName = "MyTopic1";
    InitialContext initialContext = ClientHelper
            .getInitialContextBuilder(adminUsername, "invalidPassword", "localhost", port).withTopic(topicName)
            .build();
    TopicConnectionFactory connectionFactory = (TopicConnectionFactory) initialContext
            .lookup(ClientHelper.CONNECTION_FACTORY);
    connectionFactory.createTopicConnection();
}

@Before
public void setUp() throws Exception {
  cut = new DefaultJmsMessageConverter(new XStreamSerializer());
  TopicConnectionFactory connectionFactory = embeddedBroker.createConnectionFactory();
  topicConnection = connectionFactory.createTopicConnection();
  topicSession = topicConnection.createTopicSession(false, Session.AUTO_ACKNOWLEDGE);
}

public synchronized void load() throws GenericServiceException {
    try {
        InitialContext jndi = JNDIContextFactory.getInitialContext(jndiServer);
        TopicConnectionFactory factory = (TopicConnectionFactory) jndi.lookup(jndiName);

        if (factory != null) {
            con = factory.createTopicConnection(userName, password);
            con.setExceptionListener(this);
            session = con.createTopicSession(false, Session.AUTO_ACKNOWLEDGE);
            topic = (Topic) jndi.lookup(topicName);
            if (topic != null) {
                TopicSubscriber subscriber = session.createSubscriber(topic);
                subscriber.setMessageListener(this);
                con.start();
                this.setConnected(true);
                if (Debug.infoOn()) Debug.logInfo("Listening to topic [" + topicName + "] on [" + jndiServer + "]...", module);
            } else {
                throw new GenericServiceException("Topic lookup failed.");
            }
        } else {
            throw new GenericServiceException("Factory (broker) lookup failed.");
        }
    } catch (NamingException ne) {
        throw new GenericServiceException("JNDI lookup problems; listener not running.", ne);
    } catch (JMSException je) {
        throw new GenericServiceException("JMS internal error; listener not running.", je);
    } catch (GeneralException ge) {
        throw new GenericServiceException("Problems with InitialContext; listener not running.", ge);
    }
}

@Override
public TopicConnection createTopicConnection() throws JMSException {
    if (!(this.targetConnectionFactory instanceof TopicConnectionFactory)) {
        throw new javax.jms.IllegalStateException("'targetConnectionFactory' is no TopicConnectionFactory");
    }
    TopicConnection targetConnection =
            ((TopicConnectionFactory) this.targetConnectionFactory).createTopicConnection();
    return (TopicConnection) getTransactionAwareConnectionProxy(targetConnection);
}

@Override
public TopicConnection createTopicConnection(String username, String password) throws JMSException {
    if (!(this.targetConnectionFactory instanceof TopicConnectionFactory)) {
        throw new javax.jms.IllegalStateException("'targetConnectionFactory' is no TopicConnectionFactory");
    }
    TopicConnection targetConnection =
            ((TopicConnectionFactory) this.targetConnectionFactory).createTopicConnection(username, password);
    return (TopicConnection) getTransactionAwareConnectionProxy(targetConnection);
}

@Override
public TopicConnection createTopicConnection() throws JMSException {
    ConnectionFactory cf = getTargetConnectionFactory();
    if (cf instanceof TopicConnectionFactory) {
        return ((TopicConnectionFactory) cf).createTopicConnection();
    }
    else {
        Connection con = cf.createConnection();
        if (!(con instanceof TopicConnection)) {
            throw new javax.jms.IllegalStateException("'targetConnectionFactory' is not a TopicConnectionFactory");
        }
        return (TopicConnection) con;
    }
}

@Override
public TopicConnection createTopicConnection(String username, String password) throws JMSException {
    ConnectionFactory cf = getTargetConnectionFactory();
    if (cf instanceof TopicConnectionFactory) {
        return ((TopicConnectionFactory) cf).createTopicConnection(username, password);
    }
    else {
        Connection con = cf.createConnection(username, password);
        if (!(con instanceof TopicConnection)) {
            throw new javax.jms.IllegalStateException("'targetConnectionFactory' is not a TopicConnectionFactory");
        }
        return (TopicConnection) con;
    }
}

/**
 * Create a JMS Connection via this template's ConnectionFactory.
 * @return the new JMS Connection
 * @throws javax.jms.JMSException if thrown by JMS API methods
 */
protected Connection doCreateConnection() throws JMSException {
    ConnectionFactory cf = getTargetConnectionFactory();
    if (Boolean.FALSE.equals(this.pubSubMode) && cf instanceof QueueConnectionFactory) {
        return ((QueueConnectionFactory) cf).createQueueConnection();
    }
    else if (Boolean.TRUE.equals(this.pubSubMode) && cf instanceof TopicConnectionFactory) {
        return ((TopicConnectionFactory) cf).createTopicConnection();
    }
    else {
        return getTargetConnectionFactory().createConnection();
    }
}

/**
 * This implementation delegates to the {@code createTopicConnection(username, password)}
 * method of the target TopicConnectionFactory, passing in the specified user credentials.
 * If the specified username is empty, it will simply delegate to the standard
 * {@code createTopicConnection()} method of the target ConnectionFactory.
 * @param username the username to use
 * @param password the password to use
 * @return the Connection
 * @see javax.jms.TopicConnectionFactory#createTopicConnection(String, String)
 * @see javax.jms.TopicConnectionFactory#createTopicConnection()
 */
protected TopicConnection doCreateTopicConnection(String username, String password) throws JMSException {
    Assert.state(this.targetConnectionFactory != null, "'targetConnectionFactory' is required");
    if (!(this.targetConnectionFactory instanceof TopicConnectionFactory)) {
        throw new javax.jms.IllegalStateException("'targetConnectionFactory' is not a TopicConnectionFactory");
    }
    TopicConnectionFactory queueFactory = (TopicConnectionFactory) this.targetConnectionFactory;
    if (StringUtils.hasLength(username)) {
        return queueFactory.createTopicConnection(username, password);
    }
    else {
        return queueFactory.createTopicConnection();
    }
}

@Test
public void testCachingConnectionFactoryWithTopicConnectionFactoryAndJms102Usage() throws JMSException {
    TopicConnectionFactory cf = mock(TopicConnectionFactory.class);
    TopicConnection con = mock(TopicConnection.class);
    TopicSession txSession = mock(TopicSession.class);
    TopicSession nonTxSession = mock(TopicSession.class);

    given(cf.createTopicConnection()).willReturn(con);
    given(con.createTopicSession(true, Session.AUTO_ACKNOWLEDGE)).willReturn(txSession);
    given(txSession.getTransacted()).willReturn(true);
    given(con.createTopicSession(false, Session.CLIENT_ACKNOWLEDGE)).willReturn(nonTxSession);

    CachingConnectionFactory scf = new CachingConnectionFactory(cf);
    scf.setReconnectOnException(false);
    Connection con1 = scf.createTopicConnection();
    Session session1 = con1.createSession(true, Session.AUTO_ACKNOWLEDGE);
    session1.getTransacted();
    session1.close();  // should lead to rollback
    session1 = con1.createSession(false, Session.CLIENT_ACKNOWLEDGE);
    session1.close();
    con1.start();
    TopicConnection con2 = scf.createTopicConnection();
    Session session2 = con2.createTopicSession(false, Session.CLIENT_ACKNOWLEDGE);
    session2.close();
    session2 = con2.createSession(true, Session.AUTO_ACKNOWLEDGE);
    session2.getTransacted();
    session2.close();
    con2.start();
    con1.close();
    con2.close();
    scf.destroy();  // should trigger actual close

    verify(txSession).close();
    verify(nonTxSession).close();
    verify(con).start();
    verify(con).stop();
    verify(con).close();
}

public TopicConnectionFactory lookupTopicConnectionFactory( String name )
        throws JMSException, NamingException {

    if (usingJNDI || usingMQ) {
        return super.lookupTopicConnectionFactory(name);
    } else {
        MQTopicConnectionFactory tcf = new MQTopicConnectionFactory();
        configureWBIMBConnectionFactory((MQConnectionFactory) tcf);
        return tcf;
    } // end if tcf==null

}

/**
 * Create a new vendor-specific ConnectionFactory (or delegate to JNDI if that is has been selected).
 */
public TopicConnectionFactory lookupTopicConnectionFactory(String name) throws JMSException,NamingException {

    if ( usingJNDI ) {
        return super.lookupTopicConnectionFactory(name);
    } else {
        MQTopicConnectionFactory tcf = new MQTopicConnectionFactory();
        configureMQConnectionFactory(tcf);
        return tcf;
    } // end if tcf==null

}

public void createTopicConnectionFactory(String name) throws Exception {

    if ( usingJNDI ) {
        TopicConnectionFactory tcf = new MQTopicConnectionFactory();
        configureMQConnectionFactory((MQConnectionFactory)tcf);
        try {
            getInitialContext().bind( name, tcf );
        } catch ( NameAlreadyBoundException e ) {
            // swallowed
        }
    } else {
        // No op
    }

}

public synchronized void start() throws RestException {
    try {
        String jndiProviderUrl = String.format(PROVIDER_URL_FMT, jmsHost, jmsPort);
        String jndiFactory = JNDI_FACTORY;
        String jmsFactory = JMS_FACTORY;

        // Debug
        logger.debug("Creating JNDI connection to: " + jndiProviderUrl + " using factory: " + jndiFactory);

        // Initialize JNDI connection
        Properties env = new Properties();
        env.put(Context.INITIAL_CONTEXT_FACTORY, jndiFactory);
        env.put(Context.PROVIDER_URL, jndiProviderUrl);
        context = new InitialContext(env);

        // Debug
        logger.debug("Creating JMS connection using factory: " + jmsFactory + " user: " + jmsUser + " passwd: " + jmsPassword);

        // Create the JMS topic connection and start it
        TopicConnectionFactory topicConnectionFactory = (TopicConnectionFactory) context.lookup(jmsFactory);
        topicConnection = topicConnectionFactory.createTopicConnection(jmsUser, jmsPassword);
        topicConnection.start();

        // Debug
        logger.debug("Subscribing to JMS topic: " + jmsTopic);

        // Create the subscriber
        Topic topic = (Topic) context.lookup(jmsTopic);
        TopicSession topicSession = topicConnection.createTopicSession(false, TopicSession.AUTO_ACKNOWLEDGE);
        createSubscriber(topicSession, topic);

        // Debug
        logger.info("JMS connection started");
    } catch (NamingException | JMSException ex) {
        throw new RestException(ex);
    }
}