Java 类javax.jms.XAConnection 实例源码

/**
 * Create JMS {@link XASession} instance on top of the provided {@link Connection} instance.
 *
 * @param xAConnection JMS Connection.
 * @return Session instance.
 * @throws JMSConnectorException Error when creating the XASession.
 */
public XASession createXASession(XAConnection xAConnection) throws JMSConnectorException {
    try {
        if (logger.isDebugEnabled()) {
            logger.debug("Creating a new JMS XASession on: " + this.connectionFactoryString);
        }

        if (JMSConstants.JMS_SPEC_VERSION_1_1.equals(jmsSpec) || JMSConstants.JMS_SPEC_VERSION_2_0
                .equals(jmsSpec)) {
            return xAConnection.createXASession();
        } else if (JMSConstants.JMSDestinationType.QUEUE.equals(this.destinationType)) {
            return ((XAQueueConnection) (xAConnection)).createXAQueueSession();
        } else {
            return ((XATopicConnection) (xAConnection)).createXATopicSession();
        }
    } catch (JMSException e) {
        throw new JMSConnectorException(
                "JMS Exception while obtaining session for factory " + connectionFactoryString, e);
    }
}

@Test
public void testXAPrepare() throws Exception {
   try {

      XAConnection connection = xaFactory.createXAConnection();

      XASession xasession = connection.createXASession();

      Xid xid = newXID();
      xasession.getXAResource().start(xid, XAResource.TMNOFLAGS);
      Queue queue = xasession.createQueue(queueName);
      MessageProducer producer = xasession.createProducer(queue);
      producer.send(xasession.createTextMessage("hello"));
      producer.send(xasession.createTextMessage("hello"));
      xasession.getXAResource().end(xid, XAResource.TMSUCCESS);

      xasession.getXAResource().prepare(xid);

      connection.close();

      System.err.println("Done!!!");
   } catch (Exception e) {
      e.printStackTrace();
   }
}

@Test
public void testXAResourceCommittedRemoved() throws Exception {
   Queue queue = null;

   Xid xid = newXID();
   try (XAConnection xaconnection = xaFactory.createXAConnection()) {
      XASession session = xaconnection.createXASession();
      queue = session.createQueue(queueName);
      session.getXAResource().start(xid, XAResource.TMNOFLAGS);
      MessageProducer producer = session.createProducer(queue);
      producer.send(session.createTextMessage("xa message"));
      session.getXAResource().end(xid, XAResource.TMSUCCESS);
      session.getXAResource().commit(xid, true);
   }
   XidImpl xid1 = new XidImpl(xid);
   Transaction transaction = server.getResourceManager().getTransaction(xid1);
   assertNull(transaction);
}

@Test
public void testXAResourceRolledBackRemoved() throws Exception {
   Queue queue = null;

   Xid xid = newXID();
   try (XAConnection xaconnection = xaFactory.createXAConnection()) {
      XASession session = xaconnection.createXASession();
      queue = session.createQueue(queueName);
      session.getXAResource().start(xid, XAResource.TMNOFLAGS);
      MessageProducer producer = session.createProducer(queue);
      producer.send(session.createTextMessage("xa message"));
      session.getXAResource().end(xid, XAResource.TMSUCCESS);
      session.getXAResource().rollback(xid);
   }
   XidImpl xid1 = new XidImpl(xid);
   Transaction transaction = server.getResourceManager().getTransaction(xid1);
   assertNull(transaction);
}

@Test
public void testIsSamRM() throws Exception {
   XAConnection conn = null;

   conn = xacf.createXAConnection();

   // Create a session
   XASession sess1 = conn.createXASession();
   XAResource res1 = sess1.getXAResource();

   // Create a session
   XASession sess2 = conn.createXASession();
   XAResource res2 = sess2.getXAResource();

   Assert.assertTrue(res1.isSameRM(res2));
}

