mirror of
https://github.com/alibaba/druid.git
synced 2023-03-04 19:17:16 +00:00
支持XA
git-svn-id: http://code.alibabatech.com/svn/druid/trunk@27 b9813039-fb51-4c41-a8b9-e21c2acb5095
This commit is contained in:
wenshao
@@ -62,14 +62,14 @@ public abstract class DruidDataAbstractSource implements DataSource, DataSourceP
|
||||
*/
|
||||
public static final long DEFAULT_MIN_EVICTABLE_IDLE_TIME_MILLIS = 1000L * 60L * 30L;
|
||||
|
||||
private boolean defaultAutoCommit = false;
|
||||
private Boolean defaultReadOnly;
|
||||
private Integer defaultTransactionIsolation;
|
||||
private String defaultCatalog = null;
|
||||
protected boolean defaultAutoCommit = false;
|
||||
protected Boolean defaultReadOnly;
|
||||
protected Integer defaultTransactionIsolation;
|
||||
protected String defaultCatalog = null;
|
||||
|
||||
protected String name;
|
||||
|
||||
protected String user;
|
||||
protected String username;
|
||||
protected String password;
|
||||
protected String jdbcUrl;
|
||||
protected String driverClass;
|
||||
@@ -434,7 +434,7 @@ public abstract class DruidDataAbstractSource implements DataSource, DataSourceP
|
||||
}
|
||||
|
||||
public String getUsername() {
|
||||
return user;
|
||||
return username;
|
||||
}
|
||||
|
||||
public void setUsername(String user) {
|
||||
@@ -442,7 +442,7 @@ public abstract class DruidDataAbstractSource implements DataSource, DataSourceP
|
||||
throw new UnsupportedOperationException();
|
||||
}
|
||||
|
||||
this.user = user;
|
||||
this.username = user;
|
||||
}
|
||||
|
||||
public String getPassword() {
|
||||
@@ -501,7 +501,7 @@ public abstract class DruidDataAbstractSource implements DataSource, DataSourceP
|
||||
public PrintWriter getLogWriter() {
|
||||
return logWriter;
|
||||
}
|
||||
|
||||
|
||||
protected void printStackTrace(Throwable e) {
|
||||
if (logWriter != null) {
|
||||
e.printStackTrace(logWriter);
|
||||
@@ -533,8 +533,12 @@ public abstract class DruidDataAbstractSource implements DataSource, DataSourceP
|
||||
return false;
|
||||
}
|
||||
|
||||
protected void initConnectionFactory() {
|
||||
connectionFactory = new DruidPoolConnectionFactory(this);
|
||||
protected void initConnectionFactory() throws SQLException {
|
||||
connectionFactory = createConnectionFactory();
|
||||
}
|
||||
|
||||
protected ConnectionFactory createConnectionFactory() throws SQLException {
|
||||
return new DruidPoolConnectionFactory(this);
|
||||
}
|
||||
|
||||
public Driver getDriver() {
|
||||
|
||||
@@ -247,7 +247,7 @@ public class DruidDataSource extends DruidDataAbstractSource implements DruidDat
|
||||
/**
|
||||
* 回收连接
|
||||
*/
|
||||
void recycle(PoolableConnection pooledConnection) throws SQLException {
|
||||
protected void recycle(PoolableConnection pooledConnection) throws SQLException {
|
||||
final Connection conn = pooledConnection.getConnection();
|
||||
try {
|
||||
if (activeConnectionTraceEnable) {
|
||||
|
||||
@@ -44,10 +44,10 @@ import com.alibaba.druid.pool.PreparedStatementPool.MethodType;
|
||||
/**
|
||||
* @author wenshao<szujobs@hotmail.com>
|
||||
*/
|
||||
public final class PoolableConnection implements PooledConnection, Connection {
|
||||
public class PoolableConnection implements PooledConnection, Connection {
|
||||
|
||||
private Connection conn;
|
||||
private ConnectionHolder holder;
|
||||
protected Connection conn;
|
||||
protected ConnectionHolder holder;
|
||||
|
||||
public PoolableConnection(ConnectionHolder holder){
|
||||
this.conn = holder.getConnection();
|
||||
@@ -413,8 +413,10 @@ public final class PoolableConnection implements PooledConnection, Connection {
|
||||
|
||||
@Override
|
||||
public boolean isClosed() throws SQLException {
|
||||
checkOpen();
|
||||
|
||||
if (holder == null) {
|
||||
return true;
|
||||
}
|
||||
|
||||
return conn.isClosed();
|
||||
}
|
||||
|
||||
|
||||
@@ -0,0 +1,99 @@
|
||||
package com.alibaba.druid.pool.xa;
|
||||
|
||||
import javax.sql.XAConnection;
|
||||
import javax.sql.XADataSource;
|
||||
import javax.transaction.TransactionManager;
|
||||
import javax.transaction.xa.XAResource;
|
||||
import java.sql.Connection;
|
||||
import java.sql.SQLException;
|
||||
|
||||
/**
|
||||
* An implementation of XAConnectionFactory which uses a real XADataSource to obtain connections and XAResources.
|
||||
*
|
||||
* @author Dain Sundstrom
|
||||
* @version $Revision$
|
||||
*/
|
||||
public class DataSourceXAConnectionFactory implements XAConnectionFactory {
|
||||
protected TransactionRegistry transactionRegistry;
|
||||
protected XADataSource xaDataSource;
|
||||
protected String username;
|
||||
protected String password;
|
||||
|
||||
/**
|
||||
* Creates an DataSourceXAConnectionFactory which uses the specified XADataSource to create database
|
||||
* connections. The connections are enlisted into transactions using the specified transaction manager.
|
||||
*
|
||||
* @param transactionManager the transaction manager in which connections will be enlisted
|
||||
* @param xaDataSource the data source from which connections will be retrieved
|
||||
*/
|
||||
public DataSourceXAConnectionFactory(TransactionManager transactionManager, XADataSource xaDataSource) {
|
||||
this(transactionManager, xaDataSource, null, null);
|
||||
}
|
||||
|
||||
/**
|
||||
* Creates an DataSourceXAConnectionFactory which uses the specified XADataSource to create database
|
||||
* connections. The connections are enlisted into transactions using the specified transaction manager.
|
||||
*
|
||||
* @param transactionManager the transaction manager in which connections will be enlisted
|
||||
* @param xaDataSource the data source from which connections will be retrieved
|
||||
* @param username the username used for authenticating new connections or null for unauthenticated
|
||||
* @param password the password used for authenticating new connections
|
||||
*/
|
||||
public DataSourceXAConnectionFactory(TransactionManager transactionManager, XADataSource xaDataSource, String username, String password) {
|
||||
if (transactionManager == null) throw new NullPointerException("transactionManager is null");
|
||||
if (xaDataSource == null) throw new NullPointerException("xaDataSource is null");
|
||||
|
||||
this.transactionRegistry = new TransactionRegistry(transactionManager);
|
||||
this.xaDataSource = xaDataSource;
|
||||
this.username = username;
|
||||
this.password = password;
|
||||
}
|
||||
|
||||
/**
|
||||
* Gets the username used to authenticate new connections.
|
||||
* @return the user name or null if unauthenticated connections are used
|
||||
*/
|
||||
public String getUsername() {
|
||||
return username;
|
||||
}
|
||||
|
||||
/**
|
||||
* Sets the username used to authenticate new connections.
|
||||
* @param username the username used for authenticating the connection or null for unauthenticated
|
||||
*/
|
||||
public void setUsername(String username) {
|
||||
this.username = username;
|
||||
}
|
||||
|
||||
/**
|
||||
* Sets the password used to authenticate new connections.
|
||||
* @param password the password used for authenticating the connection or null for unauthenticated
|
||||
*/
|
||||
public void setPassword(String password) {
|
||||
this.password = password;
|
||||
}
|
||||
|
||||
public TransactionRegistry getTransactionRegistry() {
|
||||
return transactionRegistry;
|
||||
}
|
||||
|
||||
public Connection createConnection() throws SQLException {
|
||||
// create a new XAConection
|
||||
XAConnection xaConnection;
|
||||
if (username == null) {
|
||||
xaConnection = xaDataSource.getXAConnection();
|
||||
} else {
|
||||
xaConnection = xaDataSource.getXAConnection(username, password);
|
||||
}
|
||||
|
||||
// get the real connection and XAResource from the connection
|
||||
Connection connection = xaConnection.getConnection();
|
||||
XAResource xaResource = xaConnection.getXAResource();
|
||||
|
||||
// register the xa resource for the connection
|
||||
transactionRegistry.registerConnection(connection, xaResource);
|
||||
|
||||
return connection;
|
||||
}
|
||||
}
|
||||
|
||||
@@ -19,21 +19,170 @@ import java.sql.SQLException;
|
||||
|
||||
import javax.sql.XAConnection;
|
||||
import javax.sql.XADataSource;
|
||||
import javax.transaction.TransactionManager;
|
||||
|
||||
import org.apache.commons.dbcp.AbandonedConfig;
|
||||
import org.apache.commons.dbcp.PoolableConnectionFactory;
|
||||
import org.apache.commons.dbcp.PoolingDataSource;
|
||||
import org.apache.commons.dbcp.managed.ManagedDataSource;
|
||||
import org.apache.commons.dbcp.managed.PoolableManagedConnectionFactory;
|
||||
import org.apache.commons.pool.KeyedObjectPoolFactory;
|
||||
|
||||
import com.alibaba.druid.pool.ConnectionFactory;
|
||||
import com.alibaba.druid.pool.DruidDataSource;
|
||||
|
||||
public class DruidXADataSource extends DruidDataSource implements XADataSource {
|
||||
public class DruidXADataSource extends DruidDataSource {
|
||||
|
||||
private static final long serialVersionUID = 1L;
|
||||
private static final long serialVersionUID = 1L;
|
||||
|
||||
@Override
|
||||
public XAConnection getXAConnection() throws SQLException {
|
||||
return null;
|
||||
/** Transaction Registry */
|
||||
private TransactionRegistry transactionRegistry;
|
||||
/** Transaction Manager */
|
||||
private transient TransactionManager transactionManager;
|
||||
/** XA datasource class name */
|
||||
private String xaDataSource;
|
||||
/** XA datasource instance */
|
||||
private XADataSource xaDataSourceInstance;
|
||||
|
||||
/**
|
||||
* Gets the XADataSource instance used by the XAConnectionFactory.
|
||||
*
|
||||
* @return the XADataSource
|
||||
*/
|
||||
public synchronized XADataSource getXaDataSourceInstance() {
|
||||
return xaDataSourceInstance;
|
||||
}
|
||||
|
||||
@Override
|
||||
public XAConnection getXAConnection(String user, String password) throws SQLException {
|
||||
return null;
|
||||
/**
|
||||
* <p>
|
||||
* Sets the XADataSource instance used by the XAConnectionFactory.
|
||||
* </p>
|
||||
* <p>
|
||||
* Note: this method currently has no effect once the pool has been initialized. The pool is initialized the first
|
||||
* time one of the following methods is invoked: <code>getConnection, setLogwriter,
|
||||
* setLoginTimeout, getLoginTimeout, getLogWriter.</code>
|
||||
* </p>
|
||||
*
|
||||
* @param xaDataSourceInstance XADataSource instance
|
||||
*/
|
||||
public synchronized void setXaDataSourceInstance(XADataSource xaDataSourceInstance) {
|
||||
this.xaDataSourceInstance = xaDataSourceInstance;
|
||||
xaDataSource = xaDataSourceInstance == null ? null : xaDataSourceInstance.getClass().getName();
|
||||
}
|
||||
|
||||
/**
|
||||
* Gets the required transaction manager property.
|
||||
*
|
||||
* @return the transaction manager used to enlist connections
|
||||
*/
|
||||
public TransactionManager getTransactionManager() {
|
||||
return transactionManager;
|
||||
}
|
||||
|
||||
/**
|
||||
* Gets the transaction registry.
|
||||
*
|
||||
* @return the transaction registry associating XAResources with managed connections
|
||||
*/
|
||||
protected synchronized TransactionRegistry getTransactionRegistry() {
|
||||
return transactionRegistry;
|
||||
}
|
||||
|
||||
/**
|
||||
* Sets the required transaction manager property.
|
||||
*
|
||||
* @param transactionManager the transaction manager used to enlist connections
|
||||
*/
|
||||
public void setTransactionManager(TransactionManager transactionManager) {
|
||||
this.transactionManager = transactionManager;
|
||||
}
|
||||
|
||||
/**
|
||||
* Gets the optional XADataSource class name.
|
||||
*
|
||||
* @return the optional XADataSource class name
|
||||
*/
|
||||
public synchronized String getXADataSource() {
|
||||
return xaDataSource;
|
||||
}
|
||||
|
||||
/**
|
||||
* Sets the optional XADataSource class name.
|
||||
*
|
||||
* @param xaDataSource the optional XADataSource class name
|
||||
*/
|
||||
public synchronized void setXADataSource(String xaDataSource) {
|
||||
this.xaDataSource = xaDataSource;
|
||||
}
|
||||
|
||||
protected ConnectionFactory createConnectionFactory() throws SQLException {
|
||||
if (transactionManager == null) {
|
||||
throw new SQLException("Transaction manager must be set before a connection can be created");
|
||||
}
|
||||
|
||||
// If xa data source is not specified a DriverConnectionFactory is created and wrapped with a
|
||||
// LocalXAConnectionFactory
|
||||
if (xaDataSource == null) {
|
||||
ConnectionFactory connectionFactory = super.createConnectionFactory();
|
||||
XAConnectionFactory xaConnectionFactory = new LocalXAConnectionFactory(getTransactionManager(),
|
||||
connectionFactory);
|
||||
transactionRegistry = xaConnectionFactory.getTransactionRegistry();
|
||||
return xaConnectionFactory;
|
||||
}
|
||||
|
||||
// Create the XADataSource instance using the configured class name if it has not been set
|
||||
if (xaDataSourceInstance == null) {
|
||||
Class xaDataSourceClass = null;
|
||||
try {
|
||||
xaDataSourceClass = Class.forName(xaDataSource);
|
||||
} catch (Throwable t) {
|
||||
String message = "Cannot load XA data source class '" + xaDataSource + "'";
|
||||
throw (SQLException) new SQLException(message).initCause(t);
|
||||
}
|
||||
|
||||
try {
|
||||
xaDataSourceInstance = (XADataSource) xaDataSourceClass.newInstance();
|
||||
} catch (Throwable t) {
|
||||
String message = "Cannot create XA data source of class '" + xaDataSource + "'";
|
||||
throw (SQLException) new SQLException(message).initCause(t);
|
||||
}
|
||||
}
|
||||
|
||||
// finally, create the XAConectionFactory using the XA data source
|
||||
XAConnectionFactory xaConnectionFactory = new DataSourceXAConnectionFactory(getTransactionManager(),
|
||||
xaDataSourceInstance, username,
|
||||
password);
|
||||
transactionRegistry = xaConnectionFactory.getTransactionRegistry();
|
||||
return xaConnectionFactory;
|
||||
}
|
||||
|
||||
/**
|
||||
* Creates the PoolableConnectionFactory and attaches it to the connection pool.
|
||||
*
|
||||
* @param driverConnectionFactory JDBC connection factory created by {@link #createConnectionFactory()}
|
||||
* @param statementPoolFactory statement pool factory (null if statement pooling is turned off)
|
||||
* @param abandonedConfig abandoned connection tracking configuration (null if no tracking)
|
||||
* @throws SQLException if an error occurs creating the PoolableConnectionFactory
|
||||
*/
|
||||
protected void createPoolableConnectionFactory(ConnectionFactory driverConnectionFactory,
|
||||
KeyedObjectPoolFactory statementPoolFactory,
|
||||
AbandonedConfig abandonedConfig) throws SQLException {
|
||||
// PoolableConnectionFactory connectionFactory = null;
|
||||
// try {
|
||||
// connectionFactory = new PoolableManagedConnectionFactory((XAConnectionFactory) driverConnectionFactory,
|
||||
// connectionPool, statementPoolFactory,
|
||||
// validationQuery, validationQueryTimeout,
|
||||
// connectionInitSqls, defaultReadOnly,
|
||||
// defaultAutoCommit, defaultTransactionIsolation,
|
||||
// defaultCatalog, abandonedConfig);
|
||||
// validateConnectionFactory(connectionFactory);
|
||||
// } catch (RuntimeException e) {
|
||||
// throw e;
|
||||
// } catch (Exception e) {
|
||||
// throw (SQLException) new SQLException("Cannot create PoolableConnectionFactory (" + e.getMessage() +
|
||||
// ")").initCause(e);
|
||||
// }
|
||||
throw new SQLException("TODO");
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
@@ -0,0 +1,285 @@
|
||||
package com.alibaba.druid.pool.xa;
|
||||
|
||||
import java.sql.Connection;
|
||||
import java.sql.SQLException;
|
||||
|
||||
import javax.transaction.TransactionManager;
|
||||
import javax.transaction.xa.XAException;
|
||||
import javax.transaction.xa.XAResource;
|
||||
import javax.transaction.xa.Xid;
|
||||
|
||||
import com.alibaba.druid.pool.ConnectionFactory;
|
||||
|
||||
/**
|
||||
* An implementation of XAConnectionFactory which manages non-XA connections in XA transactions. A non-XA connection
|
||||
* commits and rolls back as part of the XA transaction, but is not recoverable since the connection does not implement
|
||||
* the 2-phase protocol.
|
||||
*
|
||||
* @author Dain Sundstrom
|
||||
* @version $Revision$
|
||||
*/
|
||||
public class LocalXAConnectionFactory implements XAConnectionFactory {
|
||||
|
||||
protected TransactionRegistry transactionRegistry;
|
||||
protected ConnectionFactory connectionFactory;
|
||||
|
||||
/**
|
||||
* Creates an LocalXAConnectionFactory which uses the specified connection factory to create database connections.
|
||||
* The connections are enlisted into transactions using the specified transaction manager.
|
||||
*
|
||||
* @param transactionManager the transaction manager in which connections will be enlisted
|
||||
* @param connectionFactory the connection factory from which connections will be retrieved
|
||||
*/
|
||||
public LocalXAConnectionFactory(TransactionManager transactionManager, ConnectionFactory connectionFactory){
|
||||
if (transactionManager == null) throw new NullPointerException("transactionManager is null");
|
||||
if (connectionFactory == null) throw new NullPointerException("connectionFactory is null");
|
||||
|
||||
this.transactionRegistry = new TransactionRegistry(transactionManager);
|
||||
this.connectionFactory = connectionFactory;
|
||||
}
|
||||
|
||||
public TransactionRegistry getTransactionRegistry() {
|
||||
return transactionRegistry;
|
||||
}
|
||||
|
||||
public Connection createConnection() throws SQLException {
|
||||
// create a new connection
|
||||
Connection connection = connectionFactory.createConnection();
|
||||
|
||||
// create a XAResource to manage the connection during XA transactions
|
||||
XAResource xaResource = new LocalXAResource(connection);
|
||||
|
||||
// register the xa resource for the connection
|
||||
transactionRegistry.registerConnection(connection, xaResource);
|
||||
|
||||
return connection;
|
||||
}
|
||||
|
||||
/**
|
||||
* LocalXAResource is a fake XAResource for non-XA connections. When a transaction is started the connection
|
||||
* auto-commit is turned off. When the connection is committed or rolled back, the commit or rollback method is
|
||||
* called on the connection and then the original auto-commit value is restored. </p> The LocalXAResource also
|
||||
* respects the connection read-only setting. If the connection is read-only the commit method will not be called,
|
||||
* and the prepare method returns the XA_RDONLY. </p> It is assumed that the wrapper around a managed connection
|
||||
* disables the setAutoCommit(), commit(), rollback() and setReadOnly() methods while a transaction is in progress.
|
||||
*/
|
||||
protected static class LocalXAResource implements XAResource {
|
||||
|
||||
private final Connection connection;
|
||||
private Xid currentXid;
|
||||
private boolean originalAutoCommit;
|
||||
|
||||
public LocalXAResource(Connection localTransaction){
|
||||
this.connection = localTransaction;
|
||||
}
|
||||
|
||||
/**
|
||||
* Gets the current xid of the transaction branch associated with this XAResource.
|
||||
*
|
||||
* @return the current xid of the transaction branch associated with this XAResource.
|
||||
*/
|
||||
public synchronized Xid getXid() {
|
||||
return currentXid;
|
||||
}
|
||||
|
||||
/**
|
||||
* Signals that a the connection has been enrolled in a transaction. This method saves off the current auto
|
||||
* commit flag, and then disables auto commit. The original auto commit setting is restored when the transaction
|
||||
* completes.
|
||||
*
|
||||
* @param xid the id of the transaction branch for this connection
|
||||
* @param flag either XAResource.TMNOFLAGS or XAResource.TMRESUME
|
||||
* @throws XAException if the connection is already enlisted in another transaction, or if auto-commit could not
|
||||
* be disabled
|
||||
*/
|
||||
public synchronized void start(Xid xid, int flag) throws XAException {
|
||||
if (flag == XAResource.TMNOFLAGS) {
|
||||
// first time in this transaction
|
||||
|
||||
// make sure we aren't already in another tx
|
||||
if (this.currentXid != null) {
|
||||
throw new XAException("Already enlisted in another transaction with xid " + xid);
|
||||
}
|
||||
|
||||
// save off the current auto commit flag so it can be restored after the transaction completes
|
||||
try {
|
||||
originalAutoCommit = connection.getAutoCommit();
|
||||
} catch (SQLException ignored) {
|
||||
// no big deal, just assume it was off
|
||||
originalAutoCommit = true;
|
||||
}
|
||||
|
||||
// update the auto commit flag
|
||||
try {
|
||||
connection.setAutoCommit(false);
|
||||
} catch (SQLException e) {
|
||||
throw (XAException) new XAException("Count not turn off auto commit for a XA transaction").initCause(e);
|
||||
}
|
||||
|
||||
this.currentXid = xid;
|
||||
} else if (flag == XAResource.TMRESUME) {
|
||||
if (xid != this.currentXid) {
|
||||
throw new XAException("Attempting to resume in different transaction: expected " + this.currentXid
|
||||
+ ", but was " + xid);
|
||||
}
|
||||
} else {
|
||||
throw new XAException("Unknown start flag " + flag);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* This method does nothing.
|
||||
*
|
||||
* @param xid the id of the transaction branch for this connection
|
||||
* @param flag ignored
|
||||
* @throws XAException if the connection is already enlisted in another transaction
|
||||
*/
|
||||
public synchronized void end(Xid xid, int flag) throws XAException {
|
||||
if (xid == null) throw new NullPointerException("xid is null");
|
||||
if (!this.currentXid.equals(xid)) throw new XAException("Invalid Xid: expected " + this.currentXid
|
||||
+ ", but was " + xid);
|
||||
|
||||
// This notification tells us that the application server is done using this
|
||||
// connection for the time being. The connection is still associated with an
|
||||
// open transaction, so we must still wait for the commit or rollback method
|
||||
}
|
||||
|
||||
/**
|
||||
* This method does nothing since the LocalXAConnection does not support two-phase-commit. This method will
|
||||
* return XAResource.XA_RDONLY if the connection isReadOnly(). This assumes that the physical connection is
|
||||
* wrapped with a proxy that prevents an application from changing the read-only flag while enrolled in a
|
||||
* transaction.
|
||||
*
|
||||
* @param xid the id of the transaction branch for this connection
|
||||
* @return XAResource.XA_RDONLY if the connection.isReadOnly(); XAResource.XA_OK otherwise
|
||||
*/
|
||||
public synchronized int prepare(Xid xid) {
|
||||
// if the connection is read-only, then the resource is read-only
|
||||
// NOTE: this assumes that the outer proxy throws an exception when application code
|
||||
// attempts to set this in a transaction
|
||||
try {
|
||||
if (connection.isReadOnly()) {
|
||||
// update the auto commit flag
|
||||
connection.setAutoCommit(originalAutoCommit);
|
||||
|
||||
// tell the transaction manager we are read only
|
||||
return XAResource.XA_RDONLY;
|
||||
}
|
||||
} catch (SQLException ignored) {
|
||||
// no big deal
|
||||
}
|
||||
|
||||
// this is a local (one phase) only connection, so we can't prepare
|
||||
return XAResource.XA_OK;
|
||||
}
|
||||
|
||||
/**
|
||||
* Commits the transaction and restores the original auto commit setting.
|
||||
*
|
||||
* @param xid the id of the transaction branch for this connection
|
||||
* @param flag ignored
|
||||
* @throws XAException if connection.commit() throws a SQLException
|
||||
*/
|
||||
public synchronized void commit(Xid xid, boolean flag) throws XAException {
|
||||
if (xid == null) throw new NullPointerException("xid is null");
|
||||
if (!this.currentXid.equals(xid)) throw new XAException("Invalid Xid: expected " + this.currentXid
|
||||
+ ", but was " + xid);
|
||||
|
||||
try {
|
||||
// make sure the connection isn't already closed
|
||||
if (connection.isClosed()) {
|
||||
throw new XAException("Conection is closed");
|
||||
}
|
||||
|
||||
// A read only connection should not be committed
|
||||
if (!connection.isReadOnly()) {
|
||||
connection.commit();
|
||||
}
|
||||
} catch (SQLException e) {
|
||||
throw (XAException) new XAException().initCause(e);
|
||||
} finally {
|
||||
try {
|
||||
connection.setAutoCommit(originalAutoCommit);
|
||||
} catch (SQLException e) {
|
||||
}
|
||||
this.currentXid = null;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Rolls back the transaction and restores the original auto commit setting.
|
||||
*
|
||||
* @param xid the id of the transaction branch for this connection
|
||||
* @throws XAException if connection.rollback() throws a SQLException
|
||||
*/
|
||||
public synchronized void rollback(Xid xid) throws XAException {
|
||||
if (xid == null) throw new NullPointerException("xid is null");
|
||||
if (!this.currentXid.equals(xid)) throw new XAException("Invalid Xid: expected " + this.currentXid
|
||||
+ ", but was " + xid);
|
||||
|
||||
try {
|
||||
connection.rollback();
|
||||
} catch (SQLException e) {
|
||||
throw (XAException) new XAException().initCause(e);
|
||||
} finally {
|
||||
try {
|
||||
connection.setAutoCommit(originalAutoCommit);
|
||||
} catch (SQLException e) {
|
||||
}
|
||||
this.currentXid = null;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Returns true if the specified XAResource == this XAResource.
|
||||
*
|
||||
* @param xaResource the XAResource to test
|
||||
* @return true if the specified XAResource == this XAResource; false otherwise
|
||||
*/
|
||||
public boolean isSameRM(XAResource xaResource) {
|
||||
return this == xaResource;
|
||||
}
|
||||
|
||||
/**
|
||||
* Clears the currently associated transaction if it is the specified xid.
|
||||
*
|
||||
* @param xid the id of the transaction to forget
|
||||
*/
|
||||
public synchronized void forget(Xid xid) {
|
||||
if (xid != null && this.currentXid.equals(xid)) {
|
||||
this.currentXid = null;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Always returns a zero length Xid array. The LocalXAConnectionFactory can not support recovery, so no xids
|
||||
* will ever be found.
|
||||
*
|
||||
* @param flag ignored since recovery is not supported
|
||||
* @return always a zero length Xid array.
|
||||
*/
|
||||
public Xid[] recover(int flag) {
|
||||
return new Xid[0];
|
||||
}
|
||||
|
||||
/**
|
||||
* Always returns 0 since we have no way to set a transaction timeout on a JDBC connection.
|
||||
*
|
||||
* @return always 0
|
||||
*/
|
||||
public int getTransactionTimeout() {
|
||||
return 0;
|
||||
}
|
||||
|
||||
/**
|
||||
* Always returns false since we have no way to set a transaction timeout on a JDBC connection.
|
||||
*
|
||||
* @param transactionTimeout ignored since we have no way to set a transaction timeout on a JDBC connection
|
||||
* @return always false
|
||||
*/
|
||||
public boolean setTransactionTimeout(int transactionTimeout) {
|
||||
return false;
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
@@ -0,0 +1,236 @@
|
||||
package com.alibaba.druid.pool.xa;
|
||||
|
||||
import java.sql.Connection;
|
||||
import java.sql.SQLException;
|
||||
|
||||
import com.alibaba.druid.pool.ConnectionHolder;
|
||||
import com.alibaba.druid.pool.PoolableConnection;
|
||||
|
||||
/**
|
||||
* ManagedConnection is responsible for managing a database connection in a transactional environment (typically called
|
||||
* "Container Managed"). A managed connection operates like any other connection when no global transaction (a.k.a. XA
|
||||
* transaction or JTA Transaction) is in progress. When a global transaction is active a single physical connection to
|
||||
* the database is used by all ManagedConnections accessed in the scope of the transaction. Connection sharing means
|
||||
* that all data access during a transaction has a consistent view of the database. When the global transaction is
|
||||
* committed or rolled back the enlisted connections are committed or rolled back. Typically upon transaction
|
||||
* completion, a connection returns to the auto commit setting in effect before being enlisted in the transaction, but
|
||||
* some vendors do not properly implement this. When enlisted in a transaction the setAutoCommit(), commit(),
|
||||
* rollback(), and setReadOnly() methods throw a SQLException. This is necessary to assure that the transaction
|
||||
* completes as a single unit.
|
||||
*
|
||||
* @author Dain Sundstrom
|
||||
* @version $Revision$
|
||||
*/
|
||||
public class ManagedConnection extends PoolableConnection {
|
||||
|
||||
private final TransactionRegistry transactionRegistry;
|
||||
private final boolean accessToUnderlyingConnectionAllowed;
|
||||
private TransactionContext transactionContext;
|
||||
private boolean isSharedConnection;
|
||||
|
||||
public ManagedConnection(ConnectionHolder holder, TransactionRegistry transactionRegistry,
|
||||
boolean accessToUnderlyingConnectionAllowed) throws SQLException{
|
||||
super(holder);
|
||||
this.transactionRegistry = transactionRegistry;
|
||||
this.accessToUnderlyingConnectionAllowed = accessToUnderlyingConnectionAllowed;
|
||||
updateTransactionStatus();
|
||||
}
|
||||
|
||||
protected void checkOpen() throws SQLException {
|
||||
super.checkOpen();
|
||||
updateTransactionStatus();
|
||||
}
|
||||
|
||||
private void updateTransactionStatus() throws SQLException {
|
||||
// // if there is a is an active transaction context, assure the transaction context hasn't changed
|
||||
// if (transactionContext != null) {
|
||||
// if (transactionContext.isActive()) {
|
||||
// if (transactionContext != transactionRegistry.getActiveTransactionContext()) {
|
||||
// throw new SQLException("Connection can not be used while enlisted in another transaction");
|
||||
// }
|
||||
// return;
|
||||
// } else {
|
||||
// // transaction should have been cleared up by TransactionContextListener, but in
|
||||
// // rare cases another lister could have registered which uses the connection before
|
||||
// // our listener is called. In that rare case, trigger the transaction complete call now
|
||||
// transactionComplete();
|
||||
// }
|
||||
// }
|
||||
//
|
||||
// // the existing transaction context ended (or we didn't have one), get the active transaction context
|
||||
// transactionContext = transactionRegistry.getActiveTransactionContext();
|
||||
//
|
||||
// // if there is an active transaction context and it already has a shared connection, use it
|
||||
// if (transactionContext != null && transactionContext.getSharedConnection() != null) {
|
||||
// // A connection for the connection factory has already been enrolled
|
||||
// // in the transaction, replace our delegate with the enrolled connection
|
||||
//
|
||||
// // return current connection to the pool
|
||||
// Connection connection = getDelegateInternal();
|
||||
// setDelegate(null);
|
||||
// if (connection != null) {
|
||||
// try {
|
||||
// pool.returnObject(connection);
|
||||
// } catch (Exception ignored) {
|
||||
// // whatever... try to invalidate the connection
|
||||
// try {
|
||||
// pool.invalidateObject(connection);
|
||||
// } catch (Exception ignore) {
|
||||
// // no big deal
|
||||
// }
|
||||
// }
|
||||
// }
|
||||
//
|
||||
// // add a listener to the transaction context
|
||||
// transactionContext.addTransactionContextListener(new CompletionListener());
|
||||
//
|
||||
// // set our delegate to the shared connection
|
||||
// setDelegate(transactionContext.getSharedConnection());
|
||||
//
|
||||
// // remember that we are using a shared connection so it can be cleared after the
|
||||
// // transaction completes
|
||||
// isSharedConnection = true;
|
||||
// } else {
|
||||
// // if our delegate is null, create one
|
||||
// if (getDelegateInternal() == null) {
|
||||
// try {
|
||||
// // borrow a new connection from the pool
|
||||
// Connection connection = (Connection) pool.borrowObject();
|
||||
// setDelegate(connection);
|
||||
// } catch (Exception e) {
|
||||
// throw (SQLException) new SQLException("Unable to acquire a new connection from the pool").initCause(e);
|
||||
// }
|
||||
// }
|
||||
//
|
||||
// // if we have a transaction, out delegate becomes the shared delegate
|
||||
// if (transactionContext != null) {
|
||||
// // add a listener to the transaction context
|
||||
// transactionContext.addTransactionContextListener(new CompletionListener());
|
||||
//
|
||||
// // register our connection as the shared connection
|
||||
// try {
|
||||
// transactionContext.setSharedConnection(getDelegateInternal());
|
||||
// } catch (SQLException e) {
|
||||
// // transaction is hosed
|
||||
// transactionContext = null;
|
||||
// throw e;
|
||||
// }
|
||||
// }
|
||||
// }
|
||||
throw new SQLException("TODO");
|
||||
}
|
||||
|
||||
public void close() throws SQLException {
|
||||
if (transactionContext == null) {
|
||||
super.close();
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Delegates to {@link ManagedConnection#transactionComplete()} for transaction completion events.
|
||||
*/
|
||||
protected class CompletionListener implements TransactionContextListener {
|
||||
|
||||
public void afterCompletion(TransactionContext completedContext, boolean commited) {
|
||||
if (completedContext == transactionContext) {
|
||||
transactionComplete();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
protected void transactionComplete() {
|
||||
// transactionContext = null;
|
||||
//
|
||||
// // if we were using a shared connection, clear the reference now that the transaction has completed
|
||||
// if (isSharedConnection) {
|
||||
// // for now, just set the delegate to null, it will be created later if needed
|
||||
// setDelegate(null);
|
||||
// isSharedConnection = false;
|
||||
// }
|
||||
//
|
||||
// // if this connection was closed during the transaction and there is still a delegate present close it
|
||||
// Connection delegate = getDelegateInternal();
|
||||
// if (isClosed() && delegate != null) {
|
||||
// try {
|
||||
// setDelegate(null);
|
||||
//
|
||||
// // don't actually close the connection if in a transaction
|
||||
// if (!delegate.isClosed()) {
|
||||
// // don't use super.close() because it calls passivate() which marks the
|
||||
// // the connection as closed without returning it to the pool
|
||||
// delegate.close();
|
||||
// }
|
||||
// } catch (SQLException ignored) {
|
||||
// // not a whole lot we can do here as connection is closed
|
||||
// // and this is a transaction callback so there is no
|
||||
// // way to report the error
|
||||
// } finally {
|
||||
// _closed = true;
|
||||
// }
|
||||
// }
|
||||
throw new RuntimeException("TODO");
|
||||
}
|
||||
|
||||
//
|
||||
// The following methods can't be used while enlisted in a transaction
|
||||
//
|
||||
|
||||
public void setAutoCommit(boolean autoCommit) throws SQLException {
|
||||
if (transactionContext != null) {
|
||||
throw new SQLException("Auto-commit can not be set while enrolled in a transaction");
|
||||
}
|
||||
super.setAutoCommit(autoCommit);
|
||||
}
|
||||
|
||||
public void commit() throws SQLException {
|
||||
if (transactionContext != null) {
|
||||
throw new SQLException("Commit can not be set while enrolled in a transaction");
|
||||
}
|
||||
super.commit();
|
||||
}
|
||||
|
||||
public void rollback() throws SQLException {
|
||||
if (transactionContext != null) {
|
||||
throw new SQLException("Commit can not be set while enrolled in a transaction");
|
||||
}
|
||||
super.rollback();
|
||||
}
|
||||
|
||||
public void setReadOnly(boolean readOnly) throws SQLException {
|
||||
if (transactionContext != null) {
|
||||
throw new SQLException("Read-only can not be set while enrolled in a transaction");
|
||||
}
|
||||
super.setReadOnly(readOnly);
|
||||
}
|
||||
|
||||
//
|
||||
// Methods for accessing the delegate connection
|
||||
//
|
||||
|
||||
/**
|
||||
* If false, getDelegate() and getInnermostDelegate() will return null.
|
||||
*
|
||||
* @return if false, getDelegate() and getInnermostDelegate() will return null
|
||||
*/
|
||||
public boolean isAccessToUnderlyingConnectionAllowed() {
|
||||
return accessToUnderlyingConnectionAllowed;
|
||||
}
|
||||
|
||||
@Deprecated
|
||||
public Connection getDelegate() {
|
||||
if (isAccessToUnderlyingConnectionAllowed()) {
|
||||
return getRawConnection();
|
||||
} else {
|
||||
return null;
|
||||
}
|
||||
}
|
||||
|
||||
@Deprecated
|
||||
public Connection getInnermostDelegate() {
|
||||
if (isAccessToUnderlyingConnectionAllowed()) {
|
||||
return getRawConnection();
|
||||
} else {
|
||||
return null;
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,62 @@
|
||||
package com.alibaba.druid.pool.xa;
|
||||
|
||||
import java.sql.Connection;
|
||||
import java.sql.SQLException;
|
||||
|
||||
|
||||
/**
|
||||
* The ManagedDataSource is a PoolingDataSource that creates ManagedConnections.
|
||||
*
|
||||
* @author Dain Sundstrom
|
||||
* @version $Revision$
|
||||
*/
|
||||
public class ManagedDataSource {
|
||||
private TransactionRegistry transactionRegistry;
|
||||
|
||||
/**
|
||||
* Creates an uninitialized datasource. Before this data source can be used a pool and
|
||||
* transaction registry must be set.
|
||||
*/
|
||||
public ManagedDataSource() {
|
||||
}
|
||||
|
||||
/**
|
||||
* Creates a ManagedDataSource which obtains connections from the specified pool and
|
||||
* manages them using the specified transaction registry. The TransactionRegistry must
|
||||
* be the transaction registry obtained from the XAConnectionFactory used to create
|
||||
* the connection pool. If not an error will occure when attempting to use the connection
|
||||
* in a global transaction because the XAResource object associated with the connection
|
||||
* will be unavailable.
|
||||
*
|
||||
* @param pool the connection pool
|
||||
* @param transactionRegistry the transaction registry obtained from the
|
||||
* XAConnectionFactory used to create the connection pool object factory
|
||||
*/
|
||||
public ManagedDataSource(TransactionRegistry transactionRegistry) {
|
||||
this.transactionRegistry = transactionRegistry;
|
||||
}
|
||||
|
||||
/**
|
||||
* Sets the transaction registry from the XAConnectionFactory used to create the pool.
|
||||
* The transaction registry can only be set once using either a connector or this setter
|
||||
* method.
|
||||
* @param transactionRegistry the transaction registry acquired from the XAConnectionFactory
|
||||
* used to create the pool
|
||||
*/
|
||||
public void setTransactionRegistry(TransactionRegistry transactionRegistry) {
|
||||
if(this.transactionRegistry != null) throw new IllegalStateException("TransactionRegistry already set");
|
||||
if(transactionRegistry == null) throw new NullPointerException("TransactionRegistry is null");
|
||||
|
||||
this.transactionRegistry = transactionRegistry;
|
||||
}
|
||||
|
||||
public Connection getConnection() throws SQLException {
|
||||
// if (_pool == null) throw new IllegalStateException("Pool has not been set");
|
||||
// if (transactionRegistry == null) throw new IllegalStateException("TransactionRegistry has not been set");
|
||||
//
|
||||
// Connection connection = new ManagedConnection(_pool, transactionRegistry, isAccessToUnderlyingConnectionAllowed());
|
||||
// return connection;
|
||||
throw new SQLException("TODO"); // TODO
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user