From 439607e59fcb956b4e33cad1419d8f28dca43627 Mon Sep 17 00:00:00 2001 From: wenshao Date: Thu, 12 May 2011 03:19:16 +0000 Subject: [PATCH] =?UTF-8?q?=E6=94=AF=E6=8C=81XA?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit git-svn-id: http://code.alibabatech.com/svn/druid/trunk@26 b9813039-fb51-4c41-a8b9-e21c2acb5095 --- pom.xml | 6 + .../druid/pool/xa/TransactionContext.java | 132 ++++++++++++++++++ .../pool/xa/TransactionContextListener.java | 19 +++ .../druid/pool/xa/TransactionRegistry.java | 109 +++++++++++++++ .../druid/pool/xa/XAConnectionFactory.java | 41 ++++++ 5 files changed, 307 insertions(+) create mode 100644 src/main/java/com/alibaba/druid/pool/xa/TransactionContext.java create mode 100644 src/main/java/com/alibaba/druid/pool/xa/TransactionContextListener.java create mode 100644 src/main/java/com/alibaba/druid/pool/xa/TransactionRegistry.java create mode 100644 src/main/java/com/alibaba/druid/pool/xa/XAConnectionFactory.java diff --git a/pom.xml b/pom.xml index 468b9d3c4b..15361843ee 100644 --- a/pom.xml +++ b/pom.xml @@ -50,6 +50,12 @@ 2.5 provided + + jta + jta + 1.0.1b + provided + commons-logging commons-logging diff --git a/src/main/java/com/alibaba/druid/pool/xa/TransactionContext.java b/src/main/java/com/alibaba/druid/pool/xa/TransactionContext.java new file mode 100644 index 0000000000..c7e4510e53 --- /dev/null +++ b/src/main/java/com/alibaba/druid/pool/xa/TransactionContext.java @@ -0,0 +1,132 @@ +package com.alibaba.druid.pool.xa; + +import javax.transaction.RollbackException; +import javax.transaction.Status; +import javax.transaction.Synchronization; +import javax.transaction.SystemException; +import javax.transaction.Transaction; +import javax.transaction.xa.XAResource; +import java.sql.Connection; +import java.sql.SQLException; +import java.lang.ref.WeakReference; + +/** + * TransactionContext represents the association between a single XAConnectionFactory and a Transaction. + * This context contains a single shared connection which should be used by all ManagedConnections for + * the XAConnectionFactory, the ability to listen for the transaction completion event, and a method + * to check the status of the transaction. + * + * @author Dain Sundstrom + * @version $Revision$ + */ +public class TransactionContext { + private final TransactionRegistry transactionRegistry; + private final WeakReference transactionRef; + private Connection sharedConnection; + + /** + * Creates a TransactionContext for the specified Transaction and TransactionRegistry. The + * TransactionRegistry is used to obtain the XAResource for the shared connection when it is + * enlisted in the transaction. + * + * @param transactionRegistry the TransactionRegistry used to obtain the XAResource for the + * shared connection + * @param transaction the transaction + */ + public TransactionContext(TransactionRegistry transactionRegistry, Transaction transaction) { + if (transactionRegistry == null) throw new NullPointerException("transactionRegistry is null"); + if (transaction == null) throw new NullPointerException("transaction is null"); + this.transactionRegistry = transactionRegistry; + this.transactionRef = new WeakReference(transaction); + } + + /** + * Gets the connection shared by all ManagedConnections in the transaction. Specifically, + * connection using the same XAConnectionFactory from which the TransactionRegistry was + * obtained. + * @return the shared connection for this transaction + */ + public Connection getSharedConnection() { + return sharedConnection; + } + + /** + * Sets the shared connection for this transaction. The shared connection is enlisted + * in the transaction. + * + * @param sharedConnection the shared connection + * @throws SQLException if a shared connection is already set, if XAResource for the connection + * could not be found in the transaction registry, or if there was a problem enlisting the + * connection in the transaction + */ + public void setSharedConnection(Connection sharedConnection) throws SQLException { + if (this.sharedConnection != null) { + throw new IllegalStateException("A shared connection is alredy set"); + } + + // This is the first use of the connection in this transaction, so we must + // enlist it in the transaction + Transaction transaction = getTransaction(); + try { + XAResource xaResource = transactionRegistry.getXAResource(sharedConnection); + transaction.enlistResource(xaResource); + } catch (RollbackException e) { + // transaction was rolled back... proceed as if there never was a transaction + } catch (SystemException e) { + throw (SQLException) new SQLException("Unable to enlist connection the transaction").initCause(e); + } + + this.sharedConnection = sharedConnection; + } + + /** + * Adds a listener for transaction completion events. + * + * @param listener the listener to add + * @throws SQLException if a problem occurs adding the listener to the transaction + */ + public void addTransactionContextListener(final TransactionContextListener listener) throws SQLException { + try { + getTransaction().registerSynchronization(new Synchronization() { + public void beforeCompletion() { + } + + public void afterCompletion(int status) { + listener.afterCompletion(TransactionContext.this, status == Status.STATUS_COMMITTED); + } + }); + } catch (RollbackException e) { + // JTA spec doesn't let us register with a transaction marked rollback only + // just ignore this and the tx state will be cleared another way. + } catch (Exception e) { + throw (SQLException) new SQLException("Unable to register transaction context listener").initCause(e); + } + } + + /** + * True if the transaction is active or marked for rollback only. + * @return true if the transaction is active or marked for rollback only; false otherwise + * @throws SQLException if a problem occurs obtaining the transaction status + */ + public boolean isActive() throws SQLException { + try { + Transaction transaction = (Transaction) this.transactionRef.get(); + if (transaction == null) { + return false; + } + int status = transaction.getStatus(); + return status == Status.STATUS_ACTIVE || status == Status.STATUS_MARKED_ROLLBACK; + } catch (SystemException e) { + throw (SQLException) new SQLException("Unable to get transaction status").initCause(e); + } + } + + private Transaction getTransaction() throws SQLException { + Transaction transaction = (Transaction) this.transactionRef.get(); + if (transaction == null) { + throw new SQLException("Unable to enlist connection because the transaction has been garbage collected"); + } + return transaction; + } +} + diff --git a/src/main/java/com/alibaba/druid/pool/xa/TransactionContextListener.java b/src/main/java/com/alibaba/druid/pool/xa/TransactionContextListener.java new file mode 100644 index 0000000000..ab38b7cf8b --- /dev/null +++ b/src/main/java/com/alibaba/druid/pool/xa/TransactionContextListener.java @@ -0,0 +1,19 @@ +package com.alibaba.druid.pool.xa; + + +/** + * A listener for transaction completion events. + * + * @author Dain Sundstrom + * @version $Revision$ + */ +public interface TransactionContextListener { + + /** + * Occurs after the transaction commits or rolls back. + * + * @param transactionContext the transaction context that completed + * @param commited true if the transaction committed; false otherwise + */ + void afterCompletion(TransactionContext transactionContext, boolean commited); +} diff --git a/src/main/java/com/alibaba/druid/pool/xa/TransactionRegistry.java b/src/main/java/com/alibaba/druid/pool/xa/TransactionRegistry.java new file mode 100644 index 0000000000..db8ac0a298 --- /dev/null +++ b/src/main/java/com/alibaba/druid/pool/xa/TransactionRegistry.java @@ -0,0 +1,109 @@ +package com.alibaba.druid.pool.xa; + +import java.sql.Connection; +import java.sql.SQLException; +import java.util.Map; +import java.util.WeakHashMap; + +import javax.transaction.Status; +import javax.transaction.SystemException; +import javax.transaction.Transaction; +import javax.transaction.TransactionManager; +import javax.transaction.xa.XAResource; + + +/** + * TransactionRegistry tracks Connections and XAResources in a transacted environment for a single XAConnectionFactory. + *

+ * The TransactionRegistry hides the details of transaction processing from the existing DBCP pooling code, and gives + * the ManagedConnection a way to enlist connections in a transaction, allowing for the maximal rescue of DBCP. + * + * @author Dain Sundstrom + * @version $Revision$ + */ +public class TransactionRegistry { + private final TransactionManager transactionManager; + private final Map caches = new WeakHashMap(); + private final Map xaResources = new WeakHashMap(); + + /** + * Creates a TransactionRegistry for the specified transaction manager. + * @param transactionManager the transaction manager used to enlist connections + */ + public TransactionRegistry(TransactionManager transactionManager) { + this.transactionManager = transactionManager; + } + + /** + * Registers the association between a Connection and a XAResource. When a connection + * is enlisted in a transaction, it is actually the XAResource that is given to the transaction + * manager. + * + * @param connection the JDBC connection + * @param xaResource the XAResource which managed the connection within a transaction + */ + public synchronized void registerConnection(Connection connection, XAResource xaResource) { + if (connection == null) throw new NullPointerException("connection is null"); + if (xaResource == null) throw new NullPointerException("xaResource is null"); + xaResources.put(connection, xaResource); + } + + /** + * Gets the XAResource registered for the connection. + * @param connection the connection + * @return the XAResource registered for the connection; never null + * @throws SQLException if the connection does not have a registered XAResource + */ + public synchronized XAResource getXAResource(Connection connection) throws SQLException { + if (connection == null) throw new NullPointerException("connection is null"); + XAResource xaResource = (XAResource) xaResources.get(connection); + if (xaResource == null) { + throw new SQLException("Connection does not have a registered XAResource " + connection); + } + return xaResource; + } + + /** + * Gets the active TransactionContext or null if not Transaction is active. + * @return the active TransactionContext or null if not Transaction is active + * @throws SQLException if an error occurs while fetching the transaction + */ + public TransactionContext getActiveTransactionContext() throws SQLException { + Transaction transaction = null; + try { + transaction = transactionManager.getTransaction(); + + // was there a transaction? + if (transaction == null) { + return null; + } + + // is it active + int status = transaction.getStatus(); + if (status != Status.STATUS_ACTIVE && status != Status.STATUS_MARKED_ROLLBACK) { + return null; + } + } catch (SystemException e) { + throw (SQLException) new SQLException("Unable to determine current transaction ").initCause(e); + } + + // register the the context (or create a new one) + synchronized (this) { + TransactionContext cache = (TransactionContext) caches.get(transaction); + if (cache == null) { + cache = new TransactionContext(this, transaction); + caches.put(transaction, cache); + } + return cache; + } + } + + /** + * Unregisters a destroyed connection from {@link TransactionRegistry} + * @param connection + */ + public synchronized void unregisterConnection(Connection connection) { + xaResources.remove(connection); + } +} + diff --git a/src/main/java/com/alibaba/druid/pool/xa/XAConnectionFactory.java b/src/main/java/com/alibaba/druid/pool/xa/XAConnectionFactory.java new file mode 100644 index 0000000000..9f23a6c596 --- /dev/null +++ b/src/main/java/com/alibaba/druid/pool/xa/XAConnectionFactory.java @@ -0,0 +1,41 @@ +package com.alibaba.druid.pool.xa; + +import java.sql.Connection; +import java.sql.SQLException; + +import com.alibaba.druid.pool.ConnectionFactory; + +/** + * XAConnectionFactory is an extension of ConnectionFactory used to create connections + * in a transaction managed environment. The XAConnectionFactory opperates like a normal + * ConnectionFactory except an TransactionRegistry is provided from which the XAResource + * for a connection can be obtained. This allows the existing DBCP pool code to work with + * XAConnections and gives a the ManagedConnection a way to enlist a connection in the + * the transaction. + * + * @author Dain Sundstrom + * @author Rodney Waldhoff + * @version $Revision$ + */ +public interface XAConnectionFactory extends ConnectionFactory { + /** + * Gets the TransactionRegistry for this connection factory which contains a the + * XAResource for every connection created by this factory. + * + * @return the transaction registry for this connection factory + */ + TransactionRegistry getTransactionRegistry(); + + /** + * Create a new {@link java.sql.Connection} in an implementation specific fashion. + *

+ * An implementation can assume that the caller of this will wrap the connection in + * a proxy that protects access to the setAutoCommit, commit and rollback when + * enrolled in a XA transaction. + * + * @return a new {@link java.sql.Connection} + * @throws java.sql.SQLException if a database error occurs creating the connection + */ + Connection createConnection() throws SQLException; +} +