public void testRollbackXaErrorCode() throws Exception {
   String brokerName = "rollbackErrorCode";
   BrokerService broker = BrokerFactory.createBroker(new URI("broker:(tcp://localhost:0)/" + brokerName));
   broker.start();
   broker.waitUntilStarted();
   ActiveMQXAConnectionFactory cf = new ActiveMQXAConnectionFactory(broker.getTransportConnectors().get(0).getConnectUri());
   XAConnection connection = (XAConnection) cf.createConnection();
   connection.start();
   XASession session = connection.createXASession();
   XAResource resource = session.getXAResource();

   Xid tid = createXid();
   try {
      resource.rollback(tid);
      fail("Expected xa exception on no tx");
   } catch (XAException expected) {
      LOG.info("got expected xa", expected);
      assertEquals("no tx", XAException.XAER_NOTA, expected.errorCode);
   }
   connection.close();
   broker.stop();
}

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 XA connection
 *
 * @param userName The user name
 * @param password The password
 * @return The connection
 * @throws JMSException Thrown if the operation fails
 */
@Override
public XAConnection createXAConnection(final String userName, final String password) throws JMSException {
   if (ActiveMQRAConnectionFactoryImpl.trace) {
      ActiveMQRALogger.LOGGER.trace("createXAConnection(" + userName + ", ****)");
   }

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

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

   return s;
}

/**
 *
 */
private void checkSession() {
   if (session == null) {
      synchronized (this) {
         if (closed)
            throw new IllegalStateRuntimeException("Context is closed");
         if (session == null) {
            try {
               if (xa) {
                  session = ((XAConnection) connection).createXASession();
               } else {
                  session = connection.createSession(sessionMode);
               }
            } catch (JMSException e) {
               throw JmsExceptionUtils.convertToRuntimeException(e);
            }
         }
      }
   }
}

@Override
protected void run(XAConnection connection, XASession xaSession, XAResource xaResource) throws Exception {
    verbose("Starting XA transaction");
    Xid xid = createXid();
    xaResource.start(xid, 0);

    verbose("Sending message");
    MessageProducer producer = xaSession.createProducer(xaSession.createQueue(queue));
    producer.send(xaSession.createTextMessage("TEST"));

    verbose("Ending XA transaction");
    xaResource.end(xid, XAResource.TMSUCCESS);

    verbose("Preparing XA transaction");
    xaResource.prepare(xid);

    println("Created: "+toString(xid));
}

protected JMSSpecificContext createXAContext(final XAConnectionFactory cf, final Options<CreateContextOption> options) {

        if (TransactionUtil.tm == null) {
            throw new NullPointerException("TransactionManager not found; is transactions module on the classpath?");
        }

        XAConnection connection = (XAConnection) DestinationUtil.mightThrow(new Callable() {
            @Override
            public Object call() throws Exception {
                if (options.has(CreateContextOption.USERNAME)) {
                    return cf.createXAConnection(options.getString(CreateContextOption.USERNAME),
                                                 options.getString(CreateContextOption.PASSWORD));
                } else {
                    return cf.createXAConnection();
                }
            }
        });

        return new JMSXAContext(connection, this,
                                (Context.Mode)options.get(CreateContextOption.MODE),
                                options.has(CreateContextOption.HOST));
    }

protected Session session() {
    if (session == null) {
        synchronized (this) {
            if (closed) {
                throw new IllegalStateRuntimeException("Context is closed");
            }
            if (session == null) {
                try {
                    if (xa) {
                        session = XAConnection.class.cast(connection()).createXASession();
                    } else {
                        session = connection().createSession(sessionMode);
                    }
                } catch (final JMSException e) {
                    throw toRuntimeException(e);
                }
            }
        }
    }
    return session;
}

private void notifyUpdate(Queue queue, String msg) throws Exception {
    XAConnection connection = null;

    try {
        connection = xaConnectionFactory.createXAConnection();
        XASession session = connection.createXASession();
        MessageProducer messageProducer = session.createProducer(queue);

        connection.start();
        TextMessage message = session.createTextMessage();
        message.setText(msg);

        messageProducer.send(message);
        messageProducer.close();
    } finally {
        if (connection != null) {
            try {
                connection.close();
            } catch (JMSException e) {
                LOGGER.info("Error closing JMS connection: " + e.getMessage());
            }
        }
    }
}

