transactionimpl.java

来自「RESIN 3.2 最新源码」· Java 代码 · 共 1,204 行 · 第 1/3 页

/*
 * Copyright (c) 1998-2008 Caucho Technology -- all rights reserved
 *
 * This file is part of Resin(R) Open Source
 *
 * Each copy or derived work must preserve the copyright notice and this
 * notice unmodified.
 *
 * Resin Open Source is free software; you can redistribute it and/or modify
 * it under the terms of the GNU General Public License as published by
 * the Free Software Foundation; either version 2 of the License, or
 * (at your option) any later version.
 *
 * Resin Open Source is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE, or any warranty
 * of NON-INFRINGEMENT.  See the GNU General Public License for more
 * details.
 *
 * You should have received a copy of the GNU General Public License
 * along with Resin Open Source; if not, write to the
 *
 *   Free Software Foundation, Inc.
 *   59 Temple Place, Suite 330
 *   Boston, MA 02111-1307  USA
 *
 * @author Scott Ferguson
 */

package com.caucho.transaction;

import com.caucho.jca.UserTransactionImpl;
import com.caucho.jca.UserTransactionSuspendState;
import com.caucho.log.Log;
import com.caucho.transaction.xalog.AbstractXALogStream;
import com.caucho.util.Alarm;
import com.caucho.util.AlarmListener;
import com.caucho.util.CharBuffer;
import com.caucho.util.L10N;

import javax.transaction.*;
import javax.transaction.xa.XAException;
import javax.transaction.xa.XAResource;
import javax.transaction.xa.Xid;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.logging.Level;
import java.util.logging.Logger;

/**
 * Implementation of the transaction.  Transactions are normally
 * associated with a single thread.
 */
public class TransactionImpl implements Transaction, AlarmListener {
  private static final Logger log = Log.open(TransactionImpl.class);
  private static final L10N L = new L10N(TransactionImpl.class);

  private final static long DEFAULT_TIMEOUT = 60000;
  private final static long EXTRA_TIMEOUT = 60000;
  private final static long MAX_TIMEOUT = 24 * 3600 * 1000L;

  // flag when the resource is active (between getConnection() and close())
  private final static int RES_ACTIVE    = 0x1;
  // flag when the resource shares another resource manager
  private final static int RES_SHARED_RM = 0x2;
  // flag when the resource is suspended
  private final static int RES_SUSPENDED = 0x4;
  // flag when the resource wants to commit
  private final static int RES_COMMIT = 0x8;

  private TransactionManagerImpl _manager;

  // state of the transaction
  private int _status;

  // The transaction id for the resource
  private XidImpl _xid;

  // true if the transaction is suspended.
  private boolean _isSuspended;
  private boolean _isDead;

  // How many resources are available
  private int _resourceCount;
  // The current resources in the transaction
  private XAResource []_resources;
  // The xids for the resources
  private XidImpl []_resourceXid;
  // Whether the resources are active (between begin/end) or not
  private int []_resourceState;

  private UserTransactionImpl _userTransaction;
  private UserTransactionSuspendState _suspendState;

  private HashMap<String,Object> _props;

  private ArrayList<Synchronization> _syncList;

  private Throwable _rollbackException;

  private AbstractXALogStream _xaLog;

  private long _timeout = 0;
  private Alarm _alarm;

  /**
   * Creates a new transaction.
   *
   * @param manager the owning transaction manager
   */
  TransactionImpl(TransactionManagerImpl manager)
  {
    _manager = manager;
    _timeout = _manager.getTimeout();
    _status = Status.STATUS_NO_TRANSACTION;
    _alarm = new Alarm("xa-timeout", this, ClassLoader.getSystemClassLoader());
  }

  /**
   * Sets the user transaction.
   */
  public void setUserTransaction(UserTransactionImpl ut)
  {
    _userTransaction = ut;
  }

  /**
   * Returns true if the transaction has any associated resources.
   */
  boolean hasResources()
  {
    return _resourceCount > 0;
  }

  /**
   * Returns true if the transaction is currently suspended.
   */
  boolean isSuspended()
  {
    return _isSuspended;
  }

  /**
   * Returns true if the transaction is dead, i.e. failed for some reason.
   */
  boolean isDead()
  {
    return _isDead;
  }

  /**
   * Return true if the transaction has no resources.
   */
  public boolean isEmpty()
  {
    if (_isDead)
      return true;
    else if (_resourceCount > 0)
      return false;
    // XXX: ejb/3692 because TransactionContext adds itself
    else if (_syncList != null && _syncList.size() > 1)
      return false;
    else
      return true;
  }

  /**
   * Start a transaction.
   */
  void begin()
    throws SystemException, NotSupportedException
  {
    if (_status != Status.STATUS_NO_TRANSACTION) {
      int status = _status;

      try {
        rollback();
      } catch (Throwable e) {
        log.log(Level.WARNING, e.toString(), e);
      }

      throw new NotSupportedException(L.l("Nested transactions are not supported. The previous transaction for this thread did not commit() or rollback(). Check that every UserTransaction.begin() has its commit() or rollback() in a finally block.\nStatus was {0}.",
                                          xaState(status)));
    }

    if (_isDead)
      throw new IllegalStateException(L.l("Error trying to use dead transaction."));

    _status = Status.STATUS_ACTIVE;

    _rollbackException = null;

    if (_xid == null)
      _xid = _manager.createXID();

    if (log.isLoggable(Level.FINE))
      log.fine(this + " begin");

    if (_timeout > 0) {
      _alarm.queue(_timeout + EXTRA_TIMEOUT);
    }
  }

  /**
   * Enlists a resource with the current transaction.  Example
   * resources are database or JMS connections.
   *
   * @return true if successful
   */
  public boolean enlistResource(XAResource resource)
    throws RollbackException, SystemException
  {
    if (resource == null)
      throw new IllegalArgumentException(L.l("Resource must not be null in enlistResource"));

    if (_isSuspended)
      throw new IllegalStateException(L.l("can't enlist with suspended transaction"));
    if (_status == Status.STATUS_ACTIVE) {
      // normal
    }
    else {
      // validate the status
      if (_status != Status.STATUS_MARKED_ROLLBACK) {
      }
      else if (_rollbackException != null)
        throw RollbackExceptionWrapper.create(_rollbackException);
      else
        throw new RollbackException(L.l("can't enlist with rollback-only transaction"));

      if (_status == Status.STATUS_NO_TRANSACTION)
        throw new IllegalStateException(L.l("can't enlist with inactive transaction"));

      throw new IllegalStateException(L.l("can't enlist with transaction in '{0}' state",
                                          xaState(_status)));
    }

    // creates enough space in the arrays for the resource
    if (_resources == null) {
      _resources = new XAResource[16];
      _resourceState = new int[16];
      _resourceXid = new XidImpl[16];
    }
    else if (_resources.length <= _resourceCount) {
      int oldLength = _resources.length;
      int newLength = 2 * oldLength;

      XAResource []resources = new XAResource[newLength];
      int []resourceState = new int[newLength];
      XidImpl []resourceXid = new XidImpl[newLength];

      System.arraycopy(_resources, 0, resources, 0, oldLength);
      System.arraycopy(_resourceState, 0, resourceState, 0, oldLength);
      System.arraycopy(_resourceXid, 0, resourceXid, 0, oldLength);

      _resources = resources;
      _resourceState = resourceState;
      _resourceXid = resourceXid;
    }

    int flags = XAResource.TMNOFLAGS;

    // Active transaction will call the XAResource.start() method
    // to let the resource manager know that the resource is managed.
    //
    // If the resource uses the same resource manager as one of the
    // current resources, issue a TMJOIN message.
    XidImpl xid = _xid;
    boolean hasNewResource = true;

    for (int i = 0; i < _resourceCount; i++) {
      if (_resources[i] != resource) {
      }
      else if ((_resourceState[i] & RES_ACTIVE) != 0) {
        IllegalStateException exn;
        exn = new IllegalStateException(L.l("Can't enlist same resource twice.  Delist is required before calling enlist with an old resource."));

        setRollbackOnly(exn);
        throw exn;
      }

      try {
        if (_resources[i].isSameRM(resource)) {
          flags = XAResource.TMJOIN;
          xid = _resourceXid[i];

          if ((_resourceState[i] & RES_ACTIVE) == 0) {
            _resources[i] = resource;
            _resourceState[i] |= RES_ACTIVE;
            hasNewResource = false;
          }

          break;
        }
      } catch (Exception e) {
        log.log(Level.FINE, e.toString(), e);
      }
    }

    if (_resourceCount > 0 && flags != XAResource.TMJOIN)
      xid = new XidImpl(_xid, _resourceCount + 1);

    try {
      if (_timeout > 0)
        resource.setTransactionTimeout((int) (_timeout / 1000L));

      if (log.isLoggable(Level.FINER)) {
        if (flags == XAResource.TMJOIN)
          log.finer(this + " join-XA " + resource);
        else
          log.finer(this + " start-XA " + resource);
      }

      resource.start(xid, flags);
    } catch (XAException e) {
      setRollbackOnly(e);
      throw new SystemException(e);
    }

    if (hasNewResource) {
      _resources[_resourceCount] = resource;
      _resourceState[_resourceCount] = RES_ACTIVE;

      if (flags == XAResource.TMJOIN)
        _resourceState[_resourceCount] |= RES_SHARED_RM;

      _resourceXid[_resourceCount] = xid;
      _resourceCount++;
    }

    return true;
  }

  /**
   * Returns true if the local transaction optimization would be allowed.
   */
  public boolean allowLocalTransactionOptimization()
  {
    // XXX: can also check if all are non-local
    return _resourceCount == 0;
  }

  /**
   * Returns the current number of resources.
   */
  public int getEnlistedResourceCount()
  {
    return _resourceCount;
  }

  /**
   * delists a resource from the current transaction
   *
   * @param resource the resource to delist
   * @param flag XXX: ???
   *
   * @return true if successful
   */
  public boolean delistResource(XAResource resource, int flag)
    throws SystemException
  {
    if (_isSuspended)
      throw new IllegalStateException(L.l("transaction is suspended"));

    if (_resourceCount == 0)
      return true;

    int index;

    for (index = _resourceCount - 1; index >= 0; index--) {
      if (_resources[index].equals(resource))
        break;
    }

    if (index < 0)
      return false;

    // If there is no current transaction,
    // remove it from the resource list entirely.
    if (_status == Status.STATUS_NO_TRANSACTION) {
      for (; index + 1 < _resourceCount; index++) {
        _resources[index] = _resources[index + 1];
        _resourceState[index] = _resourceState[index + 1];
        _resourceXid[index] = _resourceXid[index + 1];
      }

      _resourceCount--;

      return true;
    }

    if (_status == Status.STATUS_MARKED_ROLLBACK)
      flag = XAResource.TMFAIL;

    _resourceState[index] &= ~RES_ACTIVE;

    try {
      resource.end(_resourceXid[index], flag);
    } catch (XAException e) {
      setRollbackOnly(e);
      throw new SystemException(e);
    }