From f9acc0e6b20df6f61be9a58ca8581e455b1b3bc2 Mon Sep 17 00:00:00 2001 From: wenshao Date: Tue, 21 Aug 2012 06:57:54 +0800 Subject: [PATCH] add xa support --- .../alibaba/druid/pool/DruidDataSource.java | 2 +- .../druid/pool/xa/DruidAXDataSource.java | 34 +- .../druid/pool/xa/DruidXAOracleException.java | 114 ---- .../druid/pool/xa/DruidXAOracleResource.java | 619 ------------------ .../pool/xa/DruidXAPooledConnection.java | 26 - .../druid/pool/xa/DruidXAResource.java | 205 ------ .../com/alibaba/druid/pool/xa/DruidXid.java | 82 --- .../com/alibaba/druid/util/MySqlUtils.java | 13 + .../com/alibaba/druid/util/OracleUtils.java | 7 + .../java/com/alibaba/druid/util/PGUtils.java | 14 + 10 files changed, 68 insertions(+), 1048 deletions(-) delete mode 100644 src/main/java/com/alibaba/druid/pool/xa/DruidXAOracleException.java delete mode 100644 src/main/java/com/alibaba/druid/pool/xa/DruidXAOracleResource.java delete mode 100644 src/main/java/com/alibaba/druid/pool/xa/DruidXAPooledConnection.java delete mode 100644 src/main/java/com/alibaba/druid/pool/xa/DruidXAResource.java delete mode 100644 src/main/java/com/alibaba/druid/pool/xa/DruidXid.java create mode 100644 src/main/java/com/alibaba/druid/util/MySqlUtils.java create mode 100644 src/main/java/com/alibaba/druid/util/PGUtils.java diff --git a/src/main/java/com/alibaba/druid/pool/DruidDataSource.java b/src/main/java/com/alibaba/druid/pool/DruidDataSource.java index 3ddeca21d0..7b6adf3a98 100644 --- a/src/main/java/com/alibaba/druid/pool/DruidDataSource.java +++ b/src/main/java/com/alibaba/druid/pool/DruidDataSource.java @@ -510,7 +510,7 @@ public class DruidDataSource extends DruidAbstractDataSource implements DruidDat LOG.error(errorMessage + "validationQuery not set"); } - private void initOracle() throws SQLException { + protected void initOracle() throws SQLException { isOracle = true; if (driver.getMajorVersion() < 10) { diff --git a/src/main/java/com/alibaba/druid/pool/xa/DruidAXDataSource.java b/src/main/java/com/alibaba/druid/pool/xa/DruidAXDataSource.java index 8bff65930e..51dd279ac2 100644 --- a/src/main/java/com/alibaba/druid/pool/xa/DruidAXDataSource.java +++ b/src/main/java/com/alibaba/druid/pool/xa/DruidAXDataSource.java @@ -1,19 +1,51 @@ package com.alibaba.druid.pool.xa; +import java.sql.Connection; import java.sql.SQLException; import javax.sql.XAConnection; import javax.sql.XADataSource; +import javax.transaction.xa.XAException; import com.alibaba.druid.pool.DruidDataSource; +import com.alibaba.druid.pool.DruidPooledConnection; +import com.alibaba.druid.support.logging.Log; +import com.alibaba.druid.support.logging.LogFactory; +import com.alibaba.druid.util.JdbcUtils; +import com.alibaba.druid.util.MySqlUtils; +import com.alibaba.druid.util.OracleUtils; +import com.alibaba.druid.util.PGUtils; public class DruidAXDataSource extends DruidDataSource implements XADataSource { + private final static Log LOG = LogFactory.getLog(DruidAXDataSource.class); + private static final long serialVersionUID = 1L; @Override public XAConnection getXAConnection() throws SQLException { - return null; + DruidPooledConnection conn = this.getConnection(); + + Connection physicalConn = conn.unwrap(Connection.class); + + if (JdbcUtils.ORACLE.equals(dbType)) { + try { + return OracleUtils.OracleXAConnection(physicalConn); + } catch (XAException xae) { + LOG.error("create xaConnection error", xae); + return null; + } + } + + if (JdbcUtils.MYSQL.equals(dbType)) { + return MySqlUtils.createXAConnection(physicalConn); + } + + if (JdbcUtils.POSTGRESQL.equals(dbType)) { + return PGUtils.createXAConnection(physicalConn); + } + + throw new SQLException("xa not support dbType : " + this.dbType); } @Override diff --git a/src/main/java/com/alibaba/druid/pool/xa/DruidXAOracleException.java b/src/main/java/com/alibaba/druid/pool/xa/DruidXAOracleException.java deleted file mode 100644 index 364eb81201..0000000000 --- a/src/main/java/com/alibaba/druid/pool/xa/DruidXAOracleException.java +++ /dev/null @@ -1,114 +0,0 @@ -package com.alibaba.druid.pool.xa; - -import javax.transaction.xa.XAException; - -import oracle.jdbc.internal.OracleConnection; - -public class DruidXAOracleException extends XAException { - - private static final long serialVersionUID = 1L; - private int xaError = 0; - private int primary = 0; - private int secondary = 0; - - public DruidXAOracleException(){ - } - - public DruidXAOracleException(int error){ - super(errorConvert(error)); - - this.xaError = errorConvert(error); - this.primary = (error & 0xFFFF); - this.secondary = (error >> 16); - } - - public static XAException newXAException(OracleConnection conn, int error) { - - DruidXAOracleException xaexc = new DruidXAOracleException(error); - - int xacode = xaexc.getXAError(); - if (xacode == -7) { - if (conn != null) { - // TODO conn.setUsable(false); - } - - } - - return xaexc; - } - - public int getXAError() { - return this.xaError; - } - - public int getOracleError() { - return this.primary; - } - - public int getOracleSQLError() { - return this.secondary; - } - - public static int errorConvert(int err) { - return errorConvert(err, -3); - } - - public static int errorConvert(int err, int defaultErrorCode) { - switch (err & 0xFFFF) { - case 24756: - return -4; - case 25351: - case 30006: - return 4; - case 24764: - return 7; - case 24765: - return 6; - case 24766: - return 5; - case 24767: - return 3; - case 28: - case 1031: - case 1033: - case 1034: - case 1041: - case 1089: - case 1090: - case 1092: - case 3113: - case 3114: - case 12571: - case 17002: - case 17008: - case 17410: - case 24796: - case 25400: - case 25401: - case 25402: - case 25403: - case 25404: - case 25405: - case 25406: - case 25407: - case 25408: - case 25409: - return -7; - case 2056: - case 17448: - case 24763: - case 24768: - case 24769: - case 24770: - case 24775: - case 24776: - return -6; - case 2091: - case 2092: - case 24761: - return 100; - } - return defaultErrorCode; - } - -} diff --git a/src/main/java/com/alibaba/druid/pool/xa/DruidXAOracleResource.java b/src/main/java/com/alibaba/druid/pool/xa/DruidXAOracleResource.java deleted file mode 100644 index 2f7acdbe3d..0000000000 --- a/src/main/java/com/alibaba/druid/pool/xa/DruidXAOracleResource.java +++ /dev/null @@ -1,619 +0,0 @@ -package com.alibaba.druid.pool.xa; - -import java.sql.CallableStatement; -import java.sql.Connection; -import java.sql.ResultSet; -import java.sql.SQLException; -import java.sql.Statement; -import java.util.ArrayList; - -import javax.transaction.xa.XAException; -import javax.transaction.xa.XAResource; -import javax.transaction.xa.Xid; - -import oracle.jdbc.internal.OracleConnection; - -import com.alibaba.druid.util.OracleUtils; - -public class DruidXAOracleResource extends DruidXAResource { - - private static String xa_start_post_816 = "begin ? := JAVA_XA.xa_start_new(?,?,?,?,?); end;"; - private static String xa_forget_post_816 = "begin ? := JAVA_XA.xa_forget_new (?,?,?); end;"; - private static String xa_prepare_post_816 = "begin ? := JAVA_XA.xa_prepare_new (?,?,?); end;"; - private static String xa_rollback_post_816 = "begin ? := JAVA_XA.xa_rollback_new (?,?,?); end;"; - private static String xa_commit_post_816 = "begin ? := JAVA_XA.xa_commit_new (?,?,?,?); end;"; - private static String xa_end_post_816 = "begin ? := JAVA_XA.xa_end_new(?,?,?,?); end;"; - - private String synchronizeBeforeRecoverNewCall = "BEGIN sys.dbms_xa.dist_txn_sync \n; END;"; - - private String synchronizeBeforeRecoverOldCall = "BEGIN sys.dbms_system.dist_txn_sync(0) \n; END;"; - - private String recoverySqlRows = "SELECT formatid, globalid, branchid FROM SYS.DBA_PENDING_TRANSACTIONS"; - - private int version; - private boolean needStackingForCommitRollbackPrepare = false; - boolean isTransLoose = false; - - public DruidXAOracleResource(DruidXAPooledConnection conn) throws SQLException, XAException{ - this.version = OracleUtils.getVersionNumber(conn); - - this.needStackingForCommitRollbackPrepare = (this.version < 9000); - - if (this.version < 8170) { - throw new XAException(XAException.XAER_PROTO); - } - } - - @Override - public void commit(Xid xid, boolean onePhase) throws XAException { - - synchronized (this.connection) { - if (xid == null) { - throw new XAException(-5); - } - - Xid stackedXid = null; - if (this.needStackingForCommitRollbackPrepare) { - stackedXid = super.suspendStacked(xid); - } else { - removeXidFromList(xid); - - if (this.activeXid == null) { - exitGlobalTxnMode(); - } - } - try { - try { - doCommit(xid, onePhase); - } catch (SQLException sqle) { - checkError(sqle, -3); - } - } catch (XAException xae) { - if (xae.errorCode == -7) { - try { - this.connection.close(); - } catch (SQLException ea) { - } - - } else if (this.needStackingForCommitRollbackPrepare) { - super.resumeStacked(stackedXid); - } - throw xae; - } - - if (this.needStackingForCommitRollbackPrepare) { - super.resumeStacked(stackedXid); - } - } - } - - protected void doCommit(Xid xid, boolean onePhase) throws XAException, SQLException { - - CallableStatement cstmt = null; - try { - cstmt = this.connection.prepareCall(xa_commit_post_816); - - cstmt.registerOutParameter(1, 2); - cstmt.setInt(2, xid.getFormatId()); - cstmt.setBytes(3, xid.getGlobalTransactionId()); - cstmt.setBytes(4, xid.getBranchQualifier()); - cstmt.setInt(5, onePhase ? 1 : 0); - - cstmt.execute(); - - int returnVal = cstmt.getInt(1); - checkError(returnVal, -7); - } catch (SQLException s) { - int returnVal = s.getErrorCode(); - - if (returnVal == 0) { - throw new XAException(-6); - } - - throw s; - } finally { - try { - if (cstmt != null) cstmt.close(); - } catch (SQLException s) { - } - cstmt = null; - } - } - - @Override - public void end(Xid xid, int flag) throws XAException { - - synchronized (this.connection) { - int returnVal = -1; - boolean isTMSUCCESS = false; - boolean isTMFAIL = false; - try { - if (xid == null) { - throw new XAException(-5); - } - - int validFlags = 638582786; - if ((flag & validFlags) != flag) { - throw new XAException(-5); - } - - Xid stackedXid = null; - isTMSUCCESS = (flag & 0x4000000) != 0; - isTMFAIL = (flag & 0x20000000) != 0; - - if ((isTMSUCCESS) || (isTMFAIL)) { - stackedXid = super.suspendStacked(xid); - } - try { - boolean isLocallySuspended = false; - boolean i = false; - if ((isTMSUCCESS) || (isTMFAIL)) { - isLocallySuspended = isXidSuspended(xid); - - if (isLocallySuspended) { - super.resumeStacked(xid); - } - - removeXidFromList(xid); - } else if (flag == 33554432) { - boolean[] flags = { false }; - super.createOrUpdateXid(xid, true, flags); - - i = flags[0]; - } - - returnVal = doEnd(xid, flag, i); - } catch (SQLException sqle) { - checkError(sqle, -3); - } - - if (stackedXid != null) { - super.resumeStacked(stackedXid); - } else if (isXidListEmpty()) { - exitGlobalTxnMode(); - this.activeXid = null; - } - - checkError(returnVal); - - if (((isTMSUCCESS) && (flag != 67108864)) || ((isTMFAIL) && (flag != 536870912))) { - throw new XAException(-5); - } - - } finally { - restoreAutoCommitModeForGlobalTransaction(); - } - - } - } - - protected int doEnd(Xid xid, int flag, boolean isLocallySuspended) throws XAException, SQLException { - - CallableStatement cstmt = null; - int returnVal = -1; - try { - cstmt = this.connection.prepareCall(xa_end_post_816); - - cstmt.registerOutParameter(1, 2); - cstmt.setInt(2, xid.getFormatId()); - cstmt.setBytes(3, xid.getGlobalTransactionId()); - cstmt.setBytes(4, xid.getBranchQualifier()); - cstmt.setInt(5, flag); - cstmt.execute(); - - returnVal = cstmt.getInt(1); - } catch (SQLException s) { - returnVal = s.getErrorCode(); - throw s; - } finally { - try { - if (cstmt != null) cstmt.close(); - } catch (SQLException s) { - } - cstmt = null; - } - - return returnVal; - } - - public void forget(Xid xid) throws XAException { - synchronized (this.connection) { - int returnVal = 0; - - if (xid == null) { - throw new XAException(-5); - } - - removeXidFromList(xid); - try { - returnVal = doForget(xid); - } catch (SQLException sqle) { - checkError(sqle, -3); - } - - checkError(returnVal); - } - } - - protected int doForget(Xid xid) throws XAException, SQLException { - - int returnVal = 0; - CallableStatement cstmt = null; - try { - cstmt = this.connection.prepareCall(xa_forget_post_816); - - cstmt.registerOutParameter(1, 2); - cstmt.setInt(2, xid.getFormatId()); - cstmt.setBytes(3, xid.getGlobalTransactionId()); - cstmt.setBytes(4, xid.getBranchQualifier()); - - cstmt.execute(); - - returnVal = cstmt.getInt(1); - } catch (SQLException s) { - returnVal = s.getErrorCode(); - - if (returnVal == 0) { - throw new XAException(-6); - } - - throw s; - } finally { - try { - if (cstmt != null) cstmt.close(); - } catch (SQLException s) { - } - cstmt = null; - } - - return returnVal; - } - - @Override - public boolean isSameRM(XAResource xares) throws XAException { - Connection conn1 = null; - if ((xares instanceof DruidXAOracleResource)) { - conn1 = ((DruidXAOracleResource) xares).getConnection(); - } else { - return false; - } - try { - if ((this.connection == null) || (((OracleConnection) this.connection).isClosed())) { - return false; - } - String l_url = ((OracleConnection) this.connection).getURL(); - String l_prt = ((OracleConnection) this.connection).getProtocolType(); - - if (conn1 != null) { - return (conn1.equals(this.connection)) || (((OracleConnection) conn1).getURL().equals(l_url)) - || ((((OracleConnection) conn1).getProtocolType().equals(l_prt)) && (l_prt.equals("kprb"))); - } - } catch (SQLException sqe) { - throw new XAException(-3); - } - - return false; - } - - @Override - public int prepare(Xid xid) throws XAException { - - synchronized (this.connection) { - int returnVal = 0; - - if (xid == null) { - throw new XAException(-5); - } - - Xid stackedXid = null; - if (this.needStackingForCommitRollbackPrepare) { - stackedXid = super.suspendStacked(xid); - } - try { - try { - returnVal = doPrepare(xid); - if ((returnVal != 0) && (returnVal != 3)) { - - int x_e = DruidXAOracleException.errorConvert(returnVal); - - if ((x_e != 0) && (x_e != 3)) { - XAException ex = DruidXAOracleException.newXAException(getConnectionDuringExceptionHandling(), - returnVal); - ex.fillInStackTrace(); - throw ex; - } - - returnVal = x_e; - } - - } catch (SQLException sqle) { - checkError(sqle, -3); - } - } catch (XAException xaex) { - if (xaex.errorCode == -7) { - try { - this.connection.close(); - } catch (SQLException ea) { - // - } - - } else if (this.needStackingForCommitRollbackPrepare) { - super.resumeStacked(stackedXid); - } - throw xaex; - } - - if (this.needStackingForCommitRollbackPrepare) { - super.resumeStacked(stackedXid); - } - - return returnVal; - } - } - - protected int doPrepare(Xid xid) throws XAException, SQLException { - - int returnVal = 0; - CallableStatement cstmt = null; - try { - cstmt = this.connection.prepareCall(xa_prepare_post_816); - - cstmt.registerOutParameter(1, 2); - cstmt.setInt(2, xid.getFormatId()); - cstmt.setBytes(3, xid.getGlobalTransactionId()); - cstmt.setBytes(4, xid.getBranchQualifier()); - - cstmt.execute(); - - returnVal = cstmt.getInt(1); - } catch (SQLException s) { - int errorCode = s.getErrorCode(); - - if (errorCode == 0) { - throw new XAException(-6); - } - - throw s; - } finally { - try { - if (cstmt != null) cstmt.close(); - } catch (SQLException s) { - } - cstmt = null; - } - - return returnVal; - } - - protected OracleConnection getConnectionDuringExceptionHandling() { - return (OracleConnection) this.connection; - } - - public Connection getConnection() { - return this.connection; - } - - @Override - public Xid[] recover(int flag) throws XAException { - synchronized (this.connection) { - - if ((flag & 0x1800000) != flag) { - - throw new XAException(-5); - } - - if (flag == 16777216) { - this.isTMRScanStarted = true; - } else { - Xid[] arrayOfXid1; - if ((this.isTMRScanStarted) && (flag == 8388608)) { - this.isTMRScanStarted = false; - arrayOfXid1 = NO_XID; - - return arrayOfXid1; - } - if ((this.isTMRScanStarted) && (flag == 0)) { - arrayOfXid1 = NO_XID; - - return arrayOfXid1; - } - } - Statement stmt = null; - ResultSet rset = null; - ArrayList xidCollection = new ArrayList(50); - try { - stmt = this.connection.createStatement(); - try { - stmt.execute(this.synchronizeBeforeRecoverNewCall); - } catch (Exception exc) { - stmt.execute(this.synchronizeBeforeRecoverOldCall); - } - - rset = stmt.executeQuery(this.recoverySqlRows); - - while (rset.next()) { - xidCollection.add(new DruidXid(rset.getInt(1), rset.getBytes(2), rset.getBytes(3))); - } - - } catch (SQLException sqe) { - throw new XAException(-3); - } finally { - try { - if (stmt != null) { - stmt.close(); - } - if (rset != null) rset.close(); - } catch (Exception ignore) { - } - } - int xidSize = xidCollection.size(); - Xid[] xids = new Xid[xidSize]; - System.arraycopy(xidCollection.toArray(), 0, xids, 0, xidSize); - - Xid[] arrayOfXid2 = xids; - - return arrayOfXid2; - } - } - - @Override - public void rollback(Xid xid) throws XAException { - - synchronized (this.connection) { - int returnVal = 0; - - if (xid == null) { - throw new XAException(-5); - } - - Xid stackedXid = null; - if (this.needStackingForCommitRollbackPrepare) { - stackedXid = super.suspendStacked(xid); - } else { - removeXidFromList(xid); - - if (this.activeXid == null) exitGlobalTxnMode(); - } - try { - doRollback(xid); - } catch (SQLException sqle) { - checkError(sqle, -3); - } - - if (this.needStackingForCommitRollbackPrepare) { - super.resumeStacked(stackedXid); - } - - checkError(returnVal); - } - } - - protected void doRollback(Xid xid) throws XAException, SQLException { - - CallableStatement cstmt = null; - try { - cstmt = this.connection.prepareCall(xa_rollback_post_816); - - cstmt.registerOutParameter(1, 2); - cstmt.setInt(2, xid.getFormatId()); - cstmt.setBytes(3, xid.getGlobalTransactionId()); - cstmt.setBytes(4, xid.getBranchQualifier()); - - cstmt.execute(); - - int returnVal = cstmt.getInt(1); - - checkError(returnVal, -7); - } catch (SQLException s) { - int errorCode = s.getErrorCode(); - - if (errorCode == 0) { - throw new XAException(-6); - } - - throw s; - } finally { - try { - if (cstmt != null) cstmt.close(); - } catch (SQLException s) { - } - cstmt = null; - } - } - - @Override - public void start(Xid xid, int flag) throws XAException { - synchronized (this.connection) { - int returnVal = -1; - try { - if (xid == null) { - throw new XAException(-5); - } - - int isolFlag = flag & 0xFF00; - - flag &= -65281; - - int otherFlag = flag & 0x10000 | (this.isTransLoose ? 65536 : 0); - - flag &= -65537; - - if (((flag & 0x8200002) != flag) || ((otherFlag != 0) && ((otherFlag & 0x10000) != 65536))) { - throw new XAException(-5); - } - - if (((isolFlag & 0xFF00) != 0) && (isolFlag != 256) && (isolFlag != 512) && (isolFlag != 1024)) { - throw new XAException(-5); - } - - if (((flag & 0x8200000) != 0) && (((isolFlag & 0xFF00) != 0) || ((otherFlag & 0x10000) != 0))) { - - throw new XAException(-5); - } - - flag |= isolFlag | otherFlag; - - saveAndAlterAutoCommitModeForGlobalTransaction(); - try { - returnVal = doStart(xid, flag); - } catch (SQLException sqle) { - checkError(sqle, -3); - } - - checkError(returnVal); - - boolean[] isLocallySuspended = { false }; - createOrUpdateXid(xid, false, isLocallySuspended); - } catch (XAException ea) { - restoreAutoCommitModeForGlobalTransaction(); - - throw ea; - } - } - } - - protected void enterGlobalTxnMode() { - ((OracleConnection) this.connection).setTxnMode(1); - } - - protected void exitGlobalTxnMode() { - ((OracleConnection) this.connection).setTxnMode(0); - } - - protected int doStart(Xid xid, int flag) throws XAException, SQLException { - - int returnVal = -1; - CallableStatement cstmt = null; - try { - cstmt = this.connection.prepareCall(xa_start_post_816); - - cstmt.registerOutParameter(1, 2); - cstmt.setInt(2, xid.getFormatId()); - cstmt.setBytes(3, xid.getGlobalTransactionId()); - cstmt.setBytes(4, xid.getBranchQualifier()); - cstmt.setInt(5, this.timeout); - cstmt.setInt(6, flag); - - cstmt.execute(); - - returnVal = cstmt.getInt(1); - } catch (SQLException s) { - returnVal = s.getErrorCode(); - - if (returnVal == 0) { - throw new XAException(-6); - } - - throw s; - } finally { - try { - if (cstmt != null) cstmt.close(); - } catch (SQLException s) { - } - cstmt = null; - } - - return returnVal; - } - -} diff --git a/src/main/java/com/alibaba/druid/pool/xa/DruidXAPooledConnection.java b/src/main/java/com/alibaba/druid/pool/xa/DruidXAPooledConnection.java deleted file mode 100644 index 7893c8addc..0000000000 --- a/src/main/java/com/alibaba/druid/pool/xa/DruidXAPooledConnection.java +++ /dev/null @@ -1,26 +0,0 @@ -package com.alibaba.druid.pool.xa; - -import java.sql.SQLException; - -import javax.sql.XAConnection; -import javax.transaction.xa.XAResource; - -import com.alibaba.druid.pool.DruidConnectionHolder; -import com.alibaba.druid.pool.DruidPooledConnection; - -public class DruidXAPooledConnection extends DruidPooledConnection implements XAConnection { - - protected XAResource xaResource = null; - - public DruidXAPooledConnection(DruidConnectionHolder holder){ - super(holder); - // TODO Auto-generated constructor stub - } - - @Override - public XAResource getXAResource() throws SQLException { - // TODO Auto-generated method stub - return null; - } - -} diff --git a/src/main/java/com/alibaba/druid/pool/xa/DruidXAResource.java b/src/main/java/com/alibaba/druid/pool/xa/DruidXAResource.java deleted file mode 100644 index 7988f3dcad..0000000000 --- a/src/main/java/com/alibaba/druid/pool/xa/DruidXAResource.java +++ /dev/null @@ -1,205 +0,0 @@ -package com.alibaba.druid.pool.xa; - -import java.sql.Connection; -import java.sql.SQLException; -import java.util.Hashtable; - -import javax.transaction.xa.XAException; -import javax.transaction.xa.XAResource; -import javax.transaction.xa.Xid; - -import com.alibaba.druid.support.logging.Log; -import com.alibaba.druid.support.logging.LogFactory; - -public abstract class DruidXAResource implements XAResource { - - public final static Log LOG = LogFactory.getLog(DruidXAResource.class); - - public static final int XA_OK = 0; - public static final short DEFAULT_XA_TIMEOUT = 60; - protected boolean savedConnectionAutoCommit = false; - protected boolean savedXAConnectionAutoCommit = false; - public static final int TMNOFLAGS = 0; - public static final int TMNOMIGRATE = 2; - public static final int TMENDRSCAN = 8388608; - public static final int TMFAIL = 536870912; - public static final int TMMIGRATE = 1048576; - public static final int TMJOIN = 2097152; - public static final int TMONEPHASE = 1073741824; - public static final int TMRESUME = 134217728; - public static final int TMSTARTRSCAN = 16777216; - public static final int TMSUCCESS = 67108864; - public static final int TMSUSPEND = 33554432; - public static final int ORATMREADONLY = 256; - public static final int ORATMREADWRITE = 512; - public static final int ORATMSERIALIZABLE = 1024; - public static final int ORAISOLATIONMASK = 65280; - public static final int ORATRANSLOOSE = 65536; - - protected Connection connection = null; - protected DruidXAPooledConnection xaconnection = null; - protected int timeout = 60; - protected String dblink = null; - - protected Hashtable xidHash = new Hashtable(50); - - Xid lastActiveXid = null; - - protected Xid activeXid = null; - - protected boolean canBeMigratablySuspended = false; - - protected boolean isTMRScanStarted = false; - - protected static final Xid[] NO_XID = new Xid[0]; - - public int getTransactionTimeout() throws XAException { - return this.timeout; - } - - public boolean setTransactionTimeout(int seconds) throws XAException { - - if (seconds < 0) { - throw new XAException(-5); - } - - this.timeout = seconds; - - return true; - } - - protected void saveAndAlterAutoCommitModeForGlobalTransaction() throws XAException { - try { - this.savedConnectionAutoCommit = this.connection.getAutoCommit(); - this.connection.setAutoCommit(false); - this.savedXAConnectionAutoCommit = this.connection.getAutoCommit(); - this.xaconnection.setAutoCommit(false); // TODO xaconnection - } catch (SQLException ex) { - LOG.error(ex.getMessage(), ex); - } - - } - - protected final synchronized boolean removeXidFromList(Xid xid) { - if (isSameXid(this.activeXid, xid)) { - this.activeXid = null; - } - - return this.xidHash.remove(xid) != null; - } - - protected void resumeStacked(Xid xid) throws XAException { - if (xid != null) { - start(xid, 134217728); - this.activeXid = xid; - } - - } - - protected Xid suspendStacked(Xid xid) throws XAException { - - Xid stackedXid = null; - - if ((this.activeXid != null) && (!isSameXid(this.activeXid, xid))) { - stackedXid = this.activeXid; - - if (!isXidSuspended(this.activeXid)) { - end(this.activeXid, 33554432); - this.lastActiveXid = this.activeXid; - this.activeXid = null; - } - - } - - Xid localXid1 = stackedXid; - return localXid1; - } - - protected boolean isXidSuspended(Xid xid) throws XAException { - - boolean returnValue = false; - XidListEntry x = getMatchingXidListEntry(xid); - - if (x != null) { - returnValue = x.isSuspended; - } - - return returnValue; - } - - protected void checkError(int error) throws XAException { - // TODO - } - - protected void checkError(int error, int defaultXAError) throws XAException { - // TODO - } - - protected void checkError(SQLException sqlex, int defaultXAError) throws XAException { - // TODO - } - - final boolean isSameXid(Xid xid1, Xid xid2) { - return xid1 == xid2; - } - - protected synchronized void createOrUpdateXid(Xid xid, boolean isSuspended, boolean[] isLocallySuspended) { - - XidListEntry x = getMatchingXidListEntry(xid); - - if (x != null) { - isLocallySuspended[0] = true; - - x.isSuspended = isSuspended; - } else { - x = new XidListEntry(xid, isSuspended); - this.xidHash.put(xid, x); - } - - if (isSuspended) { - this.lastActiveXid = this.activeXid; - this.activeXid = null; - } else { - enterGlobalTxnMode(); - - if ((this.lastActiveXid != null) && (isSameXid(xid, this.lastActiveXid))) { - this.lastActiveXid = null; - } - this.activeXid = x.xid; - } - } - - protected void restoreAutoCommitModeForGlobalTransaction() throws XAException { - if (this.savedConnectionAutoCommit) { - try { - this.connection.setAutoCommit(this.savedConnectionAutoCommit); - this.xaconnection.setAutoCommit(this.savedXAConnectionAutoCommit); // TODO xaconn - } catch (SQLException ignoreException) { - - } - - } - } - - abstract void enterGlobalTxnMode(); - - final synchronized XidListEntry getMatchingXidListEntry(Xid xid) { - - return (XidListEntry) this.xidHash.get(xid); - } - - protected final synchronized boolean isXidListEmpty() { - return this.xidHash.isEmpty(); - } - - class XidListEntry { - - Xid xid; - boolean isSuspended; - - XidListEntry(Xid x, boolean s){ - this.xid = x; - this.isSuspended = s; - } - } -} diff --git a/src/main/java/com/alibaba/druid/pool/xa/DruidXid.java b/src/main/java/com/alibaba/druid/pool/xa/DruidXid.java deleted file mode 100644 index 068b87e126..0000000000 --- a/src/main/java/com/alibaba/druid/pool/xa/DruidXid.java +++ /dev/null @@ -1,82 +0,0 @@ -package com.alibaba.druid.pool.xa; - -import javax.transaction.xa.XAException; -import javax.transaction.xa.Xid; - -public class DruidXid implements Xid { - - private int formatId; - private byte[] gtrid = null; - private byte[] bqual = null; - private byte[] txctx = null; - public static final int MAXGTRIDSIZE = 64; - public static final int MAXBQUALSIZE = 64; - private int state; - - public DruidXid(int fId, byte[] gId, byte[] bId) throws XAException{ - this(fId, gId, bId, null); - } - - public DruidXid(int fId, byte[] gId, byte[] bId, byte[] context) throws XAException{ - - this.formatId = fId; - - if ((gId != null) && (gId.length > 64)) { - throw new XAException(-4); - } - - this.gtrid = gId; - - if ((bId != null) && (bId.length > 64)) { - throw new XAException(-4); - } - - this.bqual = bId; - this.txctx = context; - this.state = 0; - } - - public void setState(int k) { - this.state = k; - } - - public int getState() { - return this.state; - } - - public int getFormatId() { - return this.formatId; - } - - public byte[] getGlobalTransactionId() { - return this.gtrid; - } - - public byte[] getBranchQualifier() { - return this.bqual; - } - - public byte[] getTxContext() { - return this.txctx; - } - - public void setTxContext(byte[] context) { - this.txctx = context; - } - - public static final boolean isLocalTransaction(Xid xid) { - - byte[] gtrid = xid.getGlobalTransactionId(); - - if (gtrid == null) { - return true; - } - for (int i = 0; i < gtrid.length; i++) { - if (gtrid[i] == 0) { - continue; - } - return false; - } - return true; - } -} diff --git a/src/main/java/com/alibaba/druid/util/MySqlUtils.java b/src/main/java/com/alibaba/druid/util/MySqlUtils.java new file mode 100644 index 0000000000..7f0e97ea99 --- /dev/null +++ b/src/main/java/com/alibaba/druid/util/MySqlUtils.java @@ -0,0 +1,13 @@ +package com.alibaba.druid.util; + +import java.sql.Connection; +import java.sql.SQLException; + +import com.mysql.jdbc.jdbc2.optional.MysqlXAConnection; + +public class MySqlUtils { + + public static MysqlXAConnection createXAConnection(Connection physicalConn) throws SQLException { + return new MysqlXAConnection((com.mysql.jdbc.ConnectionImpl) physicalConn, false); + } +} diff --git a/src/main/java/com/alibaba/druid/util/OracleUtils.java b/src/main/java/com/alibaba/druid/util/OracleUtils.java index d4c89e1351..139f5e949c 100644 --- a/src/main/java/com/alibaba/druid/util/OracleUtils.java +++ b/src/main/java/com/alibaba/druid/util/OracleUtils.java @@ -5,10 +5,13 @@ import java.sql.PreparedStatement; import java.sql.ResultSet; import java.sql.SQLException; +import javax.transaction.xa.XAException; + import oracle.jdbc.OracleConnection; import oracle.jdbc.OracleResultSet; import oracle.jdbc.OracleStatement; import oracle.jdbc.internal.OraclePreparedStatement; +import oracle.jdbc.xa.client.OracleXAConnection; import oracle.sql.ROWID; import com.alibaba.druid.pool.DruidPooledConnection; @@ -16,6 +19,10 @@ import com.alibaba.druid.pool.DruidPooledPreparedStatement; public class OracleUtils { + public static OracleXAConnection OracleXAConnection(Connection oracleConnection) throws XAException { + return new OracleXAConnection(oracleConnection); + } + public static void clearDefines(DruidPooledPreparedStatement stmt) throws SQLException { OracleStatement oracleStmt = stmt.unwrap(OracleStatement.class); oracleStmt.clearDefines(); diff --git a/src/main/java/com/alibaba/druid/util/PGUtils.java b/src/main/java/com/alibaba/druid/util/PGUtils.java new file mode 100644 index 0000000000..52ba89b66a --- /dev/null +++ b/src/main/java/com/alibaba/druid/util/PGUtils.java @@ -0,0 +1,14 @@ +package com.alibaba.druid.util; + +import java.sql.Connection; +import java.sql.SQLException; + +import org.postgresql.core.BaseConnection; +import org.postgresql.xa.PGXAConnection; + +public class PGUtils { + + public static PGXAConnection createXAConnection(Connection physicalConn) throws SQLException { + return new PGXAConnection((BaseConnection) physicalConn); + } +}