private void notifyUpdate(Queue queue, String msg) throws Exception {
    XAConnection connection = null;

    try {
        connection = xaConnectionFactory.createXAConnection();
        XASession session = connection.createXASession();
        MessageProducer messageProducer = session.createProducer(queue);

        connection.start();
        TextMessage message = session.createTextMessage();
        message.setText(msg);

        messageProducer.send(message);
        messageProducer.close();
    } finally {
        if (connection != null) {
            try {
                connection.close();
            } catch (JMSException e) {
                LOGGER.info("Error closing JMS connection: " + e.getMessage());
            }
        }
    }
}

@Override
public JmsPooledConnection createPooledConnection(Object xaFactory, ResourceBean bean) throws Exception {
    if (!(xaFactory instanceof XAConnectionFactory))
        throw new IllegalArgumentException("class '" + xaFactory.getClass().getName() + "' does not implement " + XAConnectionFactory.class.getName());
    XAConnectionFactory xaConnectionFactory = (XAConnectionFactory) xaFactory;

    XAConnection xaConnection;
    if (user == null || password == null) {
        if (log.isDebugEnabled()) { log.debug("creating new JMS XAConnection with no credentials"); }
        xaConnection = xaConnectionFactory.createXAConnection();
    }
    else {
        if (log.isDebugEnabled()) { log.debug("creating new JMS XAConnection with user <" + user + "> and password <" + password + ">"); }
        xaConnection = xaConnectionFactory.createXAConnection(user, password);
    }

    JmsPooledConnection jmsPooledConnection = new JmsPooledConnection(this, xaConnection);
    xaStatefulHolders.add(jmsPooledConnection);
    return jmsPooledConnection;
}

protected JmsPooledConnection(PoolingConnectionFactory poolingConnectionFactory, XAConnection connection) {
    this.poolingConnectionFactory = poolingConnectionFactory;
    this.xaConnection = connection;
    this.lastReleaseDate = new Date(MonotonicClock.currentTimeMillis());
    addStateChangeEventListener(new JmsPooledConnectionStateChangeListener());

    if (LrcXAConnectionFactory.class.getName().equals(poolingConnectionFactory.getClassName())) {
        if (log.isDebugEnabled()) { log.debug("emulating XA for resource " + poolingConnectionFactory.getUniqueName() + " - changing twoPcOrderingPosition to ALWAYS_LAST_POSITION"); }
        poolingConnectionFactory.setTwoPcOrderingPosition(Scheduler.ALWAYS_LAST_POSITION);
        if (log.isDebugEnabled()) { log.debug("emulating XA for resource " + poolingConnectionFactory.getUniqueName() + " - changing deferConnectionRelease to true"); }
        poolingConnectionFactory.setDeferConnectionRelease(true);
        if (log.isDebugEnabled()) { log.debug("emulating XA for resource " + poolingConnectionFactory.getUniqueName() + " - changing useTmJoin to true"); }
        poolingConnectionFactory.setUseTmJoin(true);
    }

    this.jmxName = "bitronix.tm:type=JMS,UniqueName=" + ManagementRegistrar.makeValidName(poolingConnectionFactory.getUniqueName()) + ",Id=" + poolingConnectionFactory.incCreatedResourcesCounter();
    ManagementRegistrar.register(jmxName, this);
}

@Override
protected XAConnection createProviderConnection(PooledConnectionKey key) throws JMSException {
    if (connectionFactory instanceof XAConnectionFactory) {
        if (key.getUserName() == null && key.getPassword() == null) {
            return ((XAConnectionFactory) connectionFactory).createXAConnection();
        } else {
            return ((XAConnectionFactory) connectionFactory).createXAConnection(key.getUserName(), key.getPassword());
        }
    } else {
        throw new IllegalStateException("connectionFactory should implement javax.jms.XAConnectionFactory");
    }
}

