Java 类javax.jms.XASession 实例源码

/**
 * 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);
    }
}

private void enlistResources(final Transaction tx) throws Exception {
   if (JMSBridgeImpl.trace) {
      ActiveMQJMSBridgeLogger.LOGGER.trace("Enlisting resources in tx");
   }

   XAResource resSource = ((XASession) sourceSession).getXAResource();

   tx.enlistResource(resSource);

   XAResource resDest = ((XASession) targetSession).getXAResource();

   tx.enlistResource(resDest);

   if (JMSBridgeImpl.trace) {
      ActiveMQJMSBridgeLogger.LOGGER.trace("Enlisted resources in tx");
   }
}

@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();
}

@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));
}

/**
 * Creates an XASession.
 *
 * @return A newly created XASession.
 * @throws JMSException If the XAConnection fails to create an XASession due to
 *                      some internal error.
 */
public synchronized XASession createXASession() throws JMSException {
    checkNotClosed();
    if (connectionCloseSignaled) {
        throw new IllegalStateException("Object " + toString() + " has been closed");
    }

    XASession xaSession = _delegate.createXASession();

    // Need to alter the class hierarchy to avoid casting
    if (xaSession instanceof XASession_9_1) {
        xaSessions.add((XASession_9_1) xaSession);
    }

    return xaSession;
}

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 XAResource getXAResource() {
    final PooledSessionHolder session;
    try {
        session = safeGetSessionHolder();
    } catch (JMSException e) {
        throw JMSExceptionSupport.createRuntimeException(e);
    }

    if (session.getSession() instanceof XASession) {
        return ((XASession) session.getSession()).getXAResource();
    }

    return null;
}

/**
 * {@inheritDoc}
 */
@Override
protected MessageConsumer createMessageConsumer() throws JMSException {
    XASession session = getSession();
    _conn.start();
    MessageConsumer consumer = session
        .createConsumer(getDestination(session, _destinationName, _destinationTypeQueue));
    return consumer;
}

/**
 * {@inheritDoc}
 */
@Override
public XASession getSession() throws JMSException {
    if (_session == null) {
        if (_conn == null) {
            _conn = _connFactory.createXAConnection();
        }
        _session = _conn.createXASession();
    }
    return _session;
}

@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);
}

@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);
}

@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;
}

private void stopSessionFailover() {
   XASession xaSource = (XASession) sourceSession;
   XASession xaTarget = (XASession) targetSession;

   ((ClientSessionInternal) xaSource.getXAResource()).getSessionContext().releaseCommunications();
   ((ClientSessionInternal) xaTarget.getXAResource()).getSessionContext().releaseCommunications();
}

@Test
public void testSimpleMessageSendAndReceiveXA() throws Exception {
   Xid xid = new XidImpl("xa1".getBytes(), 1, UUIDGenerator.getInstance().generateStringUUID().getBytes());
   XAQueueConnection queueConnection = qraConnectionFactory.createXAQueueConnection();
   XASession s = queueConnection.createXASession();

   XAResource resource = s.getXAResource();
   resource.start(xid, XAResource.TMNOFLAGS);
   Queue q = ActiveMQJMSClient.createQueue(MDBQUEUE);
   MessageProducer mp = s.createProducer(q);
   MessageConsumer consumer = s.createConsumer(q);
   Message message = s.createTextMessage("test");
   mp.send(message);
   queueConnection.start();
   TextMessage textMessage = (TextMessage) consumer.receiveNoWait();
   assertNull(textMessage);
   resource.end(xid, XAResource.TMSUCCESS);
   resource.commit(xid, true);
   resource.start(xid, XAResource.TMNOFLAGS);
   textMessage = (TextMessage) consumer.receiveNoWait();
   resource.end(xid, XAResource.TMSUCCESS);
   resource.commit(xid, true);
   assertNotNull(textMessage);
   assertEquals(textMessage.getText(), "test");

   // When I wrote this call, this method was doing an infinite loop.
   // this is just to avoid such thing again
   textMessage.getJMSDeliveryTime();

}

@Test
public void testOutgoingXAResourceWrapper() throws Exception {
   XAQueueConnection queueConnection = qraConnectionFactory.createXAQueueConnection();
   XASession s = queueConnection.createXASession();

   XAResource resource = s.getXAResource();
   assertTrue(resource instanceof ActiveMQXAResourceWrapper);

   ActiveMQXAResourceWrapperImpl xaResourceWrapper = (ActiveMQXAResourceWrapperImpl) resource;
   assertTrue(xaResourceWrapper.getJndiName().equals("java://jmsXA NodeId:" + server.getNodeID()));
   assertTrue(xaResourceWrapper.getProductVersion().equals(VersionLoader.getVersion().getFullVersion()));
   assertTrue(xaResourceWrapper.getProductName().equals(ActiveMQResourceAdapter.PRODUCT_NAME));
}

@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 session
 *
 * @return The XA session
 * @throws JMSException Thrown if an error occurs
 */
@Override
public XASession createXASession() throws JMSException {
   if (ActiveMQRASessionFactoryImpl.trace) {
      ActiveMQRALogger.LOGGER.trace("createXASession()");
   }

   checkClosed();
   return allocateConnection(type);
}

@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));
    }
}

/**
 * create an XA Session and start it if required.
 */
public XASession createXASession(int prefetchHigh, int prefetchLow) throws JMSException
{
    _conn.checkNotClosed();

    if (_conn.channelLimitReached())
    {
        throw new ChannelLimitReachedException(_conn.getMaximumChannelCount());
    }

    int channelId = _conn.getNextChannelID();
    XASession_0_10 session;
    try
    {
        session = new XASession_0_10(_qpidConnection, _conn, channelId, prefetchHigh, prefetchLow);
        _conn.registerSession(channelId, session);
        if (_conn._started)
        {
            session.start();
        }
    }
    catch (Exception e)
    {
        throw new JMSAMQException("cannot create session", e);
    }
    return session;
}

@Override
public boolean enlist() throws Exception {
    if (TransactionUtil.tm.getTransaction() == null) {
        return super.isXAEnabled();
    } else if (!WunderBoss.inContainer() ||
                isRemote()) {
        XAResource resource = ((XASession)jmsSession()).getXAResource();
        return TransactionUtil.tm.getTransaction().enlistResource(resource);
    } else {
        return true;
    }
}

private void testXAConnection() throws JMSException {
    if (!poolingConnectionFactory.getTestConnections()) {
        if (log.isDebugEnabled()) { log.debug("not testing connection of " + this); }
        return;
    }

    if (log.isDebugEnabled()) { log.debug("testing connection of " + this); }
    XASession xaSession = xaConnection.createXASession();
    try {
        TemporaryQueue tq = xaSession.createTemporaryQueue();
        tq.delete();
    } finally {
        xaSession.close();
    }
}

@Override
public XASession createXASession() throws JMSException {

    return addSession(xaConnection.createXASession());
}

@Override
public XASession createXASession() throws JMSException {

    return addSession(xaTopicConnection.createXASession());
}

@Override
public XASession createXASession() throws JMSException {

    return addSession( ((XAConnection) connection).createXASession());
}

@Override
public Session getSession() throws JMSException {

    return addSession( ((XASession) session).getSession());
}

@Override
public XAResource getXAResource() {

    return ((XASession) session).getXAResource();
}

/**
 * {@inheritDoc}
 */
@Override
protected MessageProducer createMessageProducer() throws JMSException {
    XASession session = getSession();
    return session.createProducer(getDestination(session, _destinationName, _destinationTypeQueue));
}

public XASession createXASession() throws JMSException {
    return (XASession) createSession(true, Session.SESSION_TRANSACTED);
}

public XASessionWrapper(XASession xASession, Session session, MessageProducer messageProducer) {
    super(session, messageProducer);
    this.xASession = xASession;
}