Java 类javax.transaction.xa.XAResource 实例源码

private void testXARecover(CloudSpannerXAConnection xaConnection) throws SQLException, XAException
{
    log.info("Started XA recover transaction test");
    testXATransaction(xaConnection, CommitMode.None);
    Xid[] xids = xaConnection.recover(XAResource.TMSTARTRSCAN);
    Assert.assertEquals(1, xids.length);
    xaConnection.commit(xids[0], false);
    boolean found = false;
    try (ResultSet rs = xaConnection.getConnection().createStatement()
            .executeQuery("select * from test where id=1000000"))
    {
        if (rs.next())
            found = true;
    }
    Assert.assertTrue(found);
    log.info("Finished XA recover transaction test");
}

public void testBug46925() throws Exception {
    MysqlXADataSource xads1 = new MysqlXADataSource();
    MysqlXADataSource xads2 = new MysqlXADataSource();

    Xid txid = new MysqlXid(new byte[] { 0x1 }, new byte[] { 0xf }, 3306);

    xads1.setPinGlobalTxToPhysicalConnection(true);
    xads1.setUrl(dbUrl);

    xads2.setPinGlobalTxToPhysicalConnection(true);
    xads2.setUrl(dbUrl);

    XAConnection c1 = xads1.getXAConnection();
    assertTrue(c1 instanceof SuspendableXAConnection);
    // start a transaction on one connection
    c1.getXAResource().start(txid, XAResource.TMNOFLAGS);
    c1.getXAResource().end(txid, XAResource.TMSUCCESS);

    XAConnection c2 = xads2.getXAConnection();
    assertTrue(c2 instanceof SuspendableXAConnection);
    // prepare on another one. Since we are using a "pinned" connection we should have the same "currentXAConnection" for both SuspendableXAConnection
    c2.getXAResource().prepare(txid); // this will fail without the fix.
    c2.getXAResource().commit(txid, false);
}

public void start(Xid xid, int flag) throws XAException {
    if (flag == XAResource.TMNOFLAGS) {
        // first time in this transaction
        if (this.xid != null) {
            throw new XAException("already enlisted");
        }
        this.xid = xid;
        try {
            localTransaction.begin();
        } catch (ResourceException e) {
            throw (XAException) new XAException("could not start local tx").initCause(e);
        }
    } else if (flag == XAResource.TMRESUME) {
        if (xid != this.xid) {
            throw new XAException("attempting to resume in different transaction");
        }
    } else {
        throw new XAException("unknown state");
    }
}

public void testBug46925() throws Exception {
    MysqlXADataSource xads1 = new MysqlXADataSource();
    MysqlXADataSource xads2 = new MysqlXADataSource();

    Xid txid = new MysqlXid(new byte[] { 0x1 }, new byte[] { 0xf }, 3306);

    xads1.setPinGlobalTxToPhysicalConnection(true);
    xads1.setUrl(dbUrl);

    xads2.setPinGlobalTxToPhysicalConnection(true);
    xads2.setUrl(dbUrl);

    XAConnection c1 = xads1.getXAConnection();
    assertTrue(c1 instanceof SuspendableXAConnection);
    // start a transaction on one connection
    c1.getXAResource().start(txid, XAResource.TMNOFLAGS);
    c1.getXAResource().end(txid, XAResource.TMSUCCESS);

    XAConnection c2 = xads2.getXAConnection();
    assertTrue(c2 instanceof SuspendableXAConnection);
    // prepare on another one. Since we are using a "pinned" connection we should have the same "currentXAConnection" for both SuspendableXAConnection
    c2.getXAResource().prepare(txid); // this will fail without the fix.
    c2.getXAResource().commit(txid, false);
}

private int rollbackXaTranInner(List<Map> xaList){
    boolean flag=true;
    for(Map xaInfo:xaList){
        XAResource xaResource=(XAResource) xaInfo.get("xaResource");
        MyXid myXid=(MyXid) xaInfo.get("myXid");
        try {
            xaResource.rollback(myXid);
        } catch (XAException e) {
            flag=false;
            e.printStackTrace();
        }
    }
    if(flag==false){
        return 1;
    }
    return 0;       
}