@Bean
public ConnectionFactory pooledConnectionFactory(
        XAConnectionFactoryWrapper wrapper) throws Exception {
    XAConnectionFactory connectionFactory = mock(XAConnectionFactory.class);
    XAConnection connection = mock(XAConnection.class);
    XASession session = mock(XASession.class);
    TemporaryQueue queue = mock(TemporaryQueue.class);
    XAResource resource = mock(XAResource.class);
    given(connectionFactory.createXAConnection()).willReturn(connection);
    given(connection.createXASession()).willReturn(session);
    given(session.createTemporaryQueue()).willReturn(queue);
    given(session.getXAResource()).willReturn(resource);
    return wrapper.wrapConnectionFactory(connectionFactory);
}

@Bean
public ConnectionFactory pooledConnectionFactory(
        XAConnectionFactoryWrapper wrapper) throws Exception {
    XAConnectionFactory connectionFactory = mock(XAConnectionFactory.class);
    XAConnection connection = mock(XAConnection.class);
    XASession session = mock(XASession.class);
    TemporaryQueue queue = mock(TemporaryQueue.class);
    XAResource resource = mock(XAResource.class);
    given(connectionFactory.createXAConnection()).willReturn(connection);
    given(connection.createXASession()).willReturn(session);
    given(session.createTemporaryQueue()).willReturn(queue);
    given(session.getXAResource()).willReturn(resource);
    return wrapper.wrapConnectionFactory(connectionFactory);
}

/**
 * JAVADOC Method Level Comments
 *
 * @return JAVADOC.
 *
 * @throws JMSException JAVADOC.
 */
@Override
public XAConnection createXAConnection()
    throws JMSException {
    if (StringUtils.isNotEmpty(getUsername())) {
        return createXAConnection(getUsername(), getDecryptedPassword());
    }

    return targetConnectionFactory.createXAConnection();
}

@Bean
public ConnectionFactory pooledConnectionFactory(
        XAConnectionFactoryWrapper wrapper) throws Exception {
    XAConnectionFactory connectionFactory = mock(XAConnectionFactory.class);
    XAConnection connection = mock(XAConnection.class);
    XASession session = mock(XASession.class);
    TemporaryQueue queue = mock(TemporaryQueue.class);
    XAResource resource = mock(XAResource.class);
    given(connectionFactory.createXAConnection()).willReturn(connection);
    given(connection.createXASession()).willReturn(session);
    given(session.createTemporaryQueue()).willReturn(queue);
    given(session.getXAResource()).willReturn(resource);
    return wrapper.wrapConnectionFactory(connectionFactory);
}

/**
 * Construct an <code>ActiveMQConnection</code>
 *
 * @param transport
 * @param factoryStats
 * @throws Exception
 */
protected ActiveMQConnection(final Transport transport, IdGenerator clientIdGenerator, IdGenerator connectionIdGenerator, JMSStatsImpl factoryStats) throws Exception {

    this.transport = transport;
    this.clientIdGenerator = clientIdGenerator;
    this.factoryStats = factoryStats;

    // Configure a single threaded executor who's core thread can timeout if
    // idle
    executor = new ThreadPoolExecutor(1, 1, 5, TimeUnit.SECONDS, new LinkedBlockingQueue<Runnable>(), new ThreadFactory() {
        @Override
        public Thread newThread(Runnable r) {
            Thread thread = new Thread(r, "ActiveMQ Connection Executor: " + transport);
            //Don't make these daemon threads - see https://issues.apache.org/jira/browse/AMQ-796
            //thread.setDaemon(true);
            return thread;
        }
    });
    // asyncConnectionThread.allowCoreThreadTimeOut(true);
    String uniqueId = connectionIdGenerator.generateId();
    this.info = new ConnectionInfo(new ConnectionId(uniqueId));
    this.info.setManageable(true);
    this.info.setFaultTolerant(transport.isFaultTolerant());
    this.connectionSessionId = new SessionId(info.getConnectionId(), -1);

    this.transport.setTransportListener(this);

    this.stats = new JMSConnectionStatsImpl(sessions, this instanceof XAConnection);
    this.factoryStats.addConnection(this);
    this.timeCreated = System.currentTimeMillis();
    this.connectionAudit.setCheckForDuplicates(transport.isFaultTolerant());
}

@Override
public synchronized SessionWrapper create() throws Exception {
    List<ConnectionWrapper> connectionWrappers;
    ConnectionWrapper connectionWrapper = null;
    SessionWrapper sessionWrapper = null;

    if (jmsConnectionFactory instanceof JMSClientConnectionFactory) {
        connectionWrappers = ((JMSClientConnectionFactory) jmsConnectionFactory).getConnections();

        // see if we can create more sessions on the final Connection created
        if (!connectionWrappers.isEmpty()
                && connectionWrappers.get(connectionWrappers.size() - 1).getSessionCount().get()
                < ((JMSClientConnectionFactory) jmsConnectionFactory).getMaxSessionsPerConnection()) {
            connectionWrapper = connectionWrappers.get(connectionWrappers.size() - 1);
        }

        // if it needs to create a new connectionWrapper
        if (connectionWrapper == null) {
            if (jmsConnectionFactory.isxATransacted()) {
                connectionWrapper = new ConnectionWrapper((jmsConnectionFactory.createXAConnection()));
            } else {
                connectionWrapper = new ConnectionWrapper(jmsConnectionFactory.createConnection());
            }
            connectionWrappers.add(connectionWrapper);
        }

        // Create new SessionWrapper (or XASessionWrapper) accordingly
        if (jmsConnectionFactory.isxATransacted()) {
            XASession xASession = jmsConnectionFactory
                    .createXASession((XAConnection) connectionWrapper.getConnection());
            sessionWrapper = new XASessionWrapper(xASession, xASession.getSession(),
                    jmsConnectionFactory.createMessageProducer(xASession.getSession()));
        } else {
            Session session = jmsConnectionFactory.createSession(connectionWrapper.getConnection());
            sessionWrapper = new SessionWrapper(session, jmsConnectionFactory.createMessageProducer(session));
        }
        connectionWrapper.incrementSessionCount();
    }
    return sessionWrapper;
}

@Test
public void testXAInstanceof() throws Exception {
   conn = cf.createConnection();

   assertFalse(conn instanceof XAConnection);
   Session sess = conn.createSession(false, Session.AUTO_ACKNOWLEDGE);

   assertFalse(sess instanceof XASession);
}

@Test
public void testXASimple() throws Exception {
   XAConnection connection = xaFactory.createXAConnection();

   Collection<Session> sessions = new LinkedList<>();

   for (int i = 0; i < 10; i++) {
      XASession session = connection.createXASession();
      session.getXAResource().start(newXID(), XAResource.TMNOFLAGS);
      sessions.add(session);
   }

   connection.close();

}

private void doTestCrashServerAfterXACommit(boolean onePhase) throws Exception {
   ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory();
   XAConnection connection = connectionFactory.createXAConnection();

   try {
      Session session = connection.createSession(true, Session.SESSION_TRANSACTED);
      Queue queue = session.createQueue("Queue1");
      final XASession xaSession = connection.createXASession();
      MessageConsumer consumer = xaSession.createConsumer(queue);

      MessageProducer producer = session.createProducer(queue);
      producer.send(session.createTextMessage("hello " + 1));
      session.commit();

      XAResource xaResource = xaSession.getXAResource();
      final Xid xid = newXID();
      xaResource.start(xid, XAResource.TMNOFLAGS);

      connection.start();
      Assert.assertNotNull(consumer.receive(5000));

      xaResource.end(xid, XAResource.TMSUCCESS);

      try {
         xaResource.commit(xid, onePhase);
         Assert.fail("didn't get expected exception!");
      } catch (XAException xae) {
         if (onePhase) {
            //expected error code is XAER_RMFAIL
            Assert.assertEquals(XAException.XAER_RMFAIL, xae.errorCode);
         } else {
            //expected error code is XA_RETRY
            Assert.assertEquals(XAException.XA_RETRY, xae.errorCode);
         }
      }
   } finally {
      connection.close();
   }
}

public void testCloseSendConnection() throws Exception {
   String brokerName = "closeSend";
   BrokerService broker = BrokerFactory.createBroker(new URI("broker:(tcp://localhost:0)/" + brokerName));
   broker.start();
   broker.waitUntilStarted();
   ActiveMQXAConnectionFactory cf = new ActiveMQXAConnectionFactory(broker.getTransportConnectors().get(0).getConnectUri());
   XAConnection connection = (XAConnection) cf.createConnection();
   connection.start();
   XASession session = connection.createXASession();
   XAResource resource = session.getXAResource();
   Destination dest = new ActiveMQQueue(getName());

   // publish a message
   Xid tid = createXid();
   resource.start(tid, XAResource.TMNOFLAGS);
   MessageProducer producer = session.createProducer(dest);
   ActiveMQTextMessage message = new ActiveMQTextMessage();
   message.setText(getName());
   producer.send(message);

   connection.close();

   //comment out this check as it doesn't apply to artemis
   //assertTransactionGoneFromBroker(tid);

   broker.stop();
}

@Test
public void testGetSession2() throws Exception {
   deployConnectionFactory(0, JMSFactoryType.CF, "ConnectionFactory", "/ConnectionFactory");
   XAConnection conn = getXAConnectionFactory().createXAConnection();
   XASession sess = conn.createXASession();

   sess.getSession();
   conn.close();
}

@Test
public void testGetXAResource2() throws Exception {
   XAConnection conn = getXAConnectionFactory().createXAConnection();
   XASession sess = conn.createXASession();

   sess.getXAResource();
   conn.close();
}

/**
 * Create a XA connection
 *
 * @return The connection
 * @throws JMSException Thrown if the operation fails
 */
@Override
public XAConnection createXAConnection() throws JMSException {
   if (ActiveMQRAConnectionFactoryImpl.trace) {
      ActiveMQRALogger.LOGGER.trace("createXAConnection()");
   }

   ActiveMQRASessionFactoryImpl s = new ActiveMQRASessionFactoryImpl(mcf, cm, getResourceAdapter().getTM(), ActiveMQRAConnectionFactory.XA_CONNECTION);

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

   return s;
}

protected XAConnection createConnection() throws JMSException {
    ActiveMQXAConnectionFactory factory = new ActiveMQXAConnectionFactory(url);
    factory.setUserName(user);
    factory.setPassword(password);
    verbose("Connecting to: " + url);
    XAConnection xaConnection = factory.createXAConnection();
    xaConnection.start();
    verbose("Connected");
    return xaConnection;
}

@Override
protected void run(XAConnection connection, XASession xaSession, XAResource xaResource) throws Exception {
    verbose("Getting prepared transactions");
    Xid[] recover = xaResource.recover(0);

    println("Found " + recover.length + " prepared transactions");
    for (Xid xid : recover) {
        println(toString(xid));
    }
}

/**
 * Creates a XAConnection with the default user identity.
 * <p> The XAConnection is created in stopped mode. No messages
 * will be delivered until the <code>Connection.start</code> method

 * is explicitly called.
 *
 * @return A newly created XAConnection
 * @throws JMSException         If creating the XAConnection fails due to some internal error.
 * @throws JMSSecurityException If client authentication fails due to an invalid user name or password.
 */
public XAConnection createXAConnection() throws JMSException
{
    try
    {
        return new XAConnectionImpl(_connectionDetails, _sslConfig, scheduledExecutor);
    }
    catch (Exception e)
    {
        JMSException jmse = new JMSException("Error creating connection: " + e.getMessage());
        jmse.setLinkedException(e);
        jmse.initCause(e);
        throw jmse;
    }
}

/**
 * Creates a XAConnection with the specified user identity.
 * <p> The XAConnection is created in stopped mode. No messages
 * will be delivered until the <code>Connection.start</code> method
 * is explicitly called.
 *
 * @param username the caller's user name
 * @param password the caller's password
 * @return A newly created XAConnection.
 * @throws JMSException         If creating the XAConnection fails due to some internal error.
 * @throws JMSSecurityException If client authentication fails due to an invalid user name or password.
 */
public XAConnection createXAConnection(String username, String password) throws JMSException
{
    if (_connectionDetails != null)
    {
        _connectionDetails.setUsername(username);
        _connectionDetails.setPassword(password);
    }
    else
    {
        throw new JMSException("A URL must be specified to access XA connections");
    }
    return createXAConnection();
}

protected Connection connection() {
    if (connection == null) {
        try {
            connection = username != null ? factory.createConnection(username, password) : factory.createConnection();
            xa = XAConnection.class.isInstance(connection);
        } catch (final JMSException e) {
            throw toRuntimeException(e);
        }
    }
    return connection;
}

@Test
public void xaCode() throws Exception {
    assertNotNull(xacf);

    final Connection connection = xacf.createXAConnection();
    assertThat(connection, instanceOf(XAConnection.class));
    testConnection(connection);
}

/**
 * Setup the session
 */
public void setup() throws Exception {
    JmsActivation activation = pool.getActivation();
    JmsActivationSpec spec = activation.getActivationSpec();
    Connection connection = activation.getConnection();
    XAResource xaResource = null;
    tm = activation.getTransactionManager();

    // Get the endpoint
    MessageEndpointFactory endpointFactory = activation.getMessageEndpointFactory();

    // Create the session
    if (activation.isDeliveryTransacted) {
        if (connection instanceof XAConnection) {
            log.debug("Delivery is transacted, and client JMS implementation properly implements javax.jms.XAConnection.");
            xaSession = ((XAConnection) connection).createXASession();
            session = xaSession.getSession();
            xaResource = xaSession.getXAResource();
        } else {
            throw new Exception("Delivery is transacted, but client JMS implementation does not properly implement the necessary interfaces as described in section 8 of the JMS 1.1 specification.");
        }
    } else {
        session = connection.createSession(false, spec.getAcknowledgeModeInt());
    }

    endpoint = endpointFactory.createEndpoint(xaResource);

    // Set the message listener
    session.setMessageListener(this);
}

@Override
public XAConnection createXAConnection() throws JMSException {
    try {
        Class<?> clazz = ClassLoaderUtils.loadClass(connectionFactoryClassName);
        ConnectionFactory nonXaConnectionFactory = (ConnectionFactory) clazz.newInstance();
        PropertyUtils.setProperties(nonXaConnectionFactory, properties);

        return new LrcXAConnection(nonXaConnectionFactory.createConnection());
    } catch (Exception ex) {
        throw (JMSException) new JMSException("unable to connect to non-XA resource " + connectionFactoryClassName).initCause(ex);
    }
}

@Override
public XAConnection createXAConnection(String user, String password) throws JMSException {
    try {
        Class<?> clazz = ClassLoaderUtils.loadClass(connectionFactoryClassName);
        ConnectionFactory nonXaConnectionFactory = (ConnectionFactory) clazz.newInstance();
        PropertyUtils.setProperties(nonXaConnectionFactory, properties);

        return new LrcXAConnection(nonXaConnectionFactory.createConnection(user, password));
    } catch (Exception ex) {
        throw (JMSException) new JMSException("unable to connect to non-XA resource " + connectionFactoryClassName).initCause(ex);
    }
}

@Override
public XAConnection createXAConnection() throws JMSException {
    try {
        init();
        return wrappedFactory.createXAConnection();
    } catch (NamingException ex) {
        throw (JMSException) new JMSException("error looking up wrapped XAConnectionFactory at " + name).initCause(ex);
    }
}