Java 类javax.jms.Connection 实例源码

private void putTopic(List<String> events) throws Exception {
  ConnectionFactory factory = new ActiveMQConnectionFactory(USERNAME,
      PASSWORD, BROKER_BIND_URL);
  Connection connection = factory.createConnection();
  connection.start();

  Session session = connection.createSession(true,
      Session.AUTO_ACKNOWLEDGE);
  Destination destination = session.createTopic(DESTINATION_NAME);
  MessageProducer producer = session.createProducer(destination);

  for (String event : events) {
    TextMessage message = session.createTextMessage();
    message.setText(event);
    producer.send(message);
  }
  session.commit();
  session.close();
  connection.close();
}

public Message createMessage(Object messageObject) throws JMSException {
    Connection connection = null;
    Message result = null;
    try {
        connection = startConnection();
        Session session = null;
        try {
            session = connection.createSession(isTransacted, Session.AUTO_ACKNOWLEDGE);
            if (messageObject == null) {
                result = session.createMessage();
            } else {
                if (messageObject instanceof String) {
                    result = session.createTextMessage((String) messageObject);
                } else {
                    result = session.createObjectMessage((Serializable) messageObject);
                }
            }
        } finally {
            if (session != null) session.close();
        }
    } finally {
        safeCloseConnection(connection);
    }
    return result;
}

private void sendObjectMsgSingleSession(List<? extends Serializable> objectsToSend) throws JMSException {
    Session session = null;
    Connection conn = null;
    try {
        conn = qFactory.createConnection();
        session = conn.createSession(false, Session.AUTO_ACKNOWLEDGE);
        MessageProducer producer = session.createProducer(queue);
        for (Serializable objectToSend : objectsToSend) {
            ObjectMessage msg = session.createObjectMessage();
            msg.setObject(objectToSend);
            producer.send(msg);
        }
    } finally {
        closeSession(session);
        closeConnection(conn);
    }
}

@Test(timeout = 60000)
public void testCreateSession() throws Exception {
    Connection connection = cf.createConnection();

    Session session1 = connection.createSession();
    Session session2 = connection.createSession();

    assertNotSame(session1, session2);
    assertEquals(session1.getAcknowledgeMode(), Session.AUTO_ACKNOWLEDGE);
    assertEquals(session2.getAcknowledgeMode(), Session.AUTO_ACKNOWLEDGE);

    JmsPoolSession wrapperSession1 = (JmsPoolSession) session1;
    JmsPoolSession wrapperSession2 = (JmsPoolSession) session2;

    assertNotSame(wrapperSession1.getSession(), wrapperSession2.getSession());
}

@Test(timeout = 60000)
public void testSetClientIDTwiceWithSameID() throws Exception {
    Connection connection = cf.createConnection();

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

    try {
        connection.setClientID("newID");
        connection.start();
        connection.close();
    } catch (IllegalStateException 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 {
    Connection connection = cf.createConnection();

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

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

@Test(timeout = 60000)
public void testSetClientIDAfterConnectionStart() throws Exception {
    Connection connection = cf.createConnection();

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

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

@Test(timeout = 60000)
public void testConnectionsAreRotated() throws Exception {
    JmsPoolConnectionFactory cf = createPooledConnectionFactory();
    cf.setMaxConnections(10);

    Connection previous = null;

    // Front load the pool.
    for (int i = 0; i < 10; ++i) {
        cf.createConnection();
    }

    for (int i = 0; i < 100; ++i) {
        Connection current = ((JmsPoolConnection) cf.createConnection()).getConnection();
        assertNotSame(previous, current);
        previous = current;
    }

    cf.stop();
}

@Test(timeout = 60000)
public void testSetClientIDTwiceWithSameID() throws Exception {
    LOG.debug("running testRepeatedSetClientIDCalls()");

    // test: call setClientID("newID") twice
    // this should be tolerated and not result in an exception
    ConnectionFactory cf = createPooledConnectionFactory();
    Connection conn = cf.createConnection();
    conn.setClientID("newID");

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

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

@Test(timeout = 60000)
public void testSetClientIDTwiceWithDifferentID() throws Exception {
    LOG.debug("running testRepeatedSetClientIDCalls()");

    ConnectionFactory cf = createPooledConnectionFactory();
    Connection conn = cf.createConnection();

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

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

@Test(timeout = 60000)
public void testSetClientIDAfterConnectionStart() throws Exception {
    LOG.debug("running testRepeatedSetClientIDCalls()");

    ConnectionFactory cf = createPooledConnectionFactory();
    Connection conn = cf.createConnection();

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

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

@Test(timeout = 60000)
public void testConnectionsAreRotated() throws Exception {
    cf.setMaxConnections(10);

    Connection previous = null;

    // Front load the pool.
    for (int i = 0; i < 10; ++i) {
        cf.createConnection();
    }

    for (int i = 0; i < 100; ++i) {
        Connection current = ((JmsPoolConnection) cf.createConnection()).getConnection();
        assertNotSame(previous, current);
        previous = current;
    }

    cf.stop();
}

@Test(timeout = 60000)
public void testSetClientIDTwiceWithSameID() throws Exception {
    // test: call setClientID("newID") twice
    // this should be tolerated and not result in an exception
    Connection conn = cf.createConnection();
    conn.setClientID("newID");

    try {
        conn.setClientID("newID");
        conn.start();
        conn.close();
    } catch (IllegalStateException 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 {
    Connection conn = cf.createConnection();

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

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

@Test(timeout = 60000)
public void testSetClientIDAfterConnectionStart() throws Exception {
    Connection conn = cf.createConnection();

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

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

public void sendMessages(ConnectionFactory connectionFactory) throws Exception {
    for (int i = 0; i < NUM_MESSAGES; i++) {
        Connection connection = connectionFactory.createConnection();
        connection.start();

        Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
        Destination destination = session.createQueue(QUEUE);
        MessageProducer producer = session.createProducer(destination);

        String msgTo = "hello";
        TextMessage message = session.createTextMessage(msgTo);
        producer.send(message);
        connection.close();
        LOG.debug("sent " + i + " messages using " + connectionFactory.getClass());
    }
}

@Test(timeout = 60000)
public void testConnectionsAreRotated() throws Exception {

    ActiveMQConnectionFactory amq = new ActiveMQConnectionFactory(
        "vm://broker1?marshal=false&broker.persistent=false&broker.useJmx=false");
    JmsPoolConnectionFactory cf = new JmsPoolConnectionFactory();
    cf.setConnectionFactory(amq);
    cf.setMaxConnections(10);

    Connection previous = null;

    // Front load the pool.
    for (int i = 0; i < 10; ++i) {
        cf.createConnection();
    }

    for (int i = 0; i < 100; ++i) {
        Connection current = ((JmsPoolConnection) cf.createConnection()).getConnection();
        assertNotSame(previous, current);
        previous = current;
    }

    cf.stop();
}

public static void send(String queueName, String text, int delayMillis) {
    EXECUTOR.submit(() -> {
        try {
            logger.info("*** artificial delay {}: {}", queueName, delayMillis);
            Thread.sleep(delayMillis);
            Connection connection = getConnection();
            Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
            Destination destination = session.createQueue(queueName);
            MessageProducer producer = session.createProducer(destination);
            producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
            TextMessage message = session.createTextMessage(text);
            producer.send(message);
            logger.info("*** sent message {}: {}", queueName, text);
            session.close();
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    });
}

private void sendWithReplyToTemp(ConnectionFactory cf, String serviceQueue) throws JMSException, InterruptedException {
    Connection connection = cf.createConnection();
    connection.start();
    Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
    TemporaryQueue tempQueue = session.createTemporaryQueue();
    TextMessage msg = session.createTextMessage("Request");
    msg.setJMSReplyTo(tempQueue);
    MessageProducer producer = session.createProducer(session.createQueue(serviceQueue));
    producer.send(msg);

    MessageConsumer consumer = session.createConsumer(tempQueue);
    Message replyMsg = consumer.receive();
    assertNotNull(replyMsg);

    LOG.debug("Reply message: {}", replyMsg);

    consumer.close();

    producer.close();
    session.close();
    connection.close();
}

public void receiveAndRespondWithMessageIdAsCorrelationId(ConnectionFactory connectionFactory, String queueName) throws JMSException {
    Connection connection = connectionFactory.createConnection();
    Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
    MessageConsumer consumer = session.createConsumer(session.createQueue(queueName));
    final javax.jms.Message inMessage = consumer.receive();

    String requestMessageId = inMessage.getJMSMessageID();
    LOG.debug("Received message " + requestMessageId);
    final TextMessage replyMessage = session.createTextMessage("Result");
    replyMessage.setJMSCorrelationID(inMessage.getJMSMessageID());
    final MessageProducer producer = session.createProducer(inMessage.getJMSReplyTo());
    LOG.debug("Sending reply to " + inMessage.getJMSReplyTo());
    producer.send(replyMessage);

    producer.close();
    consumer.close();
    session.close();
    connection.close();
}

@Test
public void testFailedConnectThenSucceeds() throws JMSException {
    Connection connection = pooledConnFact.createConnection("invalid", "credentials");

    try {
        connection.start();
        fail("Should fail to connect");
    } catch (JMSSecurityException ex) {
        LOG.info("Caught expected security error");
    }

    connection = pooledConnFact.createConnection("system", "manager");
    connection.start();

    LOG.info("Successfully create new connection.");

    connection.close();
}

@Test
public void testFailoverWithInvalidCredentialsCanConnect() throws JMSException {

    ActiveMQConnectionFactory cf = new ActiveMQConnectionFactory(
        "failover:(" + connectionURI + ")");

    pooledConnFact = new JmsPoolConnectionFactory();
    pooledConnFact.setConnectionFactory(cf);
    pooledConnFact.setMaxConnections(1);

    Connection connection = pooledConnFact.createConnection("invalid", "credentials");

    try {
        connection.start();
        fail("Should fail to connect");
    } catch (JMSSecurityException ex) {
        LOG.info("Caught expected security error");
    }

    connection = pooledConnFact.createConnection("system", "manager");
    connection.start();

    LOG.info("Successfully create new connection.");

    connection.close();
}

@Test(timeout = 60000)
public void testSetClientIDAfterConnectionStart() throws Exception {
    LOG.debug("running testRepeatedSetClientIDCalls()");

    JmsPoolConnectionFactory cf = createPooledConnectionFactory();
    Connection conn = cf.createConnection();

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

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

private void testReception(
        final TestClass i,
        final Destination destination, final ThrowingConsumer<Destination> destinationCheck
) throws Exception {
    final Connection connection = i.connectionFactory.createConnection();
    assertNotNull(connection);
    final Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
    assertNotNull(session);
    final MessageProducer producer = session.createProducer(destination);
    assertNotNull(producer);
    final TextMessage message = session.createTextMessage();
    assertNotNull(message);
    message.setText("I am IronMan");
    producer.send(message);

    final List<ReceivedJmsMessage> messages = i.testQueue.drainReceivedMessages();
    assertEquals(1, messages.size());
    final ReceivedJmsMessage receivedMessage = messages.get(0);
    destinationCheck.accept(receivedMessage.getDestination());
    assertTrue(receivedMessage.getJmsMessage() instanceof TextMessage);
    final TextMessage receivedTextMessage = (TextMessage) receivedMessage.getJmsMessage();
    assertEquals("I am IronMan", receivedTextMessage.getText());
}

/**
 * This test simply validates that {@link ConnectionFactory} can be setup by
 * pointing to the location of the client libraries at runtime. It uses
 * ActiveMQ which is not present at the POM but instead pulled from Maven
 * repo using {@link TestUtils#setupActiveMqLibForTesting(boolean)}, which
 * implies that for this test to run the computer must be connected to the
 * Internet. If computer is not connected to the Internet, this test will
 * quietly fail logging a message.
 */
@Test
public void validateFactoryCreationWithActiveMQLibraries() throws Exception {
    try {
        String libPath = TestUtils.setupActiveMqLibForTesting(true);

        TestRunner runner = TestRunners.newTestRunner(mock(Processor.class));
        JMSConnectionFactoryProvider cfProvider = new JMSConnectionFactoryProvider();
        runner.addControllerService("cfProvider", cfProvider);
        runner.setProperty(cfProvider, JMSConnectionFactoryProvider.BROKER_URI,
                "vm://localhost?broker.persistent=false");
        runner.setProperty(cfProvider, JMSConnectionFactoryProvider.CLIENT_LIB_DIR_PATH, libPath);
        runner.setProperty(cfProvider, JMSConnectionFactoryProvider.CONNECTION_FACTORY_IMPL,
                "org.apache.activemq.ActiveMQConnectionFactory");
        runner.enableControllerService(cfProvider);
        runner.assertValid(cfProvider);

        Connection connection = cfProvider.getConnectionFactory().createConnection();
        connection.start();

        Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
        Destination queue = session.createQueue("myqueue");
        MessageProducer producer = session.createProducer(queue);
        MessageConsumer consumer = session.createConsumer(queue);

        TextMessage message = session.createTextMessage("Hello");
        producer.send(message);
        assertEquals("Hello", ((TextMessage) consumer.receive()).getText());
        connection.stop();
        connection.close();
    } catch (Exception e) {
        logger.error("'validateFactoryCreationWithActiveMQLibraries' failed due to ", e);
    }
}

private Connection createConnection() throws JMSException {
    if (destinationName == null) {
        throw new IllegalArgumentException("Destination is null, can't send message to nowhere");
    }
    Connection connection;
    //if we don't have liveConnection, try to create fresh from factory
    if (keepAlive) {
        if (liveConnection == null) liveConnection = factory.createConnection();
        connection = liveConnection;
    } else {
        connection = factory.createConnection();
    }
    return connection;
}

public static Connection getConnection() {
    try {
        ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory("vm://localhost");
        Connection connection = connectionFactory.createConnection();
        connection.start();
        return connection;
    } catch (Exception e) {
        throw new RuntimeException(e);
    }
}

public Connection createConnection(String userName, String password)
        throws JMSException {
    ProxyConnection proxyConnection = new ProxyConnection(this);
    this.connections.add(proxyConnection);

    return proxyConnection;
}

/**
 * @param brokerUrl
 *            The JMS broker url
 * @param userId
 *            The user id
 * @param password
 *            The password
 * @return {@link Connection}
 * @throws JMSException
 * @throws NamingException
 */
private static Connection createConnection(String brokerUrl, String userId, String password)
        throws JMSException, NamingException {

    Context context = new InitialContext();

    ConnectionFactory factory = (ConnectionFactory) context.lookup(brokerUrl);
    // Get a new instance of ConnectionFactory
    // create a connection - providing the user id and password
    Connection connection = factory.createConnection(userId, password);
    return connection;
}

@Test(timeout = 60000)
public void testEvictionOfExpired() throws Exception {
    cf.setExpiryTimeout(10);
    Connection connection = cf.createConnection();
    Connection amq1 = ((JmsPoolConnection) connection).getConnection();

    // let it expire while in use
    TimeUnit.MILLISECONDS.sleep(20);
    connection.close();

    Connection connection2 = cf.createConnection();
    Connection amq2 = ((JmsPoolConnection) connection2).getConnection();
    assertTrue("not equal", !amq1.equals(amq2));
}

public void doTestCreateSessionWithGivenAckMode(int ackMode) throws Exception {
    Connection connection = cf.createConnection();

    Session session1 = connection.createSession(ackMode);
    Session session2 = connection.createSession(ackMode);

    assertNotSame(session1, session2);
    assertEquals(session1.getAcknowledgeMode(), ackMode);
    assertEquals(session2.getAcknowledgeMode(), ackMode);

    JmsPoolSession wrapperSession1 = (JmsPoolSession) session1;
    JmsPoolSession wrapperSession2 = (JmsPoolSession) session2;

    assertNotSame(wrapperSession1.getSession(), wrapperSession2.getSession());
}

public void doTestCreateSessionWithGivenAckModeAndTXFlag(boolean transacted, int ackMode) throws Exception {
    Connection connection = cf.createConnection();

    if (!transacted && ackMode == Session.SESSION_TRANSACTED) {
        try {
            connection.createSession(transacted, ackMode);
            fail("Should not allow non-transacted session with SESSION_TRANSACTED");
        } catch (JMSException jmsex) {}
    } else {
        Session session1 = connection.createSession(transacted, ackMode);
        Session session2 = connection.createSession(transacted, ackMode);

        assertNotSame(session1, session2);

        if (transacted) {
            assertEquals(session1.getAcknowledgeMode(), Session.SESSION_TRANSACTED);
            assertEquals(session2.getAcknowledgeMode(), Session.SESSION_TRANSACTED);
        } else {
            assertEquals(session1.getAcknowledgeMode(), ackMode);
            assertEquals(session2.getAcknowledgeMode(), ackMode);
        }

        JmsPoolSession wrapperSession1 = (JmsPoolSession) session1;
        JmsPoolSession wrapperSession2 = (JmsPoolSession) session2;

        assertNotSame(wrapperSession1.getSession(), wrapperSession2.getSession());
    }
}

@Test(timeout = 60000)
public void testCreateConnection() throws Exception {
    Connection connection = cf.createConnection();

    assertNotNull(connection);
    assertEquals(1, cf.getNumConnections());

    connection.close();

    assertEquals(1, cf.getNumConnections());
}

@Test(timeout = 60000)
public void testCreateConnectionWithCredentials() throws Exception {
    Connection connection = cf.createConnection("user", "pass");

    assertNotNull(connection);
    assertEquals(1, cf.getNumConnections());

    connection.close();

    assertEquals(1, cf.getNumConnections());
}

private void doTestConcurrentCreateGetsUniqueConnection(boolean createOnStart) throws Exception {
    final int numConnections = 2;

    final MockJMSConnectionFactory mock = new MockJMSConnectionFactory();
    cf = new JmsPoolConnectionFactory();
    cf.setConnectionFactory(mock);
    cf.setMaxConnections(numConnections);
    cf.setCreateConnectionOnStartup(createOnStart);
    cf.start();

    final ConcurrentMap<UUID, Connection> connections = new ConcurrentHashMap<>();
    final ExecutorService executor = Executors.newFixedThreadPool(numConnections);

    for (int i = 0; i < numConnections; ++i) {
        executor.execute(new Runnable() {

            @Override
            public void run() {
                try {
                    JmsPoolConnection pooled = (JmsPoolConnection) cf.createConnection();
                    MockJMSConnection wrapped = (MockJMSConnection) pooled.getConnection();
                    connections.put(wrapped.getConnectionId(), pooled);
                } catch (JMSException e) {
                }
            }
        });
    }

    executor.shutdown();
    assertTrue(executor.awaitTermination(30, TimeUnit.SECONDS));

    assertEquals("Should have all unique connections", numConnections, connections.size());

    connections.clear();
    cf.stop();
}

public Message receiveMessage(long timeout, String selector) throws JMSException {
    Connection connection = null;
    Message result = null;
    try {
        connection = startConnection(); //try to be smarter here and start stable connection
        Session session = null;
        try {
            session = connection.createSession(isTransacted, Session.AUTO_ACKNOWLEDGE);
            Destination dest;
            if (isQueue) {
                dest = session.createQueue(destinationName);
            } else {
                dest = session.createTopic(destinationName);
            }
            MessageConsumer consumer;
            if (selector != null) {
                consumer = session.createConsumer(dest, selector);
            } else {
                consumer = session.createConsumer(dest);
            }
            try {
                result = consumer.receive(timeout);
            } finally {
                if (consumer != null) consumer.close();
            }
        } finally {
            if (session != null) session.close();
        }
    } finally {
        safeCloseConnection(connection);
    }
    return result;
}

@Test(timeout = 60000)
public void testTemporaryQueueWithMultipleConnectionUsers() throws Exception {
    Connection pooledConnection = null;
    Connection pooledConnection2 = null;
    Session session = null;
    Session session2 = null;
    Queue tempQueue = null;
    Queue normalQueue = null;

    pooledConnection = pooledFactory.createConnection();
    session = pooledConnection.createSession(false, Session.AUTO_ACKNOWLEDGE);
    tempQueue = session.createTemporaryQueue();
    LOG.info("Created queue named: " + tempQueue.getQueueName());

    assertEquals(1, countBrokerTemporaryQueues());

    pooledConnection2 = pooledFactory.createConnection();
    session2 = pooledConnection2.createSession(false, Session.AUTO_ACKNOWLEDGE);
    normalQueue = session2.createQueue("queue:FOO.TEST");
    LOG.info("Created queue named: " + normalQueue.getQueueName());

    // didn't create a temp queue on pooledConnection2 so we should still have a temp queue
    pooledConnection2.close();
    assertEquals(1, countBrokerTemporaryQueues());

    // after closing pooledConnection, where we created the temp queue, there should
    // be no temp queues left
    pooledConnection.close();
    assertEquals(0, countBrokerTemporaryQueues());
}

@Test(timeout = 60000)
public void testTemporaryQueueLeakAfterConnectionClose() throws Exception {
    Connection pooledConnection = null;
    Session session = null;
    Queue tempQueue = null;
    for (int i = 0; i < 2; i++) {
        pooledConnection = pooledFactory.createConnection();
        session = pooledConnection.createSession(false, Session.AUTO_ACKNOWLEDGE);
        tempQueue = session.createTemporaryQueue();
        LOG.info("Created queue named: " + tempQueue.getQueueName());
        pooledConnection.close();
    }

    assertEquals(0, countBrokerTemporaryQueues());
}

@Test(timeout = 60000)
public void testTemporaryTopicLeakAfterConnectionClose() throws Exception {
    Connection pooledConnection = null;
    Session session = null;
    Topic tempTopic = null;
    for (int i = 0; i < 2; i++) {
        pooledConnection = pooledFactory.createConnection();
        session = pooledConnection.createSession(false, Session.AUTO_ACKNOWLEDGE);
        tempTopic = session.createTemporaryTopic();
        LOG.info("Created topic named: " + tempTopic.getTopicName());
        pooledConnection.close();
    }

    assertEquals(0, countBrokerTemporaryTopics());
}

public Connection getConnection() {
    if (poolConnection) {
        Connection conn = null;
        synchronized (connectionQueue) {
            conn = connectionQueue.poll();
            connectionQueue.add(conn);
        }
        return conn;
    } else {
        return connection;
    }
}