@Override
public void associate(Connection connection) throws SQLException {
    try {
        if ( transactionRunning() ) {
            boolean newEnlistment = transactionSynchronizationRegistry.getResource( key ) == null;
            transactionSynchronizationRegistry.registerInterposedSynchronization( new InterposedSynchronization( connection ) );

            if ( newEnlistment ) {
                transactionSynchronizationRegistry.putResource( key, connection );
                XAResource resource = connectable ? new ConnectableLocalXAResource( (TransactionAware) connection, jndiName ) : new LocalXAResource( (TransactionAware) connection, jndiName );
                transactionManager.getTransaction().enlistResource( resource );
            }
            else {
                ( (TransactionAware) connection ).transactionStart();
            }
        } else {
            ( (TransactionAware) connection ).transactionCheckCallback( this::transactionRunning );
        }
    } catch ( Exception e ) {
        throw new SQLException( "Exception in association of connection to existing transaction", e );
    }
}

/**
 * Implementation of call back function from ConnectionEventListener interface. This callback will
 * be invoked on connection close event.
 * 
 * @param event Connection event object
 */
public void connectionClosed(ConnectionEvent event) {
  if (isActive) {
    try {
      XAConnection conn = (XAConnection) event.getSource();
      XAResource xar = (XAResource) xaResourcesMap.get(conn);
      xaResourcesMap.remove(conn);
      Transaction txn = transManager.getTransaction();
      if (txn != null && xar != null)
        txn.delistResource(xar, XAResource.TMSUCCESS);
      provider.returnConnection(conn);
    } catch (Exception e) {
      String exception =
          "GemFireTransactionDataSource::connectionClosed: Exception occured due to " + e;
      if (logger.isDebugEnabled()) {
        logger.debug(exception, e);
      }
    }
  }
}

/**
 *  
 */
void registerTranxConnection(XAConnection xaConn) throws Exception {
  try {
    synchronized (this) {
      if (transManager == null) {
        transManager = JNDIInvoker.getTransactionManager();
      }
    }
    Transaction txn = transManager.getTransaction();
    if (txn != null) {
      XAResource xar = xaConn.getXAResource();
      txn.enlistResource(xar);
      // Add in the Map after successful registration of XAResource
      this.xaResourcesMap.put(xaConn, xar);
    }
  } catch (Exception ex) {
    Exception e = new Exception(
        LocalizedStrings.GemFireTransactionDataSource_GEMFIRETRANSACTIONDATASOURCEREGISTERTRANXCONNECTION_EXCEPTION_IN_REGISTERING_THE_XARESOURCE_WITH_THE_TRANSACTIONEXCEPTION_OCCURED_0
            .toLocalizedString(ex));
    e.initCause(ex);
    throw e;
  }
}

@Override
public XAResource getXAResource() {
    final PooledSessionHolder session;
    try {
        session = safeGetSessionHolder();
    } catch (JMSException e) {
        throw JMSExceptionSupport.createRuntimeException(e);
    }

    if (session.getSession() instanceof XASession) {
        return ((XASession) session.getSession()).getXAResource();
    }

    return null;
}

@Override
public boolean isSameRM(XAResource xares) throws XAException
{
    // This trivial implementation makes sure that the
    // application server doesn't try to use another connection
    // for prepare, commit and rollback commands.
    return xares == this;
}

@Test
public void testStart() throws SQLException, XAException
{
    CloudSpannerXAConnection subject = createSubject();
    ICloudSpannerConnection connection = subject.getConnection();
    assertTrue(connection.getAutoCommit());
    Xid xid = createXid();
    subject.start(xid, XAResource.TMNOFLAGS);
    assertNotNull(connection);
    assertFalse(connection.getAutoCommit());
    thrown.expect(CloudSpannerSQLException.class);
    connection.commit();
}

@Test
public void testEnd() throws SQLException, XAException
{
    CloudSpannerXAConnection subject = createSubject();
    Xid xid = createXid();
    subject.start(xid, XAResource.TMNOFLAGS);
    subject.end(xid, XAResource.TMSUCCESS);
}

@Test
public void testPrepare() throws SQLException, XAException
{
    CloudSpannerXAConnection subject = createSubject();
    Xid xid = createXid();
    subject.start(xid, XAResource.TMNOFLAGS);
    subject.end(xid, XAResource.TMSUCCESS);
    subject.prepare(xid);
}

@Test
public void testCommitOnePhase() throws SQLException, XAException
{
    CloudSpannerXAConnection subject = createSubject();
    Xid xid = createXid();
    subject.start(xid, XAResource.TMNOFLAGS);
    subject.end(xid, XAResource.TMSUCCESS);
    subject.commit(xid, true);
}

@Test
public void testCommitTwoPhase() throws SQLException, XAException
{
    CloudSpannerXAConnection subject = createSubject();
    Xid xid = createXid();
    subject.start(xid, XAResource.TMNOFLAGS);
    subject.end(xid, XAResource.TMSUCCESS);
    subject.prepare(xid);
    subject.commit(xid, false);
}

private void testXATransaction(CloudSpannerXAConnection xaConnection, CommitMode mode)
        throws SQLException, XAException
{
    log.info("Starting XA simple transaction test");
    Random rnd = new Random();
    Connection connection = xaConnection.getConnection();
    int id = rnd.nextInt();
    Xid xid = RecoveredXid.stringToXid(String.valueOf(id) + "_Z3RyaWQ=_YnF1YWw=");
    xaConnection.start(xid, XAResource.TMNOFLAGS);
    String sql = "insert into test (id, uuid, active, amount, description, created_date, last_updated) values (?, ?, ?, ?, ?, ?, ?)";
    PreparedStatement statement = connection.prepareStatement(sql);
    setParameterValues(statement, 1000000);
    statement.executeUpdate();
    xaConnection.end(xid, XAResource.TMSUCCESS);
    xaConnection.prepare(xid);
    if (mode != CommitMode.None)
    {
        xaConnection.commit(xid, mode == CommitMode.OnePhase);
    }

    if (mode != CommitMode.None)
    {
        boolean found = false;
        try (ResultSet rs = connection.createStatement().executeQuery("select * from test where id=1000000"))
        {
            if (rs.next())
                found = true;
        }
        Assert.assertTrue(found);
    }
    log.info("Finished XA simple transaction test");
}

private Xid prepareDeleteRow(CloudSpannerXAConnection xaConnection) throws SQLException, XAException
{
    Random rnd = new Random();
    Connection connection = xaConnection.getConnection();
    int id = rnd.nextInt();
    Xid xid = RecoveredXid.stringToXid(String.valueOf(id) + "_Z3RyaWQ=_YnF1YWw=");
    xaConnection.start(xid, XAResource.TMNOFLAGS);
    String sql = "delete from test where id=1000000";
    PreparedStatement statement = connection.prepareStatement(sql);
    statement.executeUpdate();
    xaConnection.end(xid, XAResource.TMSUCCESS);
    xaConnection.prepare(xid);

    return xid;
}

/**
 * Stub. See implementation comment in the method for why this is not
 * implemented yet.
 *
 * @return false.
 * @param xares XAResource
 * @throws XAException
 */
public boolean isSameRM(XAResource xares) throws XAException {

    if (!(xares instanceof JDBCXAResource)) {
        return false;
    }

    return xaDataSource == ((JDBCXAResource) xares).getXADataSource();
}

/**
 * Stub.  See implementation comment in the method for why this is
 * not implemented yet.
 *
 * @return false.
 */
public boolean isSameRM(XAResource xares) throws XAException {

    if (!(xares instanceof JDBCXAResource)) {
        return false;
    }

    return xaDataSource == ((JDBCXAResource) xares).getXADataSource();
}

/**
 * Prepare the transaction TODO: Store prepare results
 * 
 * @return - the tx code
 */
@SuppressWarnings("unchecked")
public int prepare() throws IndexerException
{
    boolean isPrepared = true;
    boolean isModified = false;
    Map<StoreRef, LuceneIndexer> indexers = (Map<StoreRef, LuceneIndexer>) AlfrescoTransactionSupport.getResource(indexersKey);
    if (indexers != null)
    {
        for (LuceneIndexer indexer : indexers.values())
        {
            try
            {
                isModified |= indexer.isModified();
                indexer.prepare();
            }
            catch (IndexerException e)
            {
                isPrepared = false;
                throw new IndexerException("Failed to prepare: requires rollback", e);
            }
        }
    }
    if (isPrepared)
    {
        if (isModified)
        {
            return XAResource.XA_OK;
        }
        else
        {
            return XAResource.XA_RDONLY;
        }
    }
    else
    {
        throw new IndexerException("Failed to prepare: requires rollback");
    }
}

public void start(Xid xid, int arg1) throws XAException {
    switchToXid(xid);

    if (arg1 != XAResource.TMJOIN) {
        this.currentXAResource.start(xid, arg1);

        return;
    }

    //
    // Emulate join, by using resume on the same physical connection
    //

    this.currentXAResource.start(xid, XAResource.TMRESUME);
}

@Override
public XAResource getXAResource() throws ResourceException {
    if (xaResource == null) {
        throw new NotSupportedException("XAResource not available");
    }
    return new XAResourceProxy();
}

@Override
public boolean enlistResource( final XAResource xaResource ) throws RollbackException, IllegalStateException, SystemException {
    if ( xaResource == null ) {
        return false;
    }
    return delegate.enlistResource(
                (XAResource) newProxyInstance(
                    currentThread().getContextClassLoader(),
                    new Class<?>[] { onePhaseXAResourceClass }, new DelegatingInvocationHandler( xaResource )
                )
            );
}

@Override
public boolean isSameRM(XAResource xaResource) throws XAException {
    XAResource xares = xaResource;
    if (xaResource instanceof AbstractManagedConnection<?, ?, ?, ?>.XAResourceProxy) {
        xares = ((AbstractManagedConnection<?, ?, ?, ?>.XAResourceProxy) xaResource).getXAResource();
    }
    return getXAResource().isSameRM(xares);
}

@Override
public void registerResource(ResourceFactory resource) {
    XATransactionalResource xatr = new XATransactionalResource(resource.getName()) {
        @Override
        protected XAResource refreshXAConnection() throws ResourceException {
            return resource.create();
        }
    };
    Configuration.addResource(xatr);
    resources.put(resource.getName(), resource);
    recoverables.put(resource, xatr);
}

@Test
public void testRecovery() throws Exception {
    CountDownLatch latch = new CountDownLatch(1);

    XADataSource xaDs = Mockito.mock(XADataSource.class);
    XAConnection xaCon = Mockito.mock(XAConnection.class);
    XAResource xaRes = Mockito.mock(XAResource.class);
    Connection con = Mockito.mock(Connection.class);

    Mockito.when(xaDs.getXAConnection()).thenReturn(xaCon);
    Mockito.when(xaCon.getXAResource()).thenReturn(xaRes);
    Mockito.when(xaCon.getConnection()).thenReturn(con);

    Mockito.when(xaRes.recover(anyInt())).thenAnswer(invocation -> {
        latch.countDown();
        return new Xid[0];
    });
    Mockito.when(con.getAutoCommit()).thenReturn(true);
    Mockito.when(con.isValid(anyInt())).thenReturn(true);

    DataSource ds = ManagedDataSourceBuilder.builder()
            .dataSource(xaDs)
            .transactionManager(tm)
            .name("h2")
            .build();
    assertNotNull(ds);
    assertTrue(latch.await(5, TimeUnit.SECONDS));
}

@Test
public void testRecovery() throws Exception {
    CountDownLatch latch = new CountDownLatch(1);

    XADataSource xaDs = Mockito.mock(XADataSource.class);
    XAConnection xaCon = Mockito.mock(XAConnection.class);
    XAResource xaRes = Mockito.mock(XAResource.class);
    Connection con = Mockito.mock(Connection.class);

    Mockito.when(xaDs.getXAConnection()).thenReturn(xaCon);
    Mockito.when(xaCon.getXAResource()).thenReturn(xaRes);
    Mockito.when(xaCon.getConnection()).thenReturn(con);

    Mockito.when(xaRes.recover(TMSTARTRSCAN)).thenAnswer(invocation -> {
        latch.countDown();
        return new Xid[0];
    });
    Mockito.when(con.getAutoCommit()).thenReturn(true);
    Mockito.when(con.isValid(anyInt())).thenReturn(true);

    DataSource ds = ManagedDataSourceBuilder.builder()
            .dataSource(xaDs)
            .transactionManager(tm)
            .name("h2")
            .build();
    assertNotNull(ds);
    assertTrue(latch.await(5, TimeUnit.SECONDS));
}

@Test
public void testRecovery() throws Exception {
    CountDownLatch latch = new CountDownLatch(1);

    XADataSource xaDs = Mockito.mock(XADataSource.class);
    XAConnection xaCon = Mockito.mock(XAConnection.class);
    XAResource xaRes = Mockito.mock(XAResource.class);
    Connection con = Mockito.mock(Connection.class);

    Mockito.when(xaDs.getXAConnection()).thenReturn(xaCon);
    Mockito.when(xaCon.getXAResource()).thenReturn(xaRes);
    Mockito.when(xaCon.getConnection()).thenReturn(con);

    Mockito.when(xaRes.recover(TMSTARTRSCAN + TMENDRSCAN)).thenAnswer(invocation -> {
        latch.countDown();
        return new Xid[0];
    });
    Mockito.when(con.getAutoCommit()).thenReturn(true);
    Mockito.when(con.isValid(anyInt())).thenReturn(true);

    DataSource ds = ManagedDataSourceBuilder.builder()
            .dataSource(xaDs)
            .transactionManager(tm)
            .name("h2")
            .build();
    assertNotNull(ds);
    assertTrue(latch.await(5, TimeUnit.SECONDS));
}

/**
 * Stub.  See implementation comment in the method for why this is
 * not implemented yet.
 *
 * @return false.
 */
public boolean isSameRM(XAResource xares) throws XAException {

    if (!(xares instanceof JDBCXAResource)) {
        return false;
    }

    return xaDataSource == ((JDBCXAResource) xares).getXADataSource();
}

public void start(Xid xid, int arg1) throws XAException {
    switchToXid(xid);

    if (arg1 != XAResource.TMJOIN) {
        this.currentXAResource.start(xid, arg1);

        return;
    }

    //
    // Emulate join, by using resume on the same physical connection
    //

    this.currentXAResource.start(xid, XAResource.TMRESUME);
}

public void start(Xid xid, int arg1) throws XAException {
    switchToXid(xid);

    if (arg1 != XAResource.TMJOIN) {
        this.currentXAResource.start(xid, arg1);

        return;
    }

    //
    // Emulate join, by using resume on the same physical connection
    //

    this.currentXAResource.start(xid, XAResource.TMRESUME);
}

/**
 * The standard JCA 1.5 version of {@code createEndpoint}.
 * <p>This implementation delegates to {@link #createEndpointInternal()},
 * initializing the endpoint's XAResource before the endpoint gets invoked.
 */
@Override
public MessageEndpoint createEndpoint(XAResource xaResource) throws UnavailableException {
    AbstractMessageEndpoint endpoint = createEndpointInternal();
    endpoint.initXAResource(xaResource);
    return endpoint;
}

public TransactionDelegate(XAResource xaResource) {
    if (xaResource == null) {
        if (transactionFactory != null && !transactionFactory.supportsResourceAdapterManagedTransactions()) {
            throw new IllegalStateException("ResourceAdapter-provided XAResource is required for " +
                    "transaction management. Check your ResourceAdapter's configuration.");
        }
    }
    this.xaResource = xaResource;
}

/**
 * Wrap each concrete endpoint instance with an AOP proxy,
 * exposing the message listener's interfaces as well as the
 * endpoint SPI through an AOP introduction.
 */
@Override
public MessageEndpoint createEndpoint(XAResource xaResource) throws UnavailableException {
    GenericMessageEndpoint endpoint = (GenericMessageEndpoint) super.createEndpoint(xaResource);
    ProxyFactory proxyFactory = new ProxyFactory(this.messageListener);
    DelegatingIntroductionInterceptor introduction = new DelegatingIntroductionInterceptor(endpoint);
    introduction.suppressInterface(MethodInterceptor.class);
    proxyFactory.addAdvice(introduction);
    return (MessageEndpoint) proxyFactory.getProxy();
}

public void start(Xid xid, int arg1) throws XAException {
    switchToXid(xid);

    if (arg1 != XAResource.TMJOIN) {
        this.currentXAResource.start(xid, arg1);

        return;
    }

    //
    // Emulate join, by using resume on the same physical connection
    //

    this.currentXAResource.start(xid, XAResource.TMRESUME);
}

/**
 * Stub. See implementation comment in the method for why this is not
 * implemented yet.
 *
 * @return false.
 * @param xares XAResource
 * @throws XAException
 */
public boolean isSameRM(XAResource xares) throws XAException {

    if (!(xares instanceof JDBCXAResource)) {
        return false;
    }

    return xaDataSource == ((JDBCXAResource) xares).getXADataSource();
}

/**
 * CallBack for Connection Error.
 * 
 * @param event ConnectionEvent
 */
public void connectionErrorOccurred(ConnectionEvent event) {
  if (isActive) {
    // If its an XAConnection
    ManagedConnection conn = (ManagedConnection) event.getSource();
    XAResource xar = (XAResource) xaResourcesMap.get(conn);
    xaResourcesMap.remove(conn);
    TransactionManagerImpl transManager = TransactionManagerImpl.getTransactionManager();
    try {
      Transaction txn = transManager.getTransaction();
      if (txn != null && xar != null)
        txn.delistResource(xar, XAResource.TMSUCCESS);
    } catch (SystemException se) {
      se.printStackTrace();
    }
    try {
      mannPoolCache.expirePooledConnection(conn);
      // mannPoolCache.destroyPooledConnection(conn);
    } catch (Exception ex) {
      String exception =
          "JCAConnectionManagerImpl::connectionErrorOccured: Exception occured due to " + ex;
      if (logger.isDebugEnabled()) {
        logger.debug(exception, ex);
      }
    }
  }
}

protected XAResource createXaResource(JmsPoolSession session) throws JMSException {
    return session.getXAResource();
}

@Override
public XAResource getXAResource() {
    return getSession().getXAResource();
}

@Override
public XAResource getXAResource()
{
    return this;
}