geronimo-scm mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From djen...@apache.org
Subject cvs commit: incubator-geronimo/modules/transaction/src/java/org/apache/geronimo/transaction/manager XidImporter.java TransactionImpl.java TransactionManagerImpl.java XidFactory.java XidImpl.java
Date Mon, 23 Feb 2004 20:28:43 GMT
djencks     2004/02/23 12:28:43

  Modified:    modules/connector/src/java/org/apache/geronimo/connector/work
                        GeronimoWorkManager.java WorkerContext.java
               modules/connector/src/test/org/apache/geronimo/connector/work
                        PooledWorkManagerTest.java
               modules/transaction/src/java/org/apache/geronimo/transaction
                        TransactionManagerProxy.java TransactionProxy.java
               modules/transaction/src/java/org/apache/geronimo/transaction/manager
                        TransactionImpl.java TransactionManagerImpl.java
                        XidFactory.java XidImpl.java
  Added:       modules/transaction/src/java/org/apache/geronimo/transaction
                        XAWork.java
               modules/transaction/src/java/org/apache/geronimo/transaction/manager
                        XidImporter.java
  Log:
  add remote tx importing support. no tests yet.
  
  Revision  Changes    Path
  1.2       +17 -29    incubator-geronimo/modules/connector/src/java/org/apache/geronimo/connector/work/GeronimoWorkManager.java
  
  Index: GeronimoWorkManager.java
  ===================================================================
  RCS file: /home/cvs/incubator-geronimo/modules/connector/src/java/org/apache/geronimo/connector/work/GeronimoWorkManager.java,v
  retrieving revision 1.1
  retrieving revision 1.2
  diff -u -r1.1 -r1.2
  --- GeronimoWorkManager.java	23 Jan 2004 05:56:11 -0000	1.1
  +++ GeronimoWorkManager.java	23 Feb 2004 20:28:42 -0000	1.2
  @@ -66,11 +66,10 @@
   import org.apache.geronimo.connector.work.pool.StartWorkExecutorPool;
   import org.apache.geronimo.connector.work.pool.SyncWorkExecutorPool;
   import org.apache.geronimo.connector.work.pool.WorkExecutorPool;
  -import org.apache.geronimo.gbean.GAttributeInfo;
   import org.apache.geronimo.gbean.GBeanInfo;
   import org.apache.geronimo.gbean.GBeanInfoFactory;
   import org.apache.geronimo.gbean.GConstructorInfo;
  -import org.apache.geronimo.gbean.GOperationInfo;
  +import org.apache.geronimo.transaction.XAWork;
   
   /**
    * WorkManager implementation which uses under the cover three WorkExecutorPool
  @@ -90,8 +89,6 @@
       private final static int DEFAULT_MIN_POOL_SIZE = 0;
       private final static int DEFAULT_MAX_POOL_SIZE = 10;
   
  -    private static final GBeanInfo GBEAN_INFO;
  -
       /**
        * Pool of threads used by this WorkManager in order to process
        * the Work instances submitted via the doWork methods.
  @@ -110,21 +107,24 @@
        */
       private final WorkExecutorPool scheduledWorkExecutorPool;
   
  +    private final XAWork xaWork;
  +
       /**
        * Create a WorkManager.
        */
       public GeronimoWorkManager() {
  -        this(DEFAULT_MIN_POOL_SIZE, DEFAULT_MAX_POOL_SIZE);
  +        this(DEFAULT_MIN_POOL_SIZE, DEFAULT_MAX_POOL_SIZE, null);
       }
   
  -    public GeronimoWorkManager(int minSize, int maxSize) {
  -        this(minSize, maxSize, minSize, maxSize, minSize, maxSize);
  +    public GeronimoWorkManager(int minSize, int maxSize, XAWork xaWork) {
  +        this(minSize, maxSize, minSize, maxSize, minSize, maxSize, xaWork);
       }
   
  -    public GeronimoWorkManager(int syncMinSize, int syncMaxSize, int startMinSize, int
startMaxSize, int schedMinSize, int schedMaxSize) {
  +    public GeronimoWorkManager(int syncMinSize, int syncMaxSize, int startMinSize, int
startMaxSize, int schedMinSize, int schedMaxSize, XAWork xaWork) {
           syncWorkExecutorPool = new SyncWorkExecutorPool(syncMinSize, syncMaxSize);
           startWorkExecutorPool = new StartWorkExecutorPool(startMinSize, startMaxSize);
           scheduledWorkExecutorPool = new ScheduleWorkExecutorPool(schedMinSize, schedMaxSize);
  +        this.xaWork = xaWork;
       }
   
       public int getSyncThreadCount() {
  @@ -204,7 +204,7 @@
               WorkListener workListener)
               throws WorkException {
           WorkerContext workWrapper =
  -                new WorkerContext(work, startTimeout, execContext, workListener);
  +                new WorkerContext(work, startTimeout, execContext, xaWork, workListener);
           workWrapper.setThreadPriority(Thread.currentThread().getPriority());
           syncWorkExecutorPool.executeWork(workWrapper);
       }
  @@ -229,7 +229,7 @@
               WorkListener workListener)
               throws WorkException {
           WorkerContext workWrapper =
  -                new WorkerContext(work, startTimeout, execContext, workListener);
  +                new WorkerContext(work, startTimeout, execContext, xaWork, workListener);
           workWrapper.setThreadPriority(Thread.currentThread().getPriority());
           startWorkExecutorPool.executeWork(workWrapper);
           return System.currentTimeMillis() - workWrapper.getAcceptedTime();
  @@ -254,31 +254,19 @@
               WorkListener workListener)
               throws WorkException {
           WorkerContext workWrapper =
  -                new WorkerContext(work, startTimeout, execContext, workListener);
  +                new WorkerContext(work, startTimeout, execContext, xaWork, workListener);
           workWrapper.setThreadPriority(Thread.currentThread().getPriority());
           scheduledWorkExecutorPool.executeWork(workWrapper);
       }
   
  +    public static final GBeanInfo GBEAN_INFO;
  +
       static {
           GBeanInfoFactory infoFactory = new GBeanInfoFactory(GeronimoWorkManager.class.getName());
  -        infoFactory.addAttribute(new GAttributeInfo("SyncThreadCount", true));
  -        infoFactory.addAttribute(new GAttributeInfo("SyncMinimumPoolSize", true));
  -        infoFactory.addAttribute(new GAttributeInfo("SyncMaximumPoolSize", true));
  -        infoFactory.addAttribute(new GAttributeInfo("StartThreadCount", true));
  -        infoFactory.addAttribute(new GAttributeInfo("StartMinimumPoolSize", true));
  -        infoFactory.addAttribute(new GAttributeInfo("StartMaximumPoolSize", true));
  -        infoFactory.addAttribute(new GAttributeInfo("ScheduledThreadCount", true));
  -        infoFactory.addAttribute(new GAttributeInfo("ScheduledMinimumPoolSize", true));
  -        infoFactory.addAttribute(new GAttributeInfo("ScheduledMaximumPoolSize", true));
  -        infoFactory.addOperation(new GOperationInfo("doWork", new String[]{Work.class.getName()}));
  -        infoFactory.addOperation(new GOperationInfo("doWork", new String[]{Work.class.getName(),
Long.TYPE.getName(), ExecutionContext.class.getName(), WorkListener.class.getName()}));
  -        infoFactory.addOperation(new GOperationInfo("startWork", new String[]{Work.class.getName()}));
  -        infoFactory.addOperation(new GOperationInfo("startWork", new String[]{Work.class.getName(),
Long.TYPE.getName(), ExecutionContext.class.getName(), WorkListener.class.getName()}));
  -        infoFactory.addOperation(new GOperationInfo("scheduleWork", new String[]{Work.class.getName()}));
  -        infoFactory.addOperation(new GOperationInfo("scheduleWork", new String[]{Work.class.getName(),
Long.TYPE.getName(), ExecutionContext.class.getName(), WorkListener.class.getName()}));
  +        infoFactory.addInterface(WorkManager.class, new String[]{"SyncMinimumPoolSize",
"SyncMaximumPoolSize", "StartMinimumPoolSize", "StartMaximumPoolSize", "ScheduledMinimumPoolSize",
"ScheduledMaximumPoolSize", "XAWork"});
           infoFactory.setConstructor(new GConstructorInfo(
  -                new String[]{"SyncMinimumPoolSize", "SyncMaximumPoolSize", "StartMinimumPoolSize",
"StartMaximumPoolSize", "ScheduledMinimumPoolSize", "ScheduledMaximumPoolSize"},
  -                new Class[]{Integer.TYPE, Integer.TYPE, Integer.TYPE, Integer.TYPE, Integer.TYPE,
Integer.TYPE}));
  +                new String[]{"SyncMinimumPoolSize", "SyncMaximumPoolSize", "StartMinimumPoolSize",
"StartMaximumPoolSize", "ScheduledMinimumPoolSize", "ScheduledMaximumPoolSize", "XAWork"},
  +                new Class[]{Integer.TYPE, Integer.TYPE, Integer.TYPE, Integer.TYPE, Integer.TYPE,
Integer.TYPE, XAWork.class}));
           GBEAN_INFO = infoFactory.getBeanInfo();
       }
   
  
  
  
  1.2       +22 -5     incubator-geronimo/modules/connector/src/java/org/apache/geronimo/connector/work/WorkerContext.java
  
  Index: WorkerContext.java
  ===================================================================
  RCS file: /home/cvs/incubator-geronimo/modules/connector/src/java/org/apache/geronimo/connector/work/WorkerContext.java,v
  retrieving revision 1.1
  retrieving revision 1.2
  diff -u -r1.1 -r1.2
  --- WorkerContext.java	23 Jan 2004 05:56:11 -0000	1.1
  +++ WorkerContext.java	23 Feb 2004 20:28:42 -0000	1.2
  @@ -67,6 +67,7 @@
   
   import org.apache.commons.logging.Log;
   import org.apache.commons.logging.LogFactory;
  +import org.apache.geronimo.transaction.XAWork;
   
   import EDU.oswego.cs.dl.util.concurrent.Latch;
   
  @@ -118,7 +119,9 @@
       /**
        * Execution context of the actual work to be executed.
        */
  -    private ExecutionContext executionContext;
  +    private final ExecutionContext executionContext;
  +
  +    private final XAWork xaWork;
   
       /**
        * Listener to be notified during the life-cycle of the work treatment.
  @@ -147,6 +150,8 @@
        */
       public WorkerContext(Work aWork) {
           adaptee = aWork;
  +        executionContext = null;
  +        xaWork = null;
       }
   
       /**
  @@ -162,11 +167,13 @@
        * work completed) occur.
        */
       public WorkerContext(Work aWork, long aStartTimeout,
  -            ExecutionContext execContext,
  -            WorkListener workListener) {
  +                         ExecutionContext execContext,
  +                         XAWork xaWork,
  +                         WorkListener workListener) {
           adaptee = aWork;
           startTimeOut = aStartTimeout;
           executionContext = execContext;
  +        this.xaWork = xaWork;
           if (null != workListener) {
               this.workListener = workListener;
           }
  @@ -303,7 +310,17 @@
                   new WorkEvent(this, WorkEvent.WORK_STARTED, adaptee, null));
           startLatch.release();
           try {
  -            adaptee.run();
  +            if (executionContext == null || executionContext.getXid() == null) {
  +                adaptee.run();
  +            } else {
  +                try {
  +                    xaWork.begin(executionContext.getXid(), executionContext.getTransactionTimeout());
  +                    adaptee.run();
  +                } finally {
  +                    xaWork.end(executionContext.getXid());
  +                }
  +
  +            }
               workListener.workCompleted(
                       new WorkEvent(this, WorkEvent.WORK_COMPLETED, adaptee, null));
           } catch (Throwable e) {
  
  
  
  1.2       +2 -2      incubator-geronimo/modules/connector/src/test/org/apache/geronimo/connector/work/PooledWorkManagerTest.java
  
  Index: PooledWorkManagerTest.java
  ===================================================================
  RCS file: /home/cvs/incubator-geronimo/modules/connector/src/test/org/apache/geronimo/connector/work/PooledWorkManagerTest.java,v
  retrieving revision 1.1
  retrieving revision 1.2
  diff -u -r1.1 -r1.2
  --- PooledWorkManagerTest.java	23 Jan 2004 05:56:11 -0000	1.1
  +++ PooledWorkManagerTest.java	23 Feb 2004 20:28:43 -0000	1.2
  @@ -84,7 +84,7 @@
       private static final int m_tempo = 200;
   
       protected void setUp() throws Exception {
  -        m_workManager = new GeronimoWorkManager(1, 1);
  +        m_workManager = new GeronimoWorkManager(1, 1, null);
       }
   
       public void testDoWork() throws Exception {
  
  
  
  1.3       +98 -15    incubator-geronimo/modules/transaction/src/java/org/apache/geronimo/transaction/TransactionManagerProxy.java
  
  Index: TransactionManagerProxy.java
  ===================================================================
  RCS file: /home/cvs/incubator-geronimo/modules/transaction/src/java/org/apache/geronimo/transaction/TransactionManagerProxy.java,v
  retrieving revision 1.2
  retrieving revision 1.3
  diff -u -r1.2 -r1.3
  --- TransactionManagerProxy.java	30 Jan 2004 01:32:00 -0000	1.2
  +++ TransactionManagerProxy.java	23 Feb 2004 20:28:43 -0000	1.3
  @@ -55,6 +55,11 @@
    */
   package org.apache.geronimo.transaction;
   
  +import java.util.Map;
  +import java.util.HashMap;
  +import java.util.Set;
  +import java.util.HashSet;
  +
   import javax.resource.spi.XATerminator;
   import javax.transaction.HeuristicMixedException;
   import javax.transaction.HeuristicRollbackException;
  @@ -74,6 +79,7 @@
   import org.apache.geronimo.gbean.GConstructorInfo;
   import org.apache.geronimo.gbean.GAttributeInfo;
   import org.apache.geronimo.transaction.manager.TransactionManagerImpl;
  +import org.apache.geronimo.transaction.manager.XidImporter;
   
   /**
    * A wrapper for a TransactionManager that wraps all Transactions in a TransactionProxy
  @@ -83,19 +89,23 @@
    *
    * @version $Revision$ $Date$
    */
  -public class TransactionManagerProxy implements TransactionManager, XATerminator {
  +public class TransactionManagerProxy implements TransactionManager, XATerminator, XAWork
{
   
       public static final GBeanInfo GBEAN_INFO;
   
       private final TransactionManager delegate;
  +    private final XidImporter importer;
       private final ThreadLocal threadTx = new ThreadLocal();
  +    private final Map importedTransactions = new HashMap();
  +    private Set activeTransactions = new HashSet();
   
       /**
        * Constructor taking the TransactionManager to wrap.
        * @param delegate the TransactionManager that should be wrapped
        */
  -    public TransactionManagerProxy(TransactionManager delegate) {
  +    public TransactionManagerProxy(TransactionManager delegate, XidImporter importer) {
           this.delegate = delegate;
  +        this.importer = importer;
       }
   
       /**
  @@ -103,6 +113,7 @@
        */
       public TransactionManagerProxy() {
           this.delegate = new TransactionManagerImpl();
  +        this.importer = null;
       }
   
       public void setTransactionTimeout(int timeout) throws SystemException {
  @@ -174,22 +185,55 @@
       /**
        * @see javax.resource.spi.XATerminator#commit(javax.transaction.xa.Xid, boolean)
        */
  -    public void commit(Xid arg0, boolean arg1) throws XAException {
  -        throw new XAException("Not implemented.");
  +    public void commit(Xid xid, boolean onePhase) throws XAException {
  +        TransactionProxy tx = (TransactionProxy) importedTransactions.remove(xid);
  +        if (tx == null) {
  +            throw new XAException("No imported transaction for xid: " + xid);
  +        }
  +
  +        try {
  +            int status = tx.getStatus();
  +            assert status == Status.STATUS_ACTIVE || status == Status.STATUS_PREPARED;
  +        } catch (SystemException e) {
  +            throw new XAException();
  +        }
  +        importer.commit(tx.getDelegate(), onePhase);
       }
   
       /**
        * @see javax.resource.spi.XATerminator#forget(javax.transaction.xa.Xid)
        */
  -    public void forget(Xid arg0) throws XAException {
  -        throw new XAException("Not implemented.");
  +    public void forget(Xid xid) throws XAException {
  +        TransactionProxy tx = (TransactionProxy) importedTransactions.remove(xid);
  +        if (tx == null) {
  +            throw new XAException("No imported transaction for xid: " + xid);
  +        }
  +
  +        try {
  +            int status = tx.getStatus();
  +            //assert status == Status.STATUS_ACTIVE || status == Status.STATUS_PREPARED;
  +        } catch (SystemException e) {
  +            throw new XAException();
  +        }
  +        importer.forget(tx.getDelegate());
       }
   
       /**
        * @see javax.resource.spi.XATerminator#prepare(javax.transaction.xa.Xid)
        */
  -    public int prepare(Xid arg0) throws XAException {
  -        throw new XAException("Not implemented.");
  +    public int prepare(Xid xid) throws XAException {
  +        TransactionProxy tx = (TransactionProxy) importedTransactions.get(xid);
  +        if (tx == null) {
  +            throw new XAException("No imported transaction for xid: " + xid);
  +        }
  +
  +        try {
  +            int status = tx.getStatus();
  +            assert status == Status.STATUS_ACTIVE;
  +        } catch (SystemException e) {
  +            throw new XAException();
  +        }
  +        return importer.prepare(tx.getDelegate());
       }
   
       /**
  @@ -202,8 +246,47 @@
       /**
        * @see javax.resource.spi.XATerminator#rollback(javax.transaction.xa.Xid)
        */
  -    public void rollback(Xid arg0) throws XAException {
  -        throw new XAException("Not implemented.");
  +    public void rollback(Xid xid) throws XAException {
  +        TransactionProxy tx = (TransactionProxy) importedTransactions.remove(xid);
  +        if (tx == null) {
  +            throw new XAException("No imported transaction for xid: " + xid);
  +        }
  +
  +        try {
  +            int status = tx.getStatus();
  +            assert status == Status.STATUS_ACTIVE || status == Status.STATUS_PREPARED;
  +        } catch (SystemException e) {
  +            throw new XAException();
  +        }
  +        importer.rollback(tx.getDelegate());
  +    }
  +
  +    public void begin(Xid xid, long txTimeoutMillis) throws XAException {
  +        TransactionProxy tx = (TransactionProxy) importedTransactions.get(xid);
  +        if (tx == null) {
  +            try {
  +                tx = new TransactionProxy(importer.importXid(xid));
  +            } catch (SystemException e) {
  +                throw (XAException)new XAException("Could not import xid").initCause(e);
  +            }
  +            importedTransactions.put(xid, tx);
  +        }
  +        if (activeTransactions.contains(tx)) {
  +            throw new XAException("Xid already active");
  +        }
  +        activeTransactions.add(tx);
  +        threadTx.set(tx);
  +        importer.setTransactionTimeout(txTimeoutMillis);
  +    }
  +
  +    public void end(Xid xid) throws XAException {
  +        TransactionProxy tx = (TransactionProxy) importedTransactions.get(xid);
  +        if (tx == null) {
  +            throw new XAException("No imported transaction for xid: " + xid);
  +        }
  +        if (!activeTransactions.remove(tx)) {
  +            throw new XAException("tx not active for xid: " + xid);
  +        }
       }
   
       //for now we use the default constructor.
  @@ -211,17 +294,17 @@
           GBeanInfoFactory infoFactory = new GBeanInfoFactory(TransactionManagerProxy.class.getName());
   
           infoFactory.setConstructor(new GConstructorInfo(
  -                new String[] { "Delegate" },
  -                new Class[] { TransactionManager.class }));
  +                new String[]{"Delegate"},
  +                new Class[]{TransactionManager.class}));
   
           infoFactory.addAttribute(new GAttributeInfo("Delegate", true));
   
  -        infoFactory.addOperation(new GOperationInfo("setTransactionTimeout", new String[]
{Integer.TYPE.getName()}));
  +        infoFactory.addOperation(new GOperationInfo("setTransactionTimeout", new String[]{Integer.TYPE.getName()}));
           infoFactory.addOperation(new GOperationInfo("begin"));
           infoFactory.addOperation(new GOperationInfo("getStatus"));
           infoFactory.addOperation(new GOperationInfo("getTransaction"));
           infoFactory.addOperation(new GOperationInfo("suspend"));
  -        infoFactory.addOperation(new GOperationInfo("resume", new String[] {Transaction.class.getName()}));
  +        infoFactory.addOperation(new GOperationInfo("resume", new String[]{Transaction.class.getName()}));
           infoFactory.addOperation(new GOperationInfo("commit"));
           infoFactory.addOperation(new GOperationInfo("rollback"));
           infoFactory.addOperation(new GOperationInfo("setRollbackOnly"));
  
  
  
  1.2       +5 -1      incubator-geronimo/modules/transaction/src/java/org/apache/geronimo/transaction/TransactionProxy.java
  
  Index: TransactionProxy.java
  ===================================================================
  RCS file: /home/cvs/incubator-geronimo/modules/transaction/src/java/org/apache/geronimo/transaction/TransactionProxy.java,v
  retrieving revision 1.1
  retrieving revision 1.2
  diff -u -r1.1 -r1.2
  --- TransactionProxy.java	23 Jan 2004 18:54:15 -0000	1.1
  +++ TransactionProxy.java	23 Feb 2004 20:28:43 -0000	1.2
  @@ -105,4 +105,8 @@
       public void setRollbackOnly() throws IllegalStateException, SystemException {
           delegate.setRollbackOnly();
       }
  +
  +    Transaction getDelegate() {
  +        return delegate;
  +    }
   }
  
  
  
  1.1                  incubator-geronimo/modules/transaction/src/java/org/apache/geronimo/transaction/XAWork.java
  
  Index: XAWork.java
  ===================================================================
  package org.apache.geronimo.transaction;
  
  import javax.transaction.xa.Xid;
  import javax.transaction.xa.XAException;
  
  /**
   * primarily an interface between the WorkManager/ExecutionContext and the tm.
   *
   * @version $Revision: 1.1 $ $Date: 2004/02/23 20:28:43 $
   *
   * */
  public interface XAWork {
      void begin(Xid xid, long txTimeout) throws XAException;
      void end(Xid xid) throws XAException;
  }
  
  
  
  1.2       +204 -132  incubator-geronimo/modules/transaction/src/java/org/apache/geronimo/transaction/manager/TransactionImpl.java
  
  Index: TransactionImpl.java
  ===================================================================
  RCS file: /home/cvs/incubator-geronimo/modules/transaction/src/java/org/apache/geronimo/transaction/manager/TransactionImpl.java,v
  retrieving revision 1.1
  retrieving revision 1.2
  diff -u -r1.1 -r1.2
  --- TransactionImpl.java	23 Jan 2004 18:54:16 -0000	1.1
  +++ TransactionImpl.java	23 Feb 2004 20:28:43 -0000	1.2
  @@ -63,6 +63,7 @@
   import java.util.List;
   import java.util.Map;
   import java.util.Set;
  +
   import javax.transaction.HeuristicMixedException;
   import javax.transaction.HeuristicRollbackException;
   import javax.transaction.RollbackException;
  @@ -94,9 +95,13 @@
       private Map xaResources = new HashMap(3);
   
       TransactionImpl(XidFactory xidFactory, TransactionLog txnLog) throws SystemException
{
  +        this(xidFactory.createXid(), xidFactory, txnLog);
  +    }
  +
  +    TransactionImpl(Xid xid, XidFactory xidFactory, TransactionLog txnLog) throws SystemException
{
           this.xidFactory = xidFactory;
           this.txnLog = txnLog;
  -        this.xid = xidFactory.createXid();
  +        this.xid = xid;
           try {
               txnLog.begin(xid);
           } catch (IOException e) {
  @@ -114,16 +119,16 @@
   
       public synchronized void setRollbackOnly() throws IllegalStateException, SystemException
{
           switch (status) {
  -        case Status.STATUS_ACTIVE:
  -        case Status.STATUS_PREPARING:
  -            status = Status.STATUS_MARKED_ROLLBACK;
  -            break;
  -        case Status.STATUS_MARKED_ROLLBACK:
  -        case Status.STATUS_ROLLING_BACK:
  -            // nothing to do
  -            break;
  -        default:
  -            throw new IllegalStateException("Cannot set rollback only, status is " + getStateString(status));
  +            case Status.STATUS_ACTIVE:
  +            case Status.STATUS_PREPARING:
  +                status = Status.STATUS_MARKED_ROLLBACK;
  +                break;
  +            case Status.STATUS_MARKED_ROLLBACK:
  +            case Status.STATUS_ROLLING_BACK:
  +                // nothing to do
  +                break;
  +            default:
  +                throw new IllegalStateException("Cannot set rollback only, status is "
+ getStateString(status));
           }
       }
   
  @@ -132,13 +137,13 @@
               throw new IllegalArgumentException("Synchronization is null");
           }
           switch (status) {
  -        case Status.STATUS_ACTIVE:
  -        case Status.STATUS_PREPARING:
  -            break;
  -        case Status.STATUS_MARKED_ROLLBACK:
  -            throw new RollbackException("Transaction is marked for rollback");
  -        default:
  -            throw new IllegalStateException("Status is " + getStateString(status));
  +            case Status.STATUS_ACTIVE:
  +            case Status.STATUS_PREPARING:
  +                break;
  +            case Status.STATUS_MARKED_ROLLBACK:
  +                throw new RollbackException("Transaction is marked for rollback");
  +            default:
  +                throw new IllegalStateException("Status is " + getStateString(status));
           }
           syncList.add(synch);
       }
  @@ -148,12 +153,12 @@
               throw new IllegalArgumentException("XAResource is null");
           }
           switch (status) {
  -        case Status.STATUS_ACTIVE:
  -            break;
  -        case Status.STATUS_MARKED_ROLLBACK:
  -            throw new RollbackException("Transaction is marked for rollback");
  -        default:
  -            throw new IllegalStateException("Status is " + getStateString(status));
  +            case Status.STATUS_ACTIVE:
  +                break;
  +            case Status.STATUS_MARKED_ROLLBACK:
  +                throw new RollbackException("Transaction is marked for rollback");
  +            default:
  +                throw new IllegalStateException("Status is " + getStateString(status));
           }
   
           try {
  @@ -196,11 +201,11 @@
               throw new IllegalArgumentException("XAResource is null");
           }
           switch (status) {
  -        case Status.STATUS_ACTIVE:
  -        case Status.STATUS_MARKED_ROLLBACK:
  -            break;
  -        default:
  -            throw new IllegalStateException("Status is " + getStateString(status));
  +            case Status.STATUS_ACTIVE:
  +            case Status.STATUS_MARKED_ROLLBACK:
  +                break;
  +            default:
  +                throw new IllegalStateException("Status is " + getStateString(status));
           }
           ResourceManager manager = (ResourceManager) xaResources.remove(xaRes);
           if (manager == null) {
  @@ -215,27 +220,21 @@
           }
       }
   
  +    //Transaction method, does 2pc
       public void commit() throws HeuristicMixedException, HeuristicRollbackException, RollbackException,
SecurityException, SystemException {
  -        synchronized (this) {
  -            switch (status) {
  -            case Status.STATUS_ACTIVE:
  -            case Status.STATUS_MARKED_ROLLBACK:
  -                break;
  -            default:
  -                throw new IllegalStateException("Status is " + getStateString(status));
  -            }
  -        }
  +        beforePrepare();
   
  -        beforeCompletion();
  -        endResources();
           try {
  -            LinkedList rms;
  +            if (status == Status.STATUS_MARKED_ROLLBACK) {
  +                rollbackResources(resourceManagers);
  +                throw new RollbackException("Unable to commit");
  +            }
               synchronized (this) {
                   if (status == Status.STATUS_ACTIVE) {
  -                    if (resourceManagers.size() == 0) {
  +                    if (this.resourceManagers.size() == 0) {
                           // nothing to commit
                           status = Status.STATUS_COMMITTED;
  -                    } else if (resourceManagers.size() == 1) {
  +                    } else if (this.resourceManagers.size() == 1) {
                           // one-phase commit decision
                           status = Status.STATUS_COMMITTING;
                       } else {
  @@ -244,12 +243,11 @@
                       }
                   }
                   // resourceManagers is now immutable
  -                rms = resourceManagers;
               }
   
               // one-phase
  -            if (rms.size() == 1) {
  -                ResourceManager manager = (ResourceManager) rms.getFirst();
  +            if (resourceManagers.size() == 1) {
  +                ResourceManager manager = (ResourceManager) resourceManagers.getFirst();
                   try {
                       manager.committer.commit(manager.branchId, true);
                       synchronized (this) {
  @@ -267,74 +265,72 @@
               }
   
               // two-phase
  -            try {
  -                txnLog.prepare(xid);
  -            } catch (IOException e) {
  -                try {
  -                    rollbackResources(rms);
  -                } catch (Exception se) {
  -                    log.error("Unable to rollback after failure to log prepare", se.getCause());
  -                }
  -                SystemException ex = new SystemException("Error logging prepare; transaction
was rolled back)");
  -                ex.initCause(e);
  -                throw ex;
  +            boolean willCommit = internalPrepare();
  +
  +            // notify the RMs
  +            if (willCommit) {
  +                commitResources(resourceManagers);
  +            } else {
  +                rollbackResources(resourceManagers);
  +                throw new RollbackException("Unable to commit");
               }
  -            for (Iterator i = rms.iterator(); i.hasNext();) {
  -                synchronized (this) {
  -                    if (status != Status.STATUS_PREPARING) {
  -                        // we were marked for rollback
  -                        break;
  -                    }
  -                }
  -                ResourceManager manager = (ResourceManager) i.next();
  -                try {
  -                    int vote = manager.committer.prepare(manager.branchId);
  -                    if (vote == XAResource.XA_RDONLY) {
  -                        // we don't need to consider this RM any more
  -                        i.remove();
  -                    }
  -                } catch (XAException e) {
  -                    synchronized (this) {
  -                        status = Status.STATUS_MARKED_ROLLBACK;
  -                    }
  -                }
  +        } finally {
  +            afterCompletion();
  +            synchronized (this) {
  +                status = Status.STATUS_NO_TRANSACTION;
               }
  +        }
  +    }
   
  -            // decision time...
  -            boolean willCommit;
  +    //Used from XATerminator for first phase in a remotely controlled tx.
  +    int prepare() throws SystemException, RollbackException {
  +        beforePrepare();
  +        int result = XAResource.XA_RDONLY;
  +        try {
  +            LinkedList rms;
               synchronized (this) {
  -                willCommit = (status != Status.STATUS_MARKED_ROLLBACK);
  -                if (willCommit) {
  -                    status = Status.STATUS_PREPARED;
  +                if (status == Status.STATUS_ACTIVE) {
  +                    if (resourceManagers.size() == 0) {
  +                        // nothing to commit
  +                        status = Status.STATUS_COMMITTED;
  +                        return result;
  +                    } else {
  +                        // start prepare part of two-phase
  +                        status = Status.STATUS_PREPARING;
  +                    }
                   }
  +                // resourceManagers is now immutable
  +                rms = resourceManagers;
               }
   
  -            // log our decision
  -            try {
  -                if (willCommit) {
  -                    txnLog.commit(xid);
  -                } else {
  -                    txnLog.rollback(xid);
  -                }
  -            } catch (IOException e) {
  -                try {
  -                    rollbackResources(rms);
  -                } catch (Exception se) {
  -                    log.error("Unable to rollback after failure to log decision", se.getCause());
  -                }
  -                SystemException ex = new SystemException("Error logging decision (outcome
is unknown)");
  -                ex.initCause(e);
  -                throw ex;
  -            }
  +            boolean willCommit = internalPrepare();
   
               // notify the RMs
               if (willCommit) {
  -                commitResources(rms);
  +                if (!rms.isEmpty()) {
  +                    result = XAResource.XA_OK;
  +                }
  +
               } else {
                   rollbackResources(rms);
                   throw new RollbackException("Unable to commit");
               }
           } finally {
  +            if (result == XAResource.XA_RDONLY) {
  +                afterCompletion();
  +                synchronized (this) {
  +                    status = Status.STATUS_NO_TRANSACTION;
  +                }
  +            }
  +        }
  +        return result;
  +    }
  +
  +    //used from XATerminator for commit phase of non-readonly remotely controlled tx.
  +    void preparedCommit() throws SystemException {
  +        try {
  +            commitResources(resourceManagers);
  +        } finally {
               afterCompletion();
               synchronized (this) {
                   status = Status.STATUS_NO_TRANSACTION;
  @@ -342,17 +338,94 @@
           }
       }
   
  +    //helper method used by Transaction.commit and XATerminator prepare.
  +    private void beforePrepare() {
  +        synchronized (this) {
  +            switch (status) {
  +                case Status.STATUS_ACTIVE:
  +                case Status.STATUS_MARKED_ROLLBACK:
  +                    break;
  +                default:
  +                    throw new IllegalStateException("Status is " + getStateString(status));
  +            }
  +        }
  +
  +        beforeCompletion();
  +        endResources();
  +    }
  +
  +
  +    //helper method used by Transaction.commit and XATerminator prepare.
  +    private boolean internalPrepare() throws SystemException {
  +        try {
  +            txnLog.prepare(xid);
  +        } catch (IOException e) {
  +            try {
  +                rollbackResources(resourceManagers);
  +            } catch (Exception se) {
  +                log.error("Unable to rollback after failure to log prepare", se.getCause());
  +            }
  +            throw (SystemException) new SystemException("Error logging prepare; transaction
was rolled back)").initCause(e);
  +        }
  +        for (Iterator i = resourceManagers.iterator(); i.hasNext();) {
  +            synchronized (this) {
  +                if (status != Status.STATUS_PREPARING) {
  +                    // we were marked for rollback
  +                    break;
  +                }
  +            }
  +            ResourceManager manager = (ResourceManager) i.next();
  +            try {
  +                int vote = manager.committer.prepare(manager.branchId);
  +                if (vote == XAResource.XA_RDONLY) {
  +                    // we don't need to consider this RM any more
  +                    i.remove();
  +                }
  +            } catch (XAException e) {
  +                synchronized (this) {
  +                    status = Status.STATUS_MARKED_ROLLBACK;
  +                }
  +            }
  +        }
  +
  +        // decision time...
  +        boolean willCommit;
  +        synchronized (this) {
  +            willCommit = (status != Status.STATUS_MARKED_ROLLBACK);
  +            if (willCommit) {
  +                status = Status.STATUS_PREPARED;
  +            }
  +        }
  +
  +        // log our decision
  +        try {
  +            if (willCommit) {
  +                txnLog.commit(xid);
  +            } else {
  +                txnLog.rollback(xid);
  +            }
  +        } catch (IOException e) {
  +            try {
  +                rollbackResources(resourceManagers);
  +            } catch (Exception se) {
  +                log.error("Unable to rollback after failure to log decision", se.getCause());
  +            }
  +            throw (SystemException) new SystemException("Error logging decision (outcome
is unknown)").initCause(e);
  +        }
  +        return willCommit;
  +    }
  +
       public void rollback() throws IllegalStateException, SystemException {
           List rms;
           synchronized (this) {
               switch (status) {
  -            case Status.STATUS_ACTIVE:
  -                status = Status.STATUS_MARKED_ROLLBACK;
  -                break;
  -            case Status.STATUS_MARKED_ROLLBACK:
  -                break;
  -            default:
  -                throw new IllegalStateException("Status is " + getStateString(status));
  +                case Status.STATUS_ACTIVE:
  +                    status = Status.STATUS_MARKED_ROLLBACK;
  +                    break;
  +                case Status.STATUS_MARKED_ROLLBACK:
  +                    break;
  +                default:
  +                    throw new IllegalStateException("Status is " + getStateString(status));
               }
               rms = resourceManagers;
           }
  @@ -368,9 +441,7 @@
                   } catch (Exception se) {
                       log.error("Unable to rollback after failure to log decision", se.getCause());
                   }
  -                SystemException ex = new SystemException("Error logging rollback");
  -                ex.initCause(e);
  -                throw ex;
  +                throw (SystemException) new SystemException("Error logging rollback").initCause(e);
               }
               rollbackResources(rms);
           } finally {
  @@ -494,28 +565,28 @@
   
       private static String getStateString(int status) {
           switch (status) {
  -        case Status.STATUS_ACTIVE:
  -            return "STATUS_ACTIVE";
  -        case Status.STATUS_PREPARING:
  -            return "STATUS_PREPARING";
  -        case Status.STATUS_PREPARED:
  -            return "STATUS_PREPARED";
  -        case Status.STATUS_MARKED_ROLLBACK:
  -            return "STATUS_MARKED_ROLLBACK";
  -        case Status.STATUS_ROLLING_BACK:
  -            return "STATUS_ROLLING_BACK";
  -        case Status.STATUS_COMMITTING:
  -            return "STATUS_COMMITTING";
  -        case Status.STATUS_COMMITTED:
  -            return "STATUS_COMMITTED";
  -        case Status.STATUS_ROLLEDBACK:
  -            return "STATUS_ROLLEDBACK";
  -        case Status.STATUS_NO_TRANSACTION:
  -            return "STATUS_NO_TRANSACTION";
  -        case Status.STATUS_UNKNOWN:
  -            return "STATUS_UNKNOWN";
  -        default:
  -            throw new AssertionError();
  +            case Status.STATUS_ACTIVE:
  +                return "STATUS_ACTIVE";
  +            case Status.STATUS_PREPARING:
  +                return "STATUS_PREPARING";
  +            case Status.STATUS_PREPARED:
  +                return "STATUS_PREPARED";
  +            case Status.STATUS_MARKED_ROLLBACK:
  +                return "STATUS_MARKED_ROLLBACK";
  +            case Status.STATUS_ROLLING_BACK:
  +                return "STATUS_ROLLING_BACK";
  +            case Status.STATUS_COMMITTING:
  +                return "STATUS_COMMITTING";
  +            case Status.STATUS_COMMITTED:
  +                return "STATUS_COMMITTED";
  +            case Status.STATUS_ROLLEDBACK:
  +                return "STATUS_ROLLEDBACK";
  +            case Status.STATUS_NO_TRANSACTION:
  +                return "STATUS_NO_TRANSACTION";
  +            case Status.STATUS_UNKNOWN:
  +                return "STATUS_UNKNOWN";
  +            default:
  +                throw new AssertionError();
           }
       }
   
  @@ -527,6 +598,7 @@
               return false;
           }
       }
  +
   
       private static class ResourceManager {
           private final XAResource committer;
  
  
  
  1.2       +63 -2     incubator-geronimo/modules/transaction/src/java/org/apache/geronimo/transaction/manager/TransactionManagerImpl.java
  
  Index: TransactionManagerImpl.java
  ===================================================================
  RCS file: /home/cvs/incubator-geronimo/modules/transaction/src/java/org/apache/geronimo/transaction/manager/TransactionManagerImpl.java,v
  retrieving revision 1.1
  retrieving revision 1.2
  diff -u -r1.1 -r1.2
  --- TransactionManagerImpl.java	23 Jan 2004 18:54:16 -0000	1.1
  +++ TransactionManagerImpl.java	23 Feb 2004 20:28:43 -0000	1.2
  @@ -64,6 +64,8 @@
   import javax.transaction.SystemException;
   import javax.transaction.Transaction;
   import javax.transaction.TransactionManager;
  +import javax.transaction.xa.Xid;
  +import javax.transaction.xa.XAException;
   
   import org.apache.geronimo.transaction.log.UnrecoverableLog;
   
  @@ -72,7 +74,7 @@
    *
    * @version $Revision$ $Date$
    */
  -public class TransactionManagerImpl implements TransactionManager {
  +public class TransactionManagerImpl implements TransactionManager, XidImporter {
       private final TransactionLog txnLog;
       private final XidFactory xidFactory = new XidFactory();
       private volatile int timeout;
  @@ -153,5 +155,64 @@
           } finally {
               threadTx.set(null);
           }
  +    }
  +
  +    public Transaction importXid(Xid xid) throws XAException, SystemException {
  +        if (getStatus() != Status.STATUS_NO_TRANSACTION) {
  +            throw new XAException("Transaction already active in this thread");
  +        }
  +        TransactionImpl tx = new TransactionImpl(xid, xidFactory, txnLog);
  +        threadTx.set(tx);
  +        return tx;
  +    }
  +
  +    public void commit(Transaction tx, boolean onePhase) throws XAException {
  +        if (onePhase) {
  +            try {
  +                tx.commit();
  +            } catch (HeuristicMixedException e) {
  +                throw new XAException();
  +            } catch (HeuristicRollbackException e) {
  +                throw new XAException();
  +            } catch (RollbackException e) {
  +                throw new XAException();
  +            } catch (SecurityException e) {
  +                throw new XAException();
  +            } catch (SystemException e) {
  +                throw new XAException();
  +            }
  +        } else {
  +            try {
  +                ((TransactionImpl)tx).preparedCommit();
  +            } catch (SystemException e) {
  +                throw new XAException();
  +            }
  +        }
  +    }
  +
  +    public void forget(Transaction tx) throws XAException {
  +    }
  +
  +    public int prepare(Transaction tx) throws XAException {
  +        try {
  +            return ((TransactionImpl)tx).prepare();
  +        } catch (SystemException e) {
  +            throw new XAException();
  +        } catch (RollbackException e) {
  +            throw new XAException();
  +        }
  +    }
  +
  +    public void rollback(Transaction tx) throws XAException {
  +        try {
  +            tx.rollback();
  +        } catch (IllegalStateException e) {
  +            throw new XAException();
  +        } catch (SystemException e) {
  +            throw new XAException();
  +        }
  +    }
  +
  +    public void setTransactionTimeout(long milliseconds) {
       }
   }
  
  
  
  1.2       +8 -3      incubator-geronimo/modules/transaction/src/java/org/apache/geronimo/transaction/manager/XidFactory.java
  
  Index: XidFactory.java
  ===================================================================
  RCS file: /home/cvs/incubator-geronimo/modules/transaction/src/java/org/apache/geronimo/transaction/manager/XidFactory.java,v
  retrieving revision 1.1
  retrieving revision 1.2
  diff -u -r1.1 -r1.2
  --- XidFactory.java	23 Jan 2004 18:54:16 -0000	1.1
  +++ XidFactory.java	23 Feb 2004 20:28:43 -0000	1.2
  @@ -67,7 +67,7 @@
    * <li>4 or 16 byte IP address of host</li>
    * <ol>
    * @version $Revision$ $Date$
  - * @todo Should have a way of setting baseId
  + * todo Should have a way of setting baseId
    */
   public class XidFactory {
       byte[] baseId = new byte[Xid.MAXGTRIDSIZE];
  @@ -106,6 +106,11 @@
       }
   
       public Xid createBranch(Xid globalId, int branch) {
  -        return new XidImpl(globalId, branch);
  +        byte[] branchId = (byte[]) baseId.clone();
  +        branchId[0] = (byte) branch;
  +        branchId[1] = (byte) (branch >>> 8);
  +        branchId[2] = (byte) (branch >>> 16);
  +        branchId[3] = (byte) (branch >>> 24);
  +        return new XidImpl(globalId, branchId);
       }
   }
  
  
  
  1.2       +7 -15     incubator-geronimo/modules/transaction/src/java/org/apache/geronimo/transaction/manager/XidImpl.java
  
  Index: XidImpl.java
  ===================================================================
  RCS file: /home/cvs/incubator-geronimo/modules/transaction/src/java/org/apache/geronimo/transaction/manager/XidImpl.java,v
  retrieving revision 1.1
  retrieving revision 1.2
  diff -u -r1.1 -r1.2
  --- XidImpl.java	23 Jan 2004 18:54:16 -0000	1.1
  +++ XidImpl.java	23 Feb 2004 20:28:43 -0000	1.2
  @@ -76,7 +76,7 @@
        */
       public XidImpl(byte[] globalId) {
           this.globalId = globalId;
  -        this.hash = hash(globalId);
  +        this.hash = hash(0, globalId);
           branchId = new byte[Xid.MAXBQUALSIZE];
       }
   
  @@ -85,28 +85,20 @@
        * @param global the xid of the global transaction this branch belongs to
        * @param branch the branch id
        */
  -    public XidImpl(Xid global, int branch) {
  +    public XidImpl(Xid global, byte[] branch) {
           int hash;
           if (global instanceof XidImpl) {
               globalId = ((XidImpl) global).globalId;
               hash = ((XidImpl) global).hash;
           } else {
               globalId = global.getGlobalTransactionId();
  -            hash = hash(globalId);
  +            hash = hash(0, globalId);
           }
  -        branchId = new byte[Xid.MAXBQUALSIZE];
  -        branchId[0] = (byte) branch;
  -        branchId[1] = (byte) (branch >>> 8);
  -        branchId[2] = (byte) (branch >>> 16);
  -        branchId[3] = (byte) (branch >>> 24);
  -        for (int i = 0; i < 4; i++) {
  -            hash = (hash * 37) + branchId[i];
  -        }
  -        this.hash = hash;
  +        branchId = branch;
  +        this.hash = hash(hash, branchId);
       }
   
  -    private int hash(byte[] id) {
  -        int hash = 0;
  +    private int hash(int hash, byte[] id) {
           for (int i = 0; i < id.length; i++) {
               hash = (hash * 37) + id[i];
           }
  
  
  
  1.1                  incubator-geronimo/modules/transaction/src/java/org/apache/geronimo/transaction/manager/XidImporter.java
  
  Index: XidImporter.java
  ===================================================================
  package org.apache.geronimo.transaction.manager;
  
  import javax.transaction.xa.Xid;
  import javax.transaction.xa.XAException;
  import javax.transaction.Transaction;
  import javax.transaction.SystemException;
  
  /**
   *
   *
   * @version $Revision: 1.1 $ $Date: 2004/02/23 20:28:43 $
   *
   * */
  public interface XidImporter {
  
      Transaction importXid(Xid xid) throws XAException, SystemException;
  
      void commit(Transaction tx, boolean onePhase) throws XAException;
      void forget(Transaction tx) throws XAException;
      int prepare(Transaction tx) throws XAException;
      void rollback(Transaction tx) throws XAException;
      void setTransactionTimeout(long milliseconds);
  }
  
  
  

Mime
View raw message