Transacted Session Sample Program

• When creating the session, the first parameter in the createQueueSession method is set to true to indicate a transacted session.
• Sending or discarding messages are controlled using the session's commit and rollback methods, respectively. Message are not sent until the session is committed.

package txSession;
                                                                           
import javax.naming.InitialContext;
                                                                           
import javax.jms.Queue;
import javax.jms.Session;
import javax.jms.TextMessage;
import javax.jms.QueueSender;
import javax.jms.DeliveryMode;
import javax.jms.QueueSession;
import javax.jms.QueueConnection;
import javax.jms.QueueConnectionFactory;
                                                                           
public class TxSender
{
    public static void main(String[] args) throws Exception
    {
    |   // get the initial context
    |   InitialContext ctx = new InitialContext();
    |                                                                      
    |   // lookup the queue object
    |   Queue queue = (Queue) ctx.lookup("queue/queue0");
    |                                                                      
    |   // lookup the queue connection factory
    |   QueueConnectionFactory connFactory = (QueueConnectionFactory) ctx.
    |       lookup("queue/connectionFactory");
    |                                                                      
    |   // create a queue connection
    |   QueueConnection queueConn = connFactory.createQueueConnection();
    |                                                                      
    |   // create a queue session
    |   QueueSession queueSession = queueConn.createQueueSession(true,
    |       Session.DUPS_OK_ACKNOWLEDGE);
    |                                                                      
    |   // create a queue sender
    |   QueueSender queueSender = queueSession.createSender(queue);
    |   queueSender.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
    |                                                                      
    |   TextMessage part1 = queueSession.createTextMessage("part one");
    |   TextMessage part2 = queueSession.createTextMessage("part two");
    |   TextMessage part3 = queueSession.createTextMessage("part three");
    |   TextMessage last  = queueSession.createTextMessage("final");
    |                                                                      
    |   queueSender.send(part1);
    |   queueSender.send(part2);
    |   queueSender.send(part3);
    |   queueSender.send(last);
    |                                                                      
    |   // commit the send (actually put messages on queue)
    |   queueSession.commit();
    |                                                                      
    |   System.out.println("TxSender committed 4 messages");
    |                                                                      
    |   // close the queue connection
    |   queueConn.close();
    }
}

• As was the case for the sender, the first parameter in the createQueueSession method is set to true to indicate a transacted session. The second argument is ignored.
• Acknowledgement of messages is controlled using the session's commit and rollback methods.

Rolling back a transacted session is similar to calling the session's recover method on a receiver that uses CLIENT_ACKNOWLEDGE. All messages that were received since the last call to commit will be re-delivered to the session's consumers. Also, rollback will discard all messages sent by a session's producers as described above.

Below is the source for the TxReceiver class:

package txSession;
                                                                           
import javax.naming.InitialContext;
                                                                           
import javax.jms.Queue;
import javax.jms.Session;
import javax.jms.TextMessage;
import javax.jms.QueueSession;
import javax.jms.QueueReceiver;
import javax.jms.QueueConnection;
import javax.jms.QueueConnectionFactory;
                                                                           
public class TxReceiver
{
    public static void main(String[] args) throws Exception
    {
    |   // get the initial context
    |   InitialContext ctx = new InitialContext();
    |                                                                      
    |   // lookup the queue object
    |   Queue queue = (Queue) ctx.lookup("queue/queue0");
    |                                                                      
    |   // lookup the queue connection factory
    |   QueueConnectionFactory connFactory = (QueueConnectionFactory) ctx.
    |       lookup("queue/connectionFactory");
    |                                                                      
    |   // create a queue connection
    |   QueueConnection queueConn = connFactory.createQueueConnection();
    |                                                                      
    |   // create a queue session
    |   QueueSession queueSession = queueConn.createQueueSession(true,
    |       Session.AUTO_ACKNOWLEDGE);
    |                                                                      
    |   // create a queue receiver
    |   QueueReceiver queueReceiver = queueSession.createReceiver(queue);
    |                                                                      
    |   // start the connection
    |   queueConn.start();
    |                                                                      
    |   while (true)
    |   {
    |   |   TextMessage msg = (TextMessage) queueReceiver.receive();
    |   |   System.out.println(msg.getText());
    |   |   if (msg.getText().equals("final"))
    |   |   {
    |   |   |   queueSession.commit(); // acknowledge all
    |   |   |   break;
    |   |   }
    |   }
    |                                                                      
    |   // close the queue connection
    |   queueConn.close();
    }
}