JMS XA Receive Example

This example demonstrates receiving a message within the scope of an XA transaction. When using an XA transaction the message will only be acknowledged and removed from the queue when the transaction is committed. If the transaction is not committed the message maybe redelivered after rollback or during XA recovery.

ActiveMQ is JTA aware, meaning you can use ActiveMQ in an XA transactional environment and participate in XA transactions. It provides the javax.transaction.xa.XAResource interface for that purpose. Users can get a XAConnectionFactory to create XAConnections and XASessions.

In this example we simulate a transaction manager to control the transactions. First we create an XASession for receiving and a normal session for sending. Then we start a new xa transaction and enlist the receiving XASession through its XAResource. We then send two words, 'hello' and 'world', receive them, and let the transaction roll back. The received messages are cancelled back to the queue. Next we start a new transaction with the same XAResource enlisted, but this time we commit the transaction after receiving the messages. Then we check that no more messages are to be received.

Example step-by-step

To run the example, simply type mvn verify -Pexample from this directory

  1. First we need to get an initial context so we can look-up the JMS connection factory and destination objects from JNDI. This initial context will get it's properties from the client-jndi.properties file in the directory ../common/config
    2.            InitialContext initialContext = getContext(0);
  2. We look-up the JMS queue object from JNDI
    3.            Queue queue = (Queue) initialContext.lookup("/queue/exampleQueue");
  3. We perform a lookup on the XA Connection Factory
    4.            XAConnectionFactory cf = (XAConnectionFactory) initialContext.lookup("/XAConnectionFactory");
  4. We create a JMS XAConnection
    5.            connection = cf.createXAConnection();
  5. We Start the connection
    6.            connection.start();
  6. We create a JMS XASession
    7.           XASession xaSession = connection.createXASession();
  7. We create a normal session
    8.           Session normalSession = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
  8. We create a normal Message Producer
    9.            
           MessageProducer normalProducer = normalSession.createProducer(queue);
  9. We get the JMS Session
    10.           Session session = xaSession.getSession();
  10. We create a message consumer
    11.           MessageConsumer xaConsumer = session.createConsumer(queue);
  11. We create two Text Messages
    12.           
          TextMessage helloMessage = session.createTextMessage("hello");
          TextMessage worldMessage = session.createTextMessage("world");
  12. We create a transaction
    13.           Xid xid1 = new XidImpl("xa-example1".getBytes(), 1, UUIDGenerator.getInstance().generateStringUUID().getBytes());
  13. We get the JMS XAResource
    14.           XAResource xaRes = xaSession.getXAResource();
  14. We begin the Transaction work
    15.           xaRes.start(xid1, XAResource.TMNOFLAGS);
  15. We send two messages.
    16.           
         normalProducer.send(helloMessage);
         normalProducer.send(worldMessage);
  16. We receive the messages
    17.           
          TextMessage rm1 = (TextMessage)xaConsumer.receive();
          System.out.println("Message received: " + rm1.getText());
          TextMessage rm2 = (TextMessage)xaConsumer.receive();
          System.out.println("Message received: " + rm2.getText());
  17. We stop the work
    18.           xaRes.end(xid1, XAResource.TMSUCCESS);
  18. We prepare
    19.           xaRes.prepare(xid1);
  19. We roll back the transaction 
    20.           xaRes.rollback(xid1);
  20. We create another transaction 
    21.           Xid xid2 = new XidImpl("xa-example2".getBytes(), 1, UUIDGenerator.getInstance().generateStringUUID().getBytes());
  21. We start the transaction
    22.           xaRes.start(xid2, XAResource.TMNOFLAGS);
  22. We receive those messages again
    23.            
           rm1 = (TextMessage)xaConsumer.receive();
           System.out.println("Message received again: " + rm1.getText());
           rm2 = (TextMessage)xaConsumer.receive();
           System.out.println("Message received again: " + rm2.getText());
  23. We stop the work
    24.           xaRes.end(xid2, XAResource.TMSUCCESS);
  24. We prepare
    25.           xaRes.prepare(xid2);
  25. We commit!
    26.           xaRes.commit(xid2, false);
  26. We check that no more messages are received.
    27.           
          TextMessage rm3 = (TextMessage)xaConsumer.receive(2000);
          if (rm3 == null)
          {
             System.out.println("No message received after commit.");
          }
          else
          {
             result = false;
          }
  27. And finally, always remember to close your JMS connections and resources after use, in a finally block. Closing a JMS connection will automatically close all of its sessions, consumers, producer and browser objects
    28.            finally
           {
              if (initialContext != null)
              {
                initialContext.close();
              }
              if (connection != null)
              {
                 connection.close();
              }
           }