/*

  * Copyright (C) The Apache Software Foundation. All rights reserved.

  *

  * This software is published under the terms of the Apache Software License

  * version 1.1, a copy of which has been included with this distribution in

  * the LICENSE file.

  * 

  * $Id: XAConsumerThread.java 155459 2005-02-26 13:24:44Z dirkv $

  */

 package org.apache.commons.messagelet;

 import javax.jms.MessageListener;

 import javax.transaction.SystemException;

 import javax.transaction.Transaction;

 import javax.transaction.TransactionManager;

 import javax.transaction.xa.XAResource;

 import org.apache.commons.logging.Log;

 import org.apache.commons.logging.LogFactory;

 import org.apache.commons.messenger.Messenger;

 import org.apache.commons.messenger.XACapable;

 import org.apache.commons.messenger.XACapableAdapter;

 /** 

  * <p><code>XAConsumerThread</code> is a thread which will perform XA processing

  * of messages

  *

  * @author damon.hamacha

  * @author <a href="mailto:jstrachan@apache.org">James Strachan</a>

  * @version $Revision: 155459 $

  */

 public class XAConsumerThread extends ConsumerThread {

         /** Logger */

         private static final Log log = LogFactory.getLog(XAConsumerThread.class);

         private TransactionManager transctionManager;

         private Transaction transaction;

         public XAConsumerThread() {

                 setName("XAConsumer" + getName());

         }

         /**

          * @return the TransactionManager to be used

          * @throws SystemException

          */

         public TransactionManager getTransactionManager() throws SystemException {

                 if (transctionManager == null) {

                         transctionManager = createTransactionManager();

                 }

                 return transctionManager;

         }

         /**

          * Sets the transaction manager to be used

          * 

          * @param transctionManager the transaction manager to be used

          */

         public void setTransactionManager(TransactionManager transctionManager) {

                 this.transctionManager = transctionManager;

         }

         // Implementation methods

     //-------------------------------------------------------------------------    

         /**

          * Factory method to create a TransactionManager via some mechanism.

          * By default this mechanism will lookup in JNDI 

          */

         protected TransactionManager createTransactionManager() throws SystemException {

                 return null;

         }

         /**

          * Enlists any resources with the current transaction.

          * Typically the input Messenger's Session will always be

          * enlisted. Then if the current MessageListener implements XACapable

          * then any of its resources will also be enlisted.

          * 

          * @param transaction the transaction to enlist resources with

          * @throws Exception if the enlistment fails for whatever reason

          */

         protected void enlist(Transaction transaction) throws Exception {

                 XACapable xaCapable = getXACapable( getMessenger() );

                 xaCapable.enlistResources(transaction);

                 MessageListener listener = getListener();

                 if (listener instanceof XACapable) {

                         xaCapable = (XACapable) listener;

                         xaCapable.enlistResources(transaction);

                 }

                 if (listener instanceof BridgeMDO) {

                         BridgeMDO bridge = (BridgeMDO) listener;

                         xaCapable = getXACapable( bridge.getOutputMessenger() );

                         xaCapable.enlistResources(transaction);

                 }

         }

         /**

          * Delists any resources from the current transaction.

          * This includes the current input Messenger's Session as well

          * as any resources used by the MessageListener if it implements

          * XACapable

          * 

          * @param transaction

          * @param flag is the flag used by JTA when delisting resources.

          * It is either XAResource.TMSUCCESS, XAResource.TMSUSPEND, or XAResource.TMFAIL

          * @throws Exception

          */

         protected void delist(Transaction transaction, int flag) throws Exception {

                 XACapable xaCapable = getXACapable( getMessenger() );

                 xaCapable.delistResources(transaction, flag);

                 MessageListener listener = getListener();

                 if (listener instanceof XACapable) {

                         xaCapable = (XACapable) listener;

                         xaCapable.delistResources(transaction, flag);

                 }

                 if (listener instanceof BridgeMDO) {

                         BridgeMDO bridge = (BridgeMDO) listener;

                         xaCapable = getXACapable( bridge.getOutputMessenger() );

                         xaCapable.delistResources(transaction, flag);

                 }

         }

         /**

          * Strategy method to represent the code required to start

          * a transaction.

          */

         protected void startTransaction() throws Exception {

                 getTransactionManager().begin();

                 transaction = getTransactionManager().getTransaction();

                 enlist(transaction);

         }

         /**

          * Strategy method to represent the code required to commit

          * a transaction.

          */

         protected void commitTransaction() throws Exception {

                 delist(transaction, XAResource.TMSUCCESS);

                 try {

                         transaction.commit();

                 }

                 catch (Exception e) {

                         log.error("Caught exception while committing txn: " + e, e);

                         transaction.setRollbackOnly();

                         throw e;                

                 }

         }

         /**

          * Strategy method to represent the code required to rollback

          * a transaction.

          */

         protected void rollbackTransaction() throws Exception {

                 delist(transaction, XAResource.TMFAIL);

                 transaction.rollback();

         }

         /**

          * Strategy method to represent the code required to cancel

          * a transaction. 

          * This is called when a message is not received.

          */

         protected void cancelTransaction() throws Exception {

                 delist(transaction, XAResource.TMFAIL);

                 transaction.rollback();

         }

         /**

          * @return an XACapable for the given Messenger

          */

         protected XACapable getXACapable(Messenger messenger) {                

                 if (messenger instanceof XACapable) {

                         return (XACapable) messenger;

                 }

                 return new XACapableAdapter(messenger);

         }

 }