package org.jboss.tm;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import javax.resource.spi.work.Work;
import javax.resource.spi.work.WorkCompletedException;
import javax.resource.spi.work.WorkException;
import javax.transaction.HeuristicMixedException;
import javax.transaction.HeuristicRollbackException;
import javax.transaction.RollbackException;
import javax.transaction.Status;
import javax.transaction.Synchronization;
import javax.transaction.SystemException;
import javax.transaction.Transaction;
import javax.transaction.xa.XAException;
import javax.transaction.xa.XAResource;
import javax.transaction.xa.Xid;
import org.jboss.logging.Logger;
import org.jboss.util.timeout.Timeout;
import org.jboss.util.timeout.TimeoutFactory;
import org.jboss.util.timeout.TimeoutTarget;
public class TransactionImpl
implements Transaction, TimeoutTarget
{
private static final int HEUR_NONE = XAException.XA_RETRY;
private final static int RS_NEW = 0; private final static int RS_ENLISTED = 1; private final static int RS_SUSPENDED = 2; private final static int RS_ENDED = 3; private final static int RS_VOTE_READONLY = 4; private final static int RS_VOTE_OK = 5; private final static int RS_FORGOT = 6;
private static Logger log = Logger.getLogger(TransactionImpl.class);
private boolean trace = log.isTraceEnabled();
private XidImpl xid;
private GlobalId gid;
private HashSet threads = new HashSet(1);
private HashMap transactionLocalMap = new HashMap();
private Throwable cause;
private Synchronization[] sync = new Synchronization[3];
private int syncAllocSize = 3;
private int syncCount = 0;
private ArrayList resources = new ArrayList(3);
private Resource lastResource;
private boolean resourcesEnded = false;
private long lastBranchId = 0;
private int status;
private int heuristicCode = HEUR_NONE;
private long start;
private Timeout timeout;
private long timeoutPeriod;
private boolean locked = false;
private Work work;
private boolean done = false;
static XidFactoryMBean xidFactory;
static TransactionManagerService txManagerService;
static void defaultXidFactory()
{
if (xidFactory == null)
xidFactory = new XidFactory();
}
TransactionImpl(long timeout)
{
xid = xidFactory.newXid();
gid = xid.getTrulyGlobalId();
status = Status.STATUS_ACTIVE;
start = System.currentTimeMillis();
this.timeout = TimeoutFactory.createTimeout(start+timeout, this);
this.timeoutPeriod = timeout;
if (trace)
log.trace("Created new instance for tx=" + toString());
}
TransactionImpl(GlobalId gid, long timeout)
{
this.gid = gid;
xid = xidFactory.newXid();
status = Status.STATUS_ACTIVE;
start = System.currentTimeMillis();
this.timeout = TimeoutFactory.createTimeout(start+timeout, this);
this.timeoutPeriod = timeout;
if (trace)
log.trace("Created new instance for tx=" + toString());
}
public void timedOut(Timeout timeout)
{
lock();
try
{
log.warn("Transaction " + toString() + " timed out." +
" status=" + getStringStatus(status));
if (this.timeout == null)
return; this.timeout = null;
switch (status)
{
case Status.STATUS_ROLLEDBACK:
case Status.STATUS_COMMITTED:
case Status.STATUS_NO_TRANSACTION:
return;
case Status.STATUS_ROLLING_BACK:
return;
case Status.STATUS_COMMITTING:
gotHeuristic(null, XAException.XA_HEURMIX);
status = Status.STATUS_MARKED_ROLLBACK;
return;
case Status.STATUS_PREPARED:
case Status.STATUS_ACTIVE:
status = Status.STATUS_MARKED_ROLLBACK;
case Status.STATUS_MARKED_ROLLBACK:
interruptThreads();
return;
case Status.STATUS_PREPARING:
status = Status.STATUS_MARKED_ROLLBACK;
return;
default:
log.warn("Unknown status at timeout, tx=" + toString());
return;
}
}
finally
{
unlock();
}
}
public void commit()
throws RollbackException,
HeuristicMixedException,
HeuristicRollbackException,
java.lang.SecurityException,
java.lang.IllegalStateException,
SystemException
{
lock();
try
{
if (trace)
log.trace("Committing, tx=" + this + ", status=" + getStringStatus(status));
beforePrepare();
if (status == Status.STATUS_ACTIVE)
{
switch (getCommitStrategy())
{
case 0:
if (trace)
log.trace("Zero phase commit " + this + ": No resources.");
status = Status.STATUS_COMMITTED;
break;
case 1:
if (trace)
log.trace("One phase commit " + this + ": One resource.");
commitResources(true);
break;
default:
if (trace)
log.trace("Two phase commit " + this + ": Many resources.");
if (!prepareResources())
{
boolean commitDecision =
status == Status.STATUS_PREPARED &&
(heuristicCode == HEUR_NONE ||
heuristicCode == XAException.XA_HEURCOM);
if (commitDecision)
commitResources(false);
}
else
status = Status.STATUS_COMMITTED; }
}
if (status != Status.STATUS_COMMITTED)
{
Throwable causedByThrowable = cause;
rollbackResources();
completeTransaction();
throw new JBossRollbackException("Unable to commit, tx=" +
toString() + " status=" + getStringStatus(status),
causedByThrowable);
}
completeTransaction();
checkHeuristics();
if (trace)
log.trace("Committed OK, tx=" + this);
}
finally
{
unlock();
}
}
public void rollback()
throws java.lang.IllegalStateException,
java.lang.SecurityException,
SystemException
{
lock();
try
{
if (trace)
log.trace("rollback(): Entered, tx=" + toString() +
" status=" + getStringStatus(status));
checkWork();
switch (status)
{
case Status.STATUS_ACTIVE:
status = Status.STATUS_MARKED_ROLLBACK;
case Status.STATUS_MARKED_ROLLBACK:
endResources();
rollbackResources();
completeTransaction();
heuristicCode = HEUR_NONE;
return;
case Status.STATUS_PREPARING:
status = Status.STATUS_MARKED_ROLLBACK;
return; default:
throw new IllegalStateException("Cannot rollback(), " +
"tx=" + toString() +
" status=" +
getStringStatus(status));
}
}
finally
{
Thread.interrupted(); unlock();
}
}
public boolean delistResource(XAResource xaRes, int flag)
throws java.lang.IllegalStateException,
SystemException
{
if (xaRes == null)
throw new IllegalArgumentException("null xaRes tx=" + this);
if (flag != XAResource.TMSUCCESS &&
flag != XAResource.TMSUSPEND &&
flag != XAResource.TMFAIL)
throw new IllegalArgumentException("Bad flag: " + flag + " tx=" + this);
lock();
try
{
if (trace)
log.trace("delistResource(): Entered, tx=" +
toString() + " status=" + getStringStatus(status));
Resource resource = findResource(xaRes);
if (resource == null)
throw new IllegalArgumentException("xaRes not enlisted " + xaRes);
switch (status)
{
case Status.STATUS_ACTIVE:
case Status.STATUS_MARKED_ROLLBACK:
break;
case Status.STATUS_PREPARING:
throw new IllegalStateException("Already started preparing. " + this);
case Status.STATUS_ROLLING_BACK:
throw new IllegalStateException("Already started rolling back. " + this);
case Status.STATUS_PREPARED:
throw new IllegalStateException("Already prepared. " + this);
case Status.STATUS_COMMITTING:
throw new IllegalStateException("Already started committing. " + this);
case Status.STATUS_COMMITTED:
throw new IllegalStateException("Already committed. " + this);
case Status.STATUS_ROLLEDBACK:
throw new IllegalStateException("Already rolled back. " + this);
case Status.STATUS_NO_TRANSACTION:
throw new IllegalStateException("No transaction. " + this);
case Status.STATUS_UNKNOWN:
throw new IllegalStateException("Unknown state " + this);
default:
throw new IllegalStateException("Illegal status: " + getStringStatus(status) + " tx=" + this);
}
try
{
return resource.delistResource(xaRes, flag);
}
catch (XAException xae)
{
logXAException(xae);
status = Status.STATUS_MARKED_ROLLBACK;
cause = xae;
return false;
}
}
finally
{
unlock();
}
}
public boolean enlistResource(XAResource xaRes)
throws RollbackException,
java.lang.IllegalStateException,
SystemException
{
if (xaRes == null)
throw new IllegalArgumentException("null xaRes tx=" + this);
lock();
try
{
if (trace)
log.trace("enlistResource(): Entered, tx=" +
toString() + " status=" + getStringStatus(status));
switch (status)
{
case Status.STATUS_ACTIVE:
case Status.STATUS_PREPARING:
break;
case Status.STATUS_PREPARED:
throw new IllegalStateException("Already prepared. " + this);
case Status.STATUS_COMMITTING:
throw new IllegalStateException("Already started committing. " + this);
case Status.STATUS_COMMITTED:
throw new IllegalStateException("Already committed. " + this);
case Status.STATUS_MARKED_ROLLBACK:
throw new RollbackException("Already marked for rollback " + this);
case Status.STATUS_ROLLING_BACK:
throw new RollbackException("Already started rolling back. " + this);
case Status.STATUS_ROLLEDBACK:
throw new RollbackException("Already rolled back. " + this);
case Status.STATUS_NO_TRANSACTION:
throw new IllegalStateException("No transaction. " + this);
case Status.STATUS_UNKNOWN:
throw new IllegalStateException("Unknown state " + this);
default:
throw new IllegalStateException("Illegal status: " + getStringStatus(status) + " tx=" + this);
}
if (resourcesEnded)
throw new IllegalStateException("Too late to enlist resources " + this);
try
{
Resource resource = findResource(xaRes);
if (resource != null)
{
if (resource.isEnlisted())
return false; if (resource.isDelisted(xaRes))
resource = null;
else
return resource.startResource();
}
resource = findResourceManager(xaRes);
if (resource != null)
{
resource = addResource(xaRes, resource.getXid(), resource);
return resource.startResource();
}
resource = addResource(xaRes, createXidBranch(), null);
return resource.startResource();
}
catch (XAException xae)
{
logXAException(xae);
cause = xae;
return false;
}
}
finally
{
unlock();
}
}
public int getStatus()
throws SystemException
{
if (done)
return Status.STATUS_NO_TRANSACTION;
return status;
}
public void registerSynchronization(Synchronization s)
throws RollbackException,
java.lang.IllegalStateException,
SystemException
{
if (s == null)
throw new IllegalArgumentException("Null synchronization " + this);
lock();
try
{
if (trace)
{
log.trace("registerSynchronization(): Entered, " +
"tx=" + toString() +
" status=" + getStringStatus(status));
}
switch (status)
{
case Status.STATUS_ACTIVE:
case Status.STATUS_PREPARING:
break;
case Status.STATUS_PREPARED:
throw new IllegalStateException("Already prepared. " + this);
case Status.STATUS_COMMITTING:
throw new IllegalStateException("Already started committing. " + this);
case Status.STATUS_COMMITTED:
throw new IllegalStateException("Already committed. " + this);
case Status.STATUS_MARKED_ROLLBACK:
throw new RollbackException("Already marked for rollback " + this);
case Status.STATUS_ROLLING_BACK:
throw new RollbackException("Already started rolling back. " + this);
case Status.STATUS_ROLLEDBACK:
throw new RollbackException("Already rolled back. " + this);
case Status.STATUS_NO_TRANSACTION:
throw new IllegalStateException("No transaction. " + this);
case Status.STATUS_UNKNOWN:
throw new IllegalStateException("Unknown state " + this);
default:
throw new IllegalStateException("Illegal status: " + getStringStatus(status) + " tx=" + this);
}
if (syncCount == syncAllocSize)
{
syncAllocSize = 2 * syncAllocSize;
Synchronization[] sy = new Synchronization[syncAllocSize];
System.arraycopy(sync, 0, sy, 0, syncCount);
sync = sy;
}
sync[syncCount++] = s;
}
finally
{
unlock();
}
}
public void setRollbackOnly()
throws java.lang.IllegalStateException,
SystemException
{
lock();
try
{
if (trace)
log.trace("setRollbackOnly(): Entered, tx=" +
toString() + " status=" + getStringStatus(status));
switch (status)
{
case Status.STATUS_ACTIVE:
case Status.STATUS_PREPARING:
case Status.STATUS_PREPARED:
status = Status.STATUS_MARKED_ROLLBACK;
case Status.STATUS_MARKED_ROLLBACK:
case Status.STATUS_ROLLING_BACK:
return;
case Status.STATUS_COMMITTING:
throw new IllegalStateException("Already started committing. " + this);
case Status.STATUS_COMMITTED:
throw new IllegalStateException("Already committed. " + this);
case Status.STATUS_ROLLEDBACK:
throw new IllegalStateException("Already rolled back. " + this);
case Status.STATUS_NO_TRANSACTION:
throw new IllegalStateException("No transaction. " + this);
case Status.STATUS_UNKNOWN:
throw new IllegalStateException("Unknown state " + this);
default:
throw new IllegalStateException("Illegal status: " + getStringStatus(status) + " tx=" + this);
}
}
finally
{
unlock();
}
}
public void associateCurrentThread()
{
Thread.interrupted();
lock();
try
{
threads.add(Thread.currentThread());
}
finally
{
unlock();
}
}
public void disassociateCurrentThread()
{
if (done)
{
threads.remove(Thread.currentThread());
}
else
{
lock();
try
{
threads.remove(Thread.currentThread());
}
finally
{
unlock();
}
}
Thread.interrupted();
}
public int hashCode()
{
return xid.hashCode();
}
public String toString()
{
return "TransactionImpl:" + xidFactory.toString(xid);
}
public boolean equals(Object obj)
{
if (obj != null && obj instanceof TransactionImpl)
return getLocalIdValue() == (((TransactionImpl)obj).getLocalIdValue());
return false;
}
public long getLocalIdValue()
{
return xid.getLocalIdValue();
}
public LocalId getLocalId()
{
return xid.getLocalId();
}
public GlobalId getGlobalId()
{
return gid;
}
public XidImpl getXid()
{
return xid;
}
synchronized void lock()
{
if (done)
throw new IllegalStateException("Transaction has terminated " + this);
if (locked)
{
log.warn("Lock contention, tx=" + toString());
while (locked)
{
try
{
wait();
}
catch (InterruptedException ex)
{
}
if (done)
throw new IllegalStateException("Transaction has now terminated " + this);
}
}
locked = true;
}
synchronized void unlock()
{
if (!locked)
{
log.warn("Unlocking, but not locked, tx=" + toString(),
new Throwable("[Stack trace]"));
}
locked = false;
notify();
}
int prepare() throws HeuristicMixedException, HeuristicRollbackException, RollbackException
{
lock();
try
{
if (trace)
log.trace("Preparing, tx=" + this + ", status=" + getStringStatus(status));
checkWork();
beforePrepare();
if (status == Status.STATUS_ACTIVE)
{
switch (getCommitStrategy())
{
case 0:
{
if (trace)
log.trace("Prepare tx=" + this + ": No resources.");
status = Status.STATUS_COMMITTED;
completeTransaction();
return XAResource.XA_RDONLY;
}
default:
{
if (trace)
log.trace("Prepare tx=" + this + ": Many resources.");
if (!prepareResources())
{
boolean commitDecision =
status == Status.STATUS_PREPARED &&
(heuristicCode == HEUR_NONE ||
heuristicCode == XAException.XA_HEURCOM);
}
else
{
if (trace)
log.trace("Prepared tx=" + this + ": All readonly.");
status = Status.STATUS_COMMITTED;
completeTransaction();
return XAResource.XA_RDONLY;
}
}
}
}
if (status != Status.STATUS_PREPARED)
{
Throwable causedByThrowable = cause;
rollbackResources();
completeTransaction();
throw new JBossRollbackException("Unable to prepare, tx=" +
toString() + " status=" + getStringStatus(status),
causedByThrowable);
}
return XAResource.XA_OK;
}
finally
{
unlock();
}
}
void commit(boolean onePhase) throws RollbackException, HeuristicMixedException, HeuristicRollbackException, SystemException
{
checkWork();
if (onePhase)
{
commit();
return;
}
lock();
try
{
if (trace)
log.trace("Committing two phase, tx=" + this + ", status=" + getStringStatus(status));
switch (status)
{
case Status.STATUS_PREPARING:
throw new IllegalStateException("Still preparing. " + this);
case Status.STATUS_ROLLING_BACK:
throw new IllegalStateException("Already started rolling back. " + this);
case Status.STATUS_ROLLEDBACK:
instanceDone();
checkHeuristics();
throw new IllegalStateException("Already rolled back. " + this);
case Status.STATUS_COMMITTING:
throw new IllegalStateException("Already started committing. " + this);
case Status.STATUS_COMMITTED:
instanceDone();
checkHeuristics();
throw new IllegalStateException("Already committed. " + this);
case Status.STATUS_NO_TRANSACTION:
throw new IllegalStateException("No transaction. " + this);
case Status.STATUS_UNKNOWN:
throw new IllegalStateException("Unknown state " + this);
case Status.STATUS_MARKED_ROLLBACK:
endResources();
rollbackResources();
completeTransaction();
checkHeuristics();
throw new RollbackException("Already marked for rollback " + this);
case Status.STATUS_PREPARED:
break;
default:
throw new IllegalStateException("Illegal status: " + getStringStatus(status) + " tx=" + this);
}
commitResources(false);
if (status != Status.STATUS_COMMITTED)
{
Throwable causedByThrowable = cause;
rollbackResources();
completeTransaction();
throw new JBossRollbackException("Unable to commit, tx=" +
toString() + " status=" + getStringStatus(status),
causedByThrowable);
}
completeTransaction();
checkHeuristics();
if (trace)
log.trace("Committed OK, tx=" + this);
}
finally
{
unlock();
}
}
Work getWork()
{
return work;
}
void setWork(Work work) throws WorkCompletedException
{
lock();
try
{
if (work == null)
{
this.work = null;
return;
}
if (status == Status.STATUS_NO_TRANSACTION || status == Status.STATUS_UNKNOWN)
throw new WorkCompletedException("The transaction is not active " + this + ": " + getStringStatus(status), WorkException.TX_RECREATE_FAILED);
else if (status != Status.STATUS_ACTIVE)
throw new WorkCompletedException("Too late to start work " + this + ": " + getStringStatus(status), WorkException.TX_CONCURRENT_WORK_DISALLOWED);
else if (this.work != null)
throw new WorkCompletedException("Already have work " + this + ": " + this.work, WorkException.TX_CONCURRENT_WORK_DISALLOWED);
this.work = work;
}
finally
{
unlock();
}
}
boolean isDone()
{
return done;
}
private void beforePrepare() throws HeuristicMixedException, HeuristicRollbackException, RollbackException
{
switch (status)
{
case Status.STATUS_PREPARING:
throw new IllegalStateException("Already started preparing. " + this);
case Status.STATUS_PREPARED:
throw new IllegalStateException("Already prepared. " + this);
case Status.STATUS_ROLLING_BACK:
throw new IllegalStateException("Already started rolling back. " + this);
case Status.STATUS_ROLLEDBACK:
instanceDone();
checkHeuristics();
throw new IllegalStateException("Already rolled back." + this);
case Status.STATUS_COMMITTING:
throw new IllegalStateException("Already started committing. " + this);
case Status.STATUS_COMMITTED:
instanceDone();
checkHeuristics();
throw new IllegalStateException("Already committed. " + this);
case Status.STATUS_NO_TRANSACTION:
throw new IllegalStateException("No transaction. " + this);
case Status.STATUS_UNKNOWN:
throw new IllegalStateException("Unknown state " + this);
case Status.STATUS_MARKED_ROLLBACK:
endResources();
rollbackResources();
completeTransaction();
checkHeuristics();
throw new RollbackException("Already marked for rollback " + this);
case Status.STATUS_ACTIVE:
break;
default:
throw new IllegalStateException("Illegal status: " + getStringStatus(status) + " tx=" + this);
}
doBeforeCompletion();
if (trace)
log.trace("Before completion done, tx=" + this +
", status=" + getStringStatus(status));
endResources();
}
private void completeTransaction()
{
cancelTimeout();
doAfterCompletion();
instanceDone();
}
private void interruptThreads()
{
TxManager manager = TxManager.getInstance();
if (manager.isInterruptThreads())
{
HashSet clone = (HashSet) threads.clone();
threads.clear();
for (Iterator i = clone.iterator(); i.hasNext();)
{
Thread thread = (Thread) i.next();
try
{
thread.interrupt();
}
catch (Throwable ignored)
{
if (trace)
log.trace("Ignored error interrupting thread: " + thread, ignored);
}
}
}
}
private String getStringStatus(int status)
{
switch (status)
{
case Status.STATUS_PREPARING:
return "STATUS_PREPARING";
case Status.STATUS_PREPARED:
return "STATUS_PREPARED";
case Status.STATUS_ROLLING_BACK:
return "STATUS_ROLLING_BACK";
case Status.STATUS_ROLLEDBACK:
return "STATUS_ROLLEDBACK";
case Status.STATUS_COMMITTING:
return "STATUS_COMMITING";
case Status.STATUS_COMMITTED:
return "STATUS_COMMITED";
case Status.STATUS_NO_TRANSACTION:
return "STATUS_NO_TRANSACTION";
case Status.STATUS_UNKNOWN:
return "STATUS_UNKNOWN";
case Status.STATUS_MARKED_ROLLBACK:
return "STATUS_MARKED_ROLLBACK";
case Status.STATUS_ACTIVE:
return "STATUS_ACTIVE";
default:
return "STATUS_UNKNOWN(" + status + ")";
}
}
private String getStringXAErrorCode(int errorCode)
{
switch (errorCode)
{
case XAException.XA_HEURCOM:
return "XA_HEURCOM";
case XAException.XA_HEURHAZ:
return "XA_HEURHAZ";
case XAException.XA_HEURMIX:
return "XA_HEURMIX";
case XAException.XA_HEURRB:
return "XA_HEURRB";
case XAException.XA_NOMIGRATE:
return "XA_NOMIGRATE";
case XAException.XA_RBCOMMFAIL:
return "XA_RBCOMMFAIL";
case XAException.XA_RBDEADLOCK:
return "XA_RBDEADLOCK";
case XAException.XA_RBINTEGRITY:
return "XA_RBINTEGRITY";
case XAException.XA_RBOTHER:
return "XA_RBOTHER";
case XAException.XA_RBPROTO:
return "XA_RBPROTO";
case XAException.XA_RBROLLBACK:
return "XA_RBROLLBACK";
case XAException.XA_RBTIMEOUT:
return "XA_RBTIMEOUT";
case XAException.XA_RBTRANSIENT:
return "XA_RBTRANSIENT";
case XAException.XA_RDONLY:
return "XA_RDONLY";
case XAException.XA_RETRY:
return "XA_RETRY";
case XAException.XAER_ASYNC:
return "XAER_ASYNC";
case XAException.XAER_DUPID:
return "XAER_DUPID";
case XAException.XAER_INVAL:
return "XAER_INVAL";
case XAException.XAER_NOTA:
return "XAER_NOTA";
case XAException.XAER_OUTSIDE:
return "XAER_OUTSIDE";
case XAException.XAER_PROTO:
return "XAER_PROTO";
case XAException.XAER_RMERR:
return "XAER_RMERR";
case XAException.XAER_RMFAIL:
return "XAER_RMFAIL";
default:
return "XA_UNKNOWN(" + errorCode + ")";
}
}
private void logXAException(XAException xae)
{
log.warn("XAException: tx=" + toString() + " errorCode=" +
getStringXAErrorCode(xae.errorCode), xae);
if (txManagerService != null)
txManagerService.formatXAException(xae, log);
}
private synchronized void instanceDone()
{
TxManager manager = TxManager.getInstance();
if (status == Status.STATUS_COMMITTED)
manager.incCommitCount();
else
manager.incRollbackCount();
sync = null;
resources = null;
transactionLocalMap.clear();
threads.clear();
manager.releaseTransactionImpl(this);
status = Status.STATUS_NO_TRANSACTION;
notifyAll();
done = true;
}
private void cancelTimeout()
{
if (timeout != null)
{
unlock();
try
{
timeout.cancel();
}
catch (Exception e)
{
if (trace)
log.trace("failed to cancel timeout " + this, e);
}
finally
{
lock();
}
timeout = null;
}
}
private Resource findResource(XAResource xaRes)
{
for (int idx = resources.size() - 1; idx >= 0; --idx)
{
Resource resource = (Resource) resources.get(idx);
if (xaRes == resource.getXAResource())
return resource;
}
return null;
}
private Resource findResourceManager(XAResource xaRes) throws XAException
{
for (int i = 0; i < resources.size(); ++i)
{
Resource resource = (Resource) resources.get(i);
if (resource.isResourceManager(xaRes))
return resource;
}
return null;
}
private Resource addResource(XAResource xaRes, Xid branchXid, Resource sameRMResource)
{
Resource resource = new Resource(xaRes, branchXid, sameRMResource);
resources.add(resource);
if (lastResource == null && xaRes instanceof LastResource)
lastResource = resource;
return resource;
}
private void endResources()
{
for (int idx = 0; idx < resources.size(); ++idx)
{
Resource resource = (Resource) resources.get(idx);
try
{
resource.endResource();
}
catch(XAException xae)
{
logXAException(xae);
status = Status.STATUS_MARKED_ROLLBACK;
cause = xae;
}
}
resourcesEnded = true; }
private void doBeforeCompletion()
{
unlock();
try
{
for (int i = 0; i < syncCount; i++)
{
try
{
if (trace)
log.trace("calling sync " + i + ", " + sync[i] + " tx=" + this);
sync[i].beforeCompletion();
}
catch (Throwable t)
{
if (trace)
log.trace("failed before completion " + sync[i], t);
status = Status.STATUS_MARKED_ROLLBACK;
cause = t;
break;
}
}
}
finally
{
lock();
}
}
private void doAfterCompletion()
{
unlock();
try
{
for (int i = 0; i < syncCount; i++)
{
try
{
sync[i].afterCompletion(status);
}
catch (Throwable t)
{
if (trace)
log.trace("failed after completion " + sync[i], t);
}
}
}
finally
{
lock();
}
}
private void gotHeuristic(Resource resource, int code)
{
switch (code)
{
case XAException.XA_HEURMIX:
heuristicCode = XAException.XA_HEURMIX;
break;
case XAException.XA_HEURRB:
if (heuristicCode == HEUR_NONE)
heuristicCode = XAException.XA_HEURRB;
else if (heuristicCode == XAException.XA_HEURCOM ||
heuristicCode == XAException.XA_HEURHAZ)
heuristicCode = XAException.XA_HEURMIX;
break;
case XAException.XA_HEURCOM:
if (heuristicCode == HEUR_NONE)
heuristicCode = XAException.XA_HEURCOM;
else if (heuristicCode == XAException.XA_HEURRB ||
heuristicCode == XAException.XA_HEURHAZ)
heuristicCode = XAException.XA_HEURMIX;
break;
case XAException.XA_HEURHAZ:
if (heuristicCode == HEUR_NONE)
heuristicCode = XAException.XA_HEURHAZ;
else if (heuristicCode == XAException.XA_HEURCOM ||
heuristicCode == XAException.XA_HEURRB)
heuristicCode = XAException.XA_HEURMIX;
break;
default:
throw new IllegalArgumentException();
}
if (resource != null)
resource.forget();
}
private void checkHeuristics()
throws HeuristicMixedException, HeuristicRollbackException
{
switch (heuristicCode)
{
case XAException.XA_HEURHAZ:
case XAException.XA_HEURMIX:
heuristicCode = HEUR_NONE;
if (trace)
log.trace("Throwing HeuristicMixedException, tx=" + this +
"status=" + getStringStatus(status));
throw new HeuristicMixedException();
case XAException.XA_HEURRB:
heuristicCode = HEUR_NONE;
if (trace)
log.trace("Throwing HeuristicRollbackException, tx=" + this +
"status=" + getStringStatus(status));
throw new HeuristicRollbackException();
case XAException.XA_HEURCOM:
heuristicCode = HEUR_NONE;
if (trace)
log.trace("NOT Throwing HeuristicCommitException, tx=" + this +
"status=" + getStringStatus(status));
return;
}
}
private boolean prepareResources()
{
boolean readOnly = true;
status = Status.STATUS_PREPARING;
for (int i = 0; i < resources.size(); ++i)
{
if (status != Status.STATUS_PREPARING)
return false;
Resource resource = (Resource) resources.get(i);
if (resource.isResourceManager() == false)
continue;
if (resource == lastResource)
continue;
try
{
int vote = resource.prepare();
if (vote == RS_VOTE_OK)
readOnly = false;
else if (vote != RS_VOTE_READONLY)
{
if (trace)
log.trace("illegal vote in prepare resources tx=" + this + " resource=" + resource, new Exception());
status = Status.STATUS_MARKED_ROLLBACK;
return false;
}
}
catch (XAException e)
{
readOnly = false;
logXAException(e);
switch (e.errorCode)
{
case XAException.XA_HEURCOM:
gotHeuristic(resource, e.errorCode);
break;
case XAException.XA_HEURRB:
case XAException.XA_HEURMIX:
case XAException.XA_HEURHAZ:
gotHeuristic(resource, e.errorCode);
if (status == Status.STATUS_PREPARING)
status = Status.STATUS_MARKED_ROLLBACK;
break;
default:
cause = e;
if (status == Status.STATUS_PREPARING)
status = Status.STATUS_MARKED_ROLLBACK;
break;
}
}
catch (Throwable t)
{
if (trace)
log.trace("unhandled throwable in prepareResources " + this, t);
if (status == Status.STATUS_PREPARING)
status = Status.STATUS_MARKED_ROLLBACK;
cause = t;
}
}
if (status != Status.STATUS_PREPARING)
return false;
if (lastResource != null)
{
try
{
lastResource.prepareLastResource();
lastResource.commit(false);
}
catch (XAException e)
{
logXAException(e);
switch (e.errorCode)
{
case XAException.XA_HEURRB:
case XAException.XA_HEURCOM:
case XAException.XA_HEURMIX:
case XAException.XA_HEURHAZ:
gotHeuristic(lastResource, e.errorCode);
if (status == Status.STATUS_PREPARING)
status = Status.STATUS_MARKED_ROLLBACK;
break;
default:
cause = e;
if (status == Status.STATUS_PREPARING)
status = Status.STATUS_MARKED_ROLLBACK;
break;
}
}
catch (Throwable t)
{
if (trace)
log.trace("unhandled throwable in prepareResources " + this, t);
if (status == Status.STATUS_PREPARING)
status = Status.STATUS_MARKED_ROLLBACK;
cause = t;
}
}
if (status == Status.STATUS_PREPARING)
status = Status.STATUS_PREPARED;
return readOnly;
}
private void commitResources(boolean onePhase)
{
status = Status.STATUS_COMMITTING;
for (int i = 0; i < resources.size(); ++i)
{
if (status != Status.STATUS_COMMITTING)
return;
Resource resource = (Resource) resources.get(i);
if (onePhase == false && lastResource == resource)
continue;
try
{
resource.commit(onePhase);
}
catch (XAException e)
{
logXAException(e);
switch (e.errorCode) {
case XAException.XA_HEURRB:
case XAException.XA_HEURCOM:
case XAException.XA_HEURMIX:
case XAException.XA_HEURHAZ:
gotHeuristic(resource, e.errorCode);
if (onePhase)
status = Status.STATUS_MARKED_ROLLBACK;
break;
default:
cause = e;
if (onePhase)
{
status = Status.STATUS_MARKED_ROLLBACK;
break;
}
}
}
catch (Throwable t)
{
if (trace)
log.trace("unhandled throwable in commitResources " + this, t);
}
}
if (status == Status.STATUS_COMMITTING)
status = Status.STATUS_COMMITTED;
}
private void rollbackResources()
{
status = Status.STATUS_ROLLING_BACK;
for (int i = 0; i < resources.size(); ++i)
{
Resource resource = (Resource) resources.get(i);
try
{
resource.rollback();
}
catch (XAException e)
{
logXAException(e);
switch (e.errorCode)
{
case XAException.XA_HEURRB:
gotHeuristic(resource, e.errorCode);
continue;
case XAException.XA_HEURCOM:
case XAException.XA_HEURMIX:
case XAException.XA_HEURHAZ:
gotHeuristic(resource, e.errorCode);
continue;
default:
cause = e;
break;
}
}
catch (Throwable t)
{
if (trace)
log.trace("unhandled throwable in rollbackResources " + this, t);
}
}
status = Status.STATUS_ROLLEDBACK;
}
private Xid createXidBranch()
{
long branchId = ++lastBranchId;
return xidFactory.newBranch(xid, branchId);
}
private int getCommitStrategy()
{
int resourceCount = resources.size();
if (resourceCount == 0)
return 0;
if (resourceCount == 1)
return 1;
for (int i = 1; i < resourceCount; ++i)
{
Resource resource = (Resource) resources.get(i);
if (resource.isResourceManager())
{
return 2;
}
}
return 1;
}
public long getTimeLeftBeforeTimeout()
{
return (start + timeoutPeriod) - System.currentTimeMillis();
}
Object getTransactionLocalValue(TransactionLocal tlocal)
{
return transactionLocalMap.get(tlocal);
}
void putTransactionLocalValue(TransactionLocal tlocal, Object value)
{
transactionLocalMap.put(tlocal, value);
}
boolean containsTransactionLocal(TransactionLocal tlocal)
{
return transactionLocalMap.containsKey(tlocal);
}
private void checkWork()
{
if (work != null)
throw new IllegalStateException("Work still outstanding " + work + " tx=" + this);
}
private class Resource
{
private XAResource xaResource;
private int resourceState;
private Resource resourceSameRM;
private Xid resourceXid;
public Resource(XAResource xaResource, Xid resourceXid, Resource resourceSameRM)
{
this.xaResource = xaResource;
this.resourceXid = resourceXid;
this.resourceSameRM = resourceSameRM;
resourceState = RS_NEW;
}
public XAResource getXAResource()
{
return xaResource;
}
public Xid getXid()
{
return resourceXid;
}
public boolean isEnlisted()
{
return resourceState == RS_ENLISTED;
}
public boolean isResourceManager()
{
return resourceSameRM == null;
}
public boolean isResourceManager(XAResource xaRes) throws XAException
{
return resourceSameRM == null && xaRes.isSameRM(xaResource);
}
public boolean isDelisted(XAResource xaRes) throws XAException
{
return resourceState == RS_ENDED && xaResource.isSameRM(xaRes) == false;
}
public boolean startResource()
throws XAException
{
int flags = XAResource.TMJOIN;
if (resourceSameRM == null)
{
switch (resourceState)
{
case RS_NEW:
flags = XAResource.TMNOFLAGS;
break;
case RS_SUSPENDED:
flags = XAResource.TMRESUME;
break;
default:
if (trace)
log.trace("Unhandled resource state: " + resourceState +
" (not RS_NEW or RS_SUSPENDED, using TMJOIN flags)");
}
}
if (trace)
log.trace("startResource(" +
xidFactory.toString(resourceXid) +
") entered: " + xaResource.toString() +
" flags=" + flags);
unlock();
try
{
try
{
xaResource.start(resourceXid, flags);
}
catch(XAException e)
{
throw e;
}
catch (Throwable t)
{
if (trace)
log.trace("unhandled throwable error in startResource", t);
status = Status.STATUS_MARKED_ROLLBACK;
return false;
}
resourceState = RS_ENLISTED;
}
finally
{
lock();
if (trace)
log.trace("startResource(" +
xidFactory.toString(resourceXid) +
") leaving: " + xaResource.toString() +
" flags=" + flags);
}
return true;
}
public boolean delistResource(XAResource xaRes, int flag) throws XAException
{
if (isDelisted(xaRes))
{
log.warn("Resource already delisted. tx=" + this.toString());
return false;
}
endResource(flag);
return true;
}
public void endResource() throws XAException
{
if (resourceState == RS_ENLISTED || resourceState == RS_SUSPENDED)
{
if (trace)
log.trace("endresources(" + xaResource + "): state=" +
resourceState);
endResource(XAResource.TMSUCCESS);
}
}
private void endResource(int flag) throws XAException
{
if (trace)
log.trace("endResource(" +
xidFactory.toString(resourceXid) +
") entered: " + xaResource.toString() +
" flag=" + flag);
unlock();
try
{
try
{
xaResource.end(resourceXid, flag);
}
catch(XAException e)
{
throw e;
}
catch (Throwable t)
{
if (trace)
log.trace("unhandled throwable error in endResource", t);
status = Status.STATUS_MARKED_ROLLBACK;
resourceState = RS_ENDED;
return;
}
if (flag == XAResource.TMSUSPEND)
resourceState = RS_SUSPENDED;
else
{
if (flag == XAResource.TMFAIL)
status = Status.STATUS_MARKED_ROLLBACK;
resourceState = RS_ENDED;
}
}
finally
{
lock();
if (trace)
log.trace("endResource(" +
xidFactory.toString(resourceXid) +
") leaving: " + xaResource.toString() +
" flag=" + flag);
}
}
public void forget()
{
unlock();
try
{
xaResource.forget(resourceXid);
}
catch (XAException xae)
{
logXAException(xae);
cause = xae;
}
finally
{
lock();
}
resourceState = RS_FORGOT;
}
public int prepare() throws XAException
{
int vote;
unlock();
try
{
vote = xaResource.prepare(resourceXid);
}
finally
{
lock();
}
if (vote == XAResource.XA_OK)
resourceState = RS_VOTE_OK;
else if (vote == XAResource.XA_RDONLY)
resourceState = RS_VOTE_READONLY;
return resourceState;
}
public void prepareLastResource() throws XAException
{
resourceState = RS_VOTE_OK;
}
public void commit(boolean onePhase) throws XAException
{
if (trace)
log.trace("Committing resource " + xaResource + " state=" +
resourceState);
if (!onePhase && resourceState != RS_VOTE_OK)
return;
if (resourceSameRM != null)
return;
unlock();
try
{
xaResource.commit(resourceXid, onePhase);
}
finally
{
lock();
}
}
public void rollback() throws XAException
{
if (resourceState == RS_VOTE_READONLY)
return;
if (resourceState == RS_FORGOT)
return;
if (resourceSameRM != null)
return;
unlock();
try
{
xaResource.rollback(resourceXid);
}
finally
{
lock();
}
}
}
}