Java 类javax.jms.QueueConnection 实例源码

@Test(timeout = 60000)
public void testSenderAndPublisherDest() throws Exception {
    JmsPoolXAConnectionFactory pcf = new JmsPoolXAConnectionFactory();
    pcf.setConnectionFactory(new ActiveMQXAConnectionFactory(
        "vm://test?broker.persistent=false&broker.useJmx=false"));

    QueueConnection connection = pcf.createQueueConnection();
    QueueSession session = connection.createQueueSession(false, Session.AUTO_ACKNOWLEDGE);
    QueueSender sender = session.createSender(session.createQueue("AA"));
    assertNotNull(sender.getQueue().getQueueName());

    connection.close();

    TopicConnection topicConnection = pcf.createTopicConnection();
    TopicSession topicSession = topicConnection.createTopicSession(false, Session.AUTO_ACKNOWLEDGE);
    TopicPublisher topicPublisher = topicSession.createPublisher(topicSession.createTopic("AA"));
    assertNotNull(topicPublisher.getTopic().getTopicName());

    topicConnection.close();
    pcf.stop();
}

public static ManagedConnection create(
                                        final Connection connection ) {

    if ( (connection instanceof XAQueueConnection) && (connection instanceof XATopicConnection)) {
        return new ManagedXAQueueTopicConnection(connection);
    } else if (connection instanceof XAQueueConnection) {
        return new ManagedXAQueueConnection((XAQueueConnection) connection);
    } else if (connection instanceof XATopicConnection) {
        return new ManagedXATopicConnection((XATopicConnection) connection);
    } else if ( (connection instanceof QueueConnection) && (connection instanceof TopicConnection)) {
        return new ManagedQueueTopicConnection(connection);
    } else if (connection instanceof QueueConnection) {
        return new ManagedQueueConnection((QueueConnection) connection);
    } else if (connection instanceof TopicConnection) {
        return new ManagedTopicConnection((TopicConnection) connection);
    } else {
        return new ManagedConnection(connection);
    }
}

/** Send a JSON message to our notification queue.
 */
public void invokeJMS(JsonObject json) throws JMSException, NamingException {
    if (!initialized) initialize(); //gets our JMS managed resources (Q and QCF)

    QueueConnection connection = queueCF.createQueueConnection();
    QueueSession session = connection.createQueueSession(false, Session.AUTO_ACKNOWLEDGE);

    TextMessage message = session.createTextMessage(json.toString());
    System.out.println("Sending "+json.toString()+" to "+queue.getQueueName());
    QueueSender sender = session.createSender(queue);
    sender.send(message);

    sender.close();
    session.close();
    connection.close();

    System.out.println("Message sent successfully!");
}

/**
 * Create a default Session for this ConnectionFactory,
 * adapting to JMS 1.0.2 style queue/topic mode if necessary.
 * @param con the JMS Connection to operate on
 * @param mode the Session acknowledgement mode
 * ({@code Session.TRANSACTED} or one of the common modes)
 * @return the newly created Session
 * @throws JMSException if thrown by the JMS API
 */
protected Session createSession(Connection con, Integer mode) throws JMSException {
    // Determine JMS API arguments...
    boolean transacted = (mode == Session.SESSION_TRANSACTED);
    int ackMode = (transacted ? Session.AUTO_ACKNOWLEDGE : mode);
    // Now actually call the appropriate JMS factory method...
    if (Boolean.FALSE.equals(this.pubSubMode) && con instanceof QueueConnection) {
        return ((QueueConnection) con).createQueueSession(transacted, ackMode);
    }
    else if (Boolean.TRUE.equals(this.pubSubMode) && con instanceof TopicConnection) {
        return ((TopicConnection) con).createTopicSession(transacted, ackMode);
    }
    else {
        return con.createSession(transacted, ackMode);
    }
}

@Test
public void testWithQueueConnection() throws JMSException {
    Connection con = mock(QueueConnection.class);

    SingleConnectionFactory scf = new SingleConnectionFactory(con);
    QueueConnection con1 = scf.createQueueConnection();
    con1.start();
    con1.stop();
    con1.close();
    QueueConnection con2 = scf.createQueueConnection();
    con2.start();
    con2.stop();
    con2.close();
    scf.destroy();  // should trigger actual close

    verify(con, times(2)).start();
    verify(con, times(2)).stop();
    verify(con).close();
    verifyNoMoreInteractions(con);
}

@Test
public void testWithQueueConnectionFactoryAndJms11Usage() throws JMSException {
    QueueConnectionFactory cf = mock(QueueConnectionFactory.class);
    QueueConnection con = mock(QueueConnection.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 testWithQueueConnectionFactoryAndJms102Usage() throws JMSException {
    QueueConnectionFactory cf = mock(QueueConnectionFactory.class);
    QueueConnection con = mock(QueueConnection.class);

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

    SingleConnectionFactory scf = new SingleConnectionFactory(cf);
    Connection con1 = scf.createQueueConnection();
    Connection con2 = scf.createQueueConnection();
    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 QueueConnection getQueueConnection(QueueConnectionFactory qcf)
        throws JMSException {

    final QueueConnection qc;
    final String username = Config.parms.getString("us");
    if (username != null && username.length() != 0) {
        Log.logger.log(Level.INFO, "getQueueConnection(): authenticating as \"" + username + "\"");
        final String password = Config.parms.getString("pw");
        qc = qcf.createQueueConnection(username, password);
    } else {
        qc = qcf.createQueueConnection();
    }

    return qc;

}

private void acknowledge(Message jmsMessage, String msgid) throws JMSException, ServiceLocatorException {
        QueueConnection connection = null;
        QueueSession session = null;
        QueueSender sender = null;
        try {
            Queue respQueue = (Queue) jmsMessage.getJMSReplyTo();
            QueueConnectionFactory qcf = JMSServices.getInstance().getQueueConnectionFactory(null);
            connection = qcf.createQueueConnection();
            session = connection.createQueueSession(false, QueueSession.DUPS_OK_ACKNOWLEDGE);
            sender = session.createSender(respQueue);
            Message respMsg = session.createTextMessage(msgid);
//        respMsg.setJMSCorrelationID(correlationId); not used
            sender.send(respMsg);
        } finally {
            if (sender != null) sender.close();
            if (session != null) session.close();
            if (connection != null) connection.close();
        }
    }

private void acknowledge(Message jmsMessage, String msgid) throws JMSException, ServiceLocatorException {
        QueueConnection connection = null;
        QueueSession session = null;
        QueueSender sender = null;
        try {
            Queue respQueue = (Queue) jmsMessage.getJMSReplyTo();
            QueueConnectionFactory qcf = JMSServices.getInstance().getQueueConnectionFactory(null);
            connection = qcf.createQueueConnection();
            session = connection.createQueueSession(false, QueueSession.DUPS_OK_ACKNOWLEDGE);
            sender = session.createSender(respQueue);
            Message respMsg = session.createTextMessage(msgid);
//        respMsg.setJMSCorrelationID(correlationId); not used
            sender.send(respMsg);
        } finally {
            if (sender != null) sender.close();
            if (session != null) session.close();
            if (connection != null) connection.close();
        }
    }

/**
 * Send a message to testInboundQueue queue
 *
 * @throws Exception
 */
private void sendMessage() throws Exception {
    InitialContext initialContext = JmsClientHelper.getActiveMqInitialContext();
    QueueConnectionFactory connectionFactory
            = (QueueConnectionFactory) initialContext.lookup(JmsClientHelper.QUEUE_CONNECTION_FACTORY);
    QueueConnection queueConnection = connectionFactory.createQueueConnection();
    QueueSession queueSession = queueConnection.createQueueSession(false, QueueSession.AUTO_ACKNOWLEDGE);
    QueueSender sender = queueSession.createSender(queueSession.createQueue(QUEUE_NAME));

    String message = "<?xml version='1.0' encoding='UTF-8'?>" +
            "    <ser:getQuote xmlns:ser=\"http://services.samples\" xmlns:xsd=\"http://services.samples/xsd\"> " +
            "      <ser:request>" +
            "        <xsd:symbol>IBM</xsd:symbol>" +
            "      </ser:request>" +
            "    </ser:getQuote>";
    try {
        TextMessage jmsMessage = queueSession.createTextMessage(message);
        jmsMessage.setJMSType("incorrecttype");
        sender.send(jmsMessage);
    } finally {
        queueConnection.close();
    }
}

public ReorderRequestMessageListener() {
    try {
        Properties properties = new Properties();
        properties.put(Context.INITIAL_CONTEXT_FACTORY, QPID_ICF);
        properties.put(CF_NAME_PREFIX + CF_NAME, getTCPConnectionURL(USERNAME, PASSWORD));
        properties.put(QUEUE_NAME_PREFIX + REORDER_REQUEST_QUEUE, REORDER_REQUEST_QUEUE);
        InitialContext ctx = new InitialContext(properties);
        // Lookup connection factory
        QueueConnectionFactory connFactory = (QueueConnectionFactory) ctx.lookup(CF_NAME);
        QueueConnection queueConnection = connFactory.createQueueConnection();
        queueConnection.start();
        QueueSession queueSession = queueConnection.createQueueSession(false, QueueSession.AUTO_ACKNOWLEDGE);
        //Receive message
        Queue queue = (Queue) ctx.lookup(REORDER_REQUEST_QUEUE);
        MessageConsumer consumer = queueSession.createConsumer(queue);
        consumer.setMessageListener(this);
    } catch (NamingException | JMSException e) {
        e.printStackTrace();
    }
}

protected void initializeInboundDestinationBridgesOutboundSide(QueueConnection connection) throws JMSException {
    if (inboundQueueBridges != null) {
        QueueSession outboundSession = connection.createQueueSession(false, Session.AUTO_ACKNOWLEDGE);

        for (InboundQueueBridge bridge : inboundQueueBridges) {
            String queueName = bridge.getInboundQueueName();
            Queue foreignQueue = createForeignQueue(outboundSession, queueName);
            bridge.setConsumer(null);
            bridge.setConsumerQueue(foreignQueue);
            bridge.setConsumerConnection(connection);
            bridge.setJmsConnector(this);
            addInboundBridge(bridge);
        }
        outboundSession.close();
    }
}

protected void initializeInboundDestinationBridgesLocalSide(QueueConnection connection) throws JMSException {
    if (inboundQueueBridges != null) {
        QueueSession localSession = connection.createQueueSession(false,Session.AUTO_ACKNOWLEDGE);

        for (InboundQueueBridge bridge : inboundQueueBridges) {
            String localQueueName = bridge.getLocalQueueName();
            Queue activemqQueue = createActiveMQQueue(localSession, localQueueName);
            bridge.setProducerQueue(activemqQueue);
            bridge.setProducerConnection(connection);
            if (bridge.getJmsMessageConvertor() == null) {
                bridge.setJmsMessageConvertor(getInboundMessageConvertor());
            }
            bridge.setJmsConnector(this);
            addInboundBridge(bridge);
        }
        localSession.close();
    }
}

protected void initializeOutboundDestinationBridgesOutboundSide(QueueConnection connection) throws JMSException {
    if (outboundQueueBridges != null) {
        QueueSession outboundSession = connection.createQueueSession(false, Session.AUTO_ACKNOWLEDGE);

        for (OutboundQueueBridge bridge : outboundQueueBridges) {
            String queueName = bridge.getOutboundQueueName();
            Queue foreignQueue = createForeignQueue(outboundSession, queueName);
            bridge.setProducerQueue(foreignQueue);
            bridge.setProducerConnection(connection);
            if (bridge.getJmsMessageConvertor() == null) {
                bridge.setJmsMessageConvertor(getOutboundMessageConvertor());
            }
            bridge.setJmsConnector(this);
            addOutboundBridge(bridge);
        }
        outboundSession.close();
    }
}

protected void initializeOutboundDestinationBridgesLocalSide(QueueConnection connection) throws JMSException {
    if (outboundQueueBridges != null) {
        QueueSession localSession =
                connection.createQueueSession(false, Session.AUTO_ACKNOWLEDGE);

        for (OutboundQueueBridge bridge : outboundQueueBridges) {
            String localQueueName = bridge.getLocalQueueName();
            Queue activemqQueue = createActiveMQQueue(localSession, localQueueName);
            bridge.setConsumer(null);
            bridge.setConsumerQueue(activemqQueue);
            bridge.setConsumerConnection(connection);
            bridge.setJmsConnector(this);
            addOutboundBridge(bridge);
        }
        localSession.close();
    }
}

/**
 * Create JMS {@link Session} instance on top of the provided {@link Connection} instance.
 *
 * @param connection JMS Connection.
 * @return Session instance.
 * @throws JMSConnectorException Error when creating the JMS Session.
 */
public Session createSession(Connection connection) throws JMSConnectorException {
    try {
        if (logger.isDebugEnabled()) {
            logger.debug("Creating a new JMS Session on: " + this.connectionFactoryString);
        }
        if (JMSConstants.JMS_SPEC_VERSION_1_1.equals(jmsSpec) || JMSConstants.JMS_SPEC_VERSION_2_0
                .equals(jmsSpec)) {
            return connection.createSession(transactedSession, sessionAckMode);
        } else if (JMSConstants.JMSDestinationType.QUEUE.equals(this.destinationType)) {
            return ((QueueConnection) (connection)).createQueueSession(transactedSession, sessionAckMode);
        } else {
            return ((TopicConnection) (connection)).createTopicSession(transactedSession, sessionAckMode);

        }
    } catch (JMSException e) {
        throw new JMSConnectorException(
                "JMS Exception while obtaining session for factory " + connectionFactoryString, e);
    }
}

/**
 * Close a JMS {@link Connection}.
 * @param connection Connection that need to be closed.
 * @throws JMSException if an error occurs while closing the connection.
 */
public void closeConnection(Connection connection) throws JMSException {
    if (connection != null) {
        if (logger.isDebugEnabled()) {
            logger.debug("Closing a JMS Connection of: " + this.connectionFactoryString);
        }
        if ((JMSConstants.JMS_SPEC_VERSION_1_1.equals(jmsSpec)) || (JMSConstants.JMS_SPEC_VERSION_2_0
                .equals(jmsSpec))) {
            connection.close();
        } else {
            if (JMSConstants.JMSDestinationType.QUEUE.equals(this.destinationType)) {
                ((QueueConnection) connection).close();
            } else {
                ((TopicConnection) connection).close();
            }
        }
    }
}

/**
 * To publish the messages to a queue.
 *
 * @throws JMSException         JMS Exception.
 * @throws InterruptedException Interrupted exception while waiting in between messages.
 */
public void publishMessagesToQueue(String queueName) throws JMSException, InterruptedException {
    QueueConnection queueConn = (QueueConnection) connectionFactory.createConnection();
    queueConn.start();
    QueueSession queueSession = queueConn.createQueueSession(false, Session.AUTO_ACKNOWLEDGE);
    Destination destination = queueSession.createQueue(queueName);
    MessageProducer queueSender = queueSession.createProducer(destination);
    queueSender.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
    for (int index = 0; index < 10; index++) {
        String queueText = "Queue Message : " + (index + 1);
        TextMessage queueMessage = queueSession.createTextMessage(queueText);
        queueSender.send(queueMessage);
        Thread.sleep(1000);
        logger.info("Publishing " + queueText + " to queue " + queueName);
    }
    queueConn.close();
    queueSession.close();
    queueSender.close();
}

/**
 * To receive a message from a queue.
 *
 * @throws JMSException         JMS Exception.
 * @throws InterruptedException Interrupted exception while waiting in between messages.
 */
public void receiveMessagesFromQueue() throws JMSException, InterruptedException {
    QueueConnection queueConn = (QueueConnection) connectionFactory.createConnection();
    QueueSession queueSession = queueConn.createQueueSession(false, Session.AUTO_ACKNOWLEDGE);
    Destination destination = queueSession.createQueue(JMSTestConstants.QUEUE_NAME_1);
    MessageConsumer queueReceiver = queueSession.createConsumer(destination);
    MessageListener listener = message -> {
        try {
            if (message instanceof TextMessage) {
                TextMessage textMessage = (TextMessage) message;
                logger.info("Message text received : " + (textMessage.getText()));
            }
        } catch (JMSException e) {
            logger.info("JMS exception occurred.");
        }
    };
    queueReceiver.setMessageListener(listener);
    queueConn.start();
}

@Test
public void testMultipleSessionsThrowsException() throws Exception {
   resourceAdapter = newResourceAdapter();
   MyBootstrapContext ctx = new MyBootstrapContext();
   resourceAdapter.start(ctx);
   ActiveMQRAManagedConnectionFactory mcf = new ActiveMQRAManagedConnectionFactory();
   mcf.setResourceAdapter(resourceAdapter);
   ActiveMQRAConnectionFactory qraConnectionFactory = new ActiveMQRAConnectionFactoryImpl(mcf, qraConnectionManager);
   QueueConnection queueConnection = qraConnectionFactory.createQueueConnection();
   Session s = queueConnection.createSession(false, Session.AUTO_ACKNOWLEDGE);
   try {
      Session s2 = queueConnection.createSession(false, Session.AUTO_ACKNOWLEDGE);
      fail("should throw javax,jms.IllegalStateException: Only allowed one session per connection. See the J2EE spec, e.g. J2EE1.4 Section 6.6");
   } catch (JMSException e) {
   }
}

@Test
public void testConnectionCredentialsFail() throws Exception {
   resourceAdapter = newResourceAdapter();
   MyBootstrapContext ctx = new MyBootstrapContext();
   resourceAdapter.start(ctx);
   ActiveMQRAManagedConnectionFactory mcf = new ActiveMQRAManagedConnectionFactory();
   mcf.setResourceAdapter(resourceAdapter);
   ActiveMQRAConnectionFactory qraConnectionFactory = new ActiveMQRAConnectionFactoryImpl(mcf, qraConnectionManager);
   QueueConnection queueConnection = qraConnectionFactory.createQueueConnection();
   QueueSession session = queueConnection.createQueueSession(false, Session.AUTO_ACKNOWLEDGE);

   ManagedConnection mc = ((ActiveMQRASession) session).getManagedConnection();
   queueConnection.close();
   mc.destroy();

   try {
      queueConnection = qraConnectionFactory.createQueueConnection("testuser", "testwrongpassword");
      queueConnection.createQueueSession(false, Session.AUTO_ACKNOWLEDGE).close();
      fail("should throw esxception");
   } catch (JMSException e) {
      //pass
   }
}

@Test
public void testConnectionCredentialsFailRecovery() throws Exception {
   resourceAdapter = newResourceAdapter();
   MyBootstrapContext ctx = new MyBootstrapContext();
   resourceAdapter.start(ctx);
   ActiveMQRAManagedConnectionFactory mcf = new ActiveMQRAManagedConnectionFactory();
   mcf.setResourceAdapter(resourceAdapter);
   ActiveMQRAConnectionFactory qraConnectionFactory = new ActiveMQRAConnectionFactoryImpl(mcf, qraConnectionManager);
   try {
      QueueConnection queueConnection = qraConnectionFactory.createQueueConnection("testuser", "testwrongpassword");
      queueConnection.createQueueSession(false, Session.AUTO_ACKNOWLEDGE).close();
      fail("should throw esxception");
   } catch (JMSException e) {
      //make sure the recovery is null
      assertNull(mcf.getResourceRecovery());
   }
}

/**
 * com.sun.ts.tests.jms.ee.all.queueconn.QueueConnTest line 171
 */
@Test
public void testCreateReceiverWithMessageSelector() throws Exception {
   QueueConnection qc = null;

   try {
      qc = createQueueConnection();
      QueueSession qs = qc.createQueueSession(false, Session.AUTO_ACKNOWLEDGE);

      QueueReceiver qreceiver = qs.createReceiver(queue1, "targetMessage = TRUE");

      qc.start();

      TextMessage m = qs.createTextMessage();
      m.setText("one");
      m.setBooleanProperty("targetMessage", false);

      QueueSender qsender = qs.createSender(queue1);

      qsender.send(m);

      m.setText("two");
      m.setBooleanProperty("targetMessage", true);

      qsender.send(m);

      TextMessage rm = (TextMessage) qreceiver.receive(1000);

      ProxyAssertSupport.assertEquals("two", rm.getText());
   } finally {
      if (qc != null) {
         qc.close();
      }
      Thread.sleep(2000);
      removeAllMessages(queue1.getQueueName(), true);
      checkEmpty(queue1);
   }
}

private void assertConnectionType(Connection conn, String type) {
   if ("generic".equals(type) || "queue".equals(type) || "topic".equals(type)) {
      //generic
      Assert.assertFalse(conn instanceof XAConnection);
      Assert.assertTrue(conn instanceof QueueConnection);
      Assert.assertFalse(conn instanceof XAQueueConnection);
      Assert.assertTrue(conn instanceof TopicConnection);
      Assert.assertFalse(conn instanceof XATopicConnection);
   } else if ("xa".equals(type) || "xa-queue".equals(type) || "xa-topic".equals(type)) {
      Assert.assertTrue(conn instanceof XAConnection);
      Assert.assertTrue(conn instanceof QueueConnection);
      Assert.assertTrue(conn instanceof XAQueueConnection);
      Assert.assertTrue(conn instanceof TopicConnection);
      Assert.assertTrue(conn instanceof XATopicConnection);
   } else {
      Assert.fail("Unknown connection type: " + type);
   }
}

/**
 * Create a queue connection
 *
 * @param userName The user name
 * @param password The password
 * @return The connection
 * @throws JMSException Thrown if the operation fails
 */
@Override
public QueueConnection createQueueConnection(final String userName, final String password) throws JMSException {
   if (ActiveMQRAConnectionFactoryImpl.trace) {
      ActiveMQRALogger.LOGGER.trace("createQueueConnection(" + userName + ", ****)");
   }

   ActiveMQRASessionFactoryImpl s = new ActiveMQRASessionFactoryImpl(mcf, cm, getResourceAdapter().getTM(), ActiveMQRAConnectionFactory.QUEUE_CONNECTION);
   s.setUserName(userName);
   s.setPassword(password);

   validateUser(s);

   if (ActiveMQRAConnectionFactoryImpl.trace) {
      ActiveMQRALogger.LOGGER.trace("Created queue connection: " + s);
   }

   return s;
}

@Before
public void setUp() throws Exception {
    System.setProperty(Context.INITIAL_CONTEXT_FACTORY, TestContextFactory.class.getName());
    QueueConnectionFactory queueConnectionFactory = Mockito.mock(QueueConnectionFactory.class);
    Queue queue = Mockito.mock(Queue.class);
    Context context = Mockito.mock(Context.class);
    TestContextFactory.context = context;
    when(context.lookup(eq("jms/queueConnectionFactory"))).thenReturn(queueConnectionFactory);
    when(context.lookup(eq("jms/mailQueue"))).thenReturn(queue);
    queueSender = Mockito.mock(QueueSender.class);
    QueueConnection queueConnection = Mockito.mock(QueueConnection.class);
    when(queueConnectionFactory.createQueueConnection()).thenReturn(queueConnection);
    when(queueConnectionFactory.createQueueConnection(anyString(), anyString())).thenReturn(queueConnection);
    QueueSession queueSession = Mockito.mock(QueueSession.class);
    bytesMessage = Mockito.mock(BytesMessage.class);
    when(queueSession.createBytesMessage()).thenReturn(bytesMessage);
    when(queueConnection.createQueueSession(anyBoolean(), anyInt())).thenReturn(queueSession);
    when(queueSession.createSender(eq(queue))).thenReturn(queueSender);
    transport = new SmtpJmsTransport(Session.getDefaultInstance(new Properties()), new URLName("jms"));
    transportListener = Mockito.mock(TransportListener.class);
    transport.addTransportListener(transportListener);
}

/**
 * Create a default Session for this ConnectionFactory,
 * adaptign to JMS 1.0.2 style queue/topic mode if necessary.
 * @param con the JMS Connection to operate on
 * @param mode the Session acknowledgement mode
 * ({@code Session.TRANSACTED} or one of the common modes)
 * @return the newly created Session
 * @throws JMSException if thrown by the JMS API
 */
protected Session createSession(Connection con, Integer mode) throws JMSException {
    // Determine JMS API arguments...
    boolean transacted = (mode == Session.SESSION_TRANSACTED);
    int ackMode = (transacted ? Session.AUTO_ACKNOWLEDGE : mode);
    // Now actually call the appropriate JMS factory method...
    if (Boolean.FALSE.equals(this.pubSubMode) && con instanceof QueueConnection) {
        return ((QueueConnection) con).createQueueSession(transacted, ackMode);
    }
    else if (Boolean.TRUE.equals(this.pubSubMode) && con instanceof TopicConnection) {
        return ((TopicConnection) con).createTopicSession(transacted, ackMode);
    }
    else {
        return con.createSession(transacted, ackMode);
    }
}

@Test
@Deprecated
public void testTransactionCommit102WithQueue() throws JMSException {
    QueueConnectionFactory cf = mock(QueueConnectionFactory.class);
    QueueConnection con = mock(QueueConnection.class);
    final QueueSession session = mock(QueueSession.class);

    given(cf.createQueueConnection()).willReturn(con);
    given(con.createQueueSession(true, Session.AUTO_ACKNOWLEDGE)).willReturn(session);

    JmsTransactionManager tm = new JmsTransactionManager102(cf, false);
    TransactionStatus ts = tm.getTransaction(new DefaultTransactionDefinition());
    JmsTemplate jt = new JmsTemplate102(cf, false);
    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 testWithQueueConnection() throws JMSException {
    Connection con = mock(QueueConnection.class);

    SingleConnectionFactory scf = new SingleConnectionFactory(con);
    QueueConnection con1 = scf.createQueueConnection();
    con1.start();
    con1.stop();  // should be ignored
    con1.close();  // should be ignored
    QueueConnection con2 = scf.createQueueConnection();
    con2.start();
    con2.stop();  // should be ignored
    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 testWithQueueConnectionFactoryAndJms11Usage() throws JMSException {
    QueueConnectionFactory cf = mock(QueueConnectionFactory.class);
    QueueConnection con = mock(QueueConnection.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 testWithQueueConnectionFactoryAndJms102Usage() throws JMSException {
    QueueConnectionFactory cf = mock(QueueConnectionFactory.class);
    QueueConnection con = mock(QueueConnection.class);

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

    SingleConnectionFactory scf = new SingleConnectionFactory(cf);
    Connection con1 = scf.createQueueConnection();
    con1.start();
    con1.close();  // should be ignored
    Connection con2 = scf.createQueueConnection();
    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 testConnectionFactory102WithQueue() throws JMSException {
    QueueConnectionFactory cf = mock(QueueConnectionFactory.class);
    QueueConnection con = mock(QueueConnection.class);

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

    SingleConnectionFactory scf = new SingleConnectionFactory102(cf, false);
    QueueConnection con1 = scf.createQueueConnection();
    con1.start();
    con1.close();  // should be ignored
    QueueConnection con2 = scf.createQueueConnection();
    con2.start();
    con2.close();  // should be ignored
    scf.destroy();  // should trigger actual close

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

/**
 * This is a JMS spec independent method to create a Session. Please be cautious when
 * making any changes
 *
 * @param connection the JMS Connection
 * @param transacted should the session be transacted?
 * @param ackMode the ACK mode for the session
 * @param jmsSpec should we use the JMS 1.1, 1.0.2b or 2.0
 * @param isQueue is this Session to deal with a Queue?
 * @return a Session created for the given information
 * @throws JMSException on errors, to be handled and logged by the caller
 */
public static Session createSession(Connection connection, boolean transacted, int ackMode,
    String jmsSpec, Boolean isQueue) throws JMSException {

    if (JMSConstants.JMS_SPEC_VERSION_1_1.equals(jmsSpec) || JMSConstants.JMS_SPEC_VERSION_2_0.equals(jmsSpec)
            || isQueue == null) {
        return connection.createSession(transacted, ackMode);

    } else {
        if (isQueue) {
            return ((QueueConnection) connection).createQueueSession(transacted, ackMode);
        } else {
            return ((TopicConnection) connection).createTopicSession(transacted, ackMode);
        }
    }
}

public void testCreateReceiverQueue() throws Exception
{        
    QueueSession session;       
    session = ((QueueConnection)connection).createQueueSession(false, Session.AUTO_ACKNOWLEDGE);

    try
    {
        session.createReceiver(null);
        fail("Should have failed");
    }
    catch (JMSException e)
    {
        assertTrue(e.getMessage().indexOf("destination") != -1);
    }

    session.createReceiver(queue1);
    session.close();
}

public void testCreateReceiverQueueString() throws Exception
{
    QueueSession session;       
    session = ((QueueConnection)connection).createQueueSession(false, Session.AUTO_ACKNOWLEDGE);

    try
    {
        session.createReceiver(null,null);
        fail("Should have failed");
    }
    catch (JMSException e)
    {
        assertTrue(e.getMessage().indexOf("destination") != -1);
    }

    session.createReceiver(queue1,null);
    session.createReceiver(queue1,"JMSMessageID='toto'");
    session.close();
}

/**
 * Constructor
 */
public QueueSenderThread(String name,
                         DummyMessageFactory msgFactory,
                         SynchronizationPoint startSynchro,
                         QueueConnection connection,
                         boolean transacted, 
                         int messageCount,
                         int messageSize,
                         int minDelay,
                         int maxDelay,
                         Queue queue,
                         int deliveryMode,
                         int priority,
                         long timeToLive)
{
    super(name, startSynchro, connection, transacted, Session.AUTO_ACKNOWLEDGE, queue);
    this.msgFactory = msgFactory;
    this.messageCount = messageCount;
    this.messageSize = messageSize;
    this.minDelay = minDelay;
    this.maxDelay = maxDelay;
    this.deliveryMode = deliveryMode;
    this.priority = priority;
    this.timeToLive = timeToLive;
}

@Override
public QueueConnection createQueueConnection(String userName, String password) throws JMSException
   {
       URI providerURL = getProviderURI();

       String scheme = providerURL.getScheme();
       if (scheme.equals(PacketTransportType.VM))
       {
           String engineName = providerURL.getHost();
           return FFMQEngine.getDeployedInstance(engineName).openQueueConnection(userName, password, clientID);
       }
       else 
       if (scheme.equals(PacketTransportType.TCP) ||
           scheme.equals(PacketTransportType.TCPS) ||
           scheme.equals(PacketTransportType.TCPNIO))
       {
           return new RemoteQueueConnection(providerURL, userName, password, clientID);
       }
       else
           throw new FFMQException("Unknown transport protocol : " + scheme,"INVALID_TRANSPORT_PROTOCOL");
   }

@Override
public QueueConnection createQueueConnection(String username, String password) throws JMSException {
    JmsQueueConnection connection = null;

    try {
        JmsConnectionInfo connectionInfo = configureConnectionInfo(username, password);
        Provider provider = createProvider(remoteURI);

        connection = new JmsQueueConnection(connectionInfo, provider);
        connection.setExceptionListener(exceptionListener);
        connection.connect();
    } catch (Exception e) {
        if (connection != null) {
            try {
                connection.close();
            } catch (Throwable ignored) {}
        }
        throw JmsExceptionSupport.create(e);
    }

    return connection;
}

@Test(timeout=20000)
public void testCreateQueueConnectionGoodProviderString() throws Exception {
    try (TestAmqpPeer testPeer = new TestAmqpPeer();) {
        // Ignore errors from peer close due to not sending any Open / Close frames
        testPeer.setSuppressReadExceptionOnClose(true);

        // DONT create a test fixture, we will drive everything directly.
        testPeer.expectSaslAnonymous();

        JmsConnectionFactory factory = new JmsConnectionFactory("amqp://127.0.0.1:" + testPeer.getServerPort());
        QueueConnection connection = factory.createQueueConnection();
        assertNotNull(connection);

        testPeer.waitForAllHandlersToComplete(1000);

        testPeer.expectOpen();
        testPeer.expectClose();

        connection.close();

        testPeer.waitForAllHandlersToCompleteNoAssert(1000);
    }
}