RE: help with jdbc transaction

Mario/Andrew,

 

I’ve have also had problems recently with JDBC and XA and I think that Mario will have other problems even he uses a XADataSource.

 

There appears to be a bug in JdbcConnector that means that the XADataSource never gets enlisted in the transaction. I’ve tracked this down to the following method in JdbcConnector:

 

public void setDataSource(DataSource dataSource)

{

                        this.dataSource = dataSource;

            }

 

In order to get XA working I’ve changed this method to:

 

public void setDataSource(DataSource dataSource)

             {                     

                        if (dataSource != null && dataSource instanceof XADataSource)

                        {

                                    if (MuleManager.getInstance().getTransactionManager() != null)

                                    {

                                                this.dataSource = new DataSourceWrapper((XADataSource)dataSource,

                                                            MuleManager.getInstance().getTransactionManager());

                                    }

                        }

                        else

                        {

                                    this.dataSource = dataSource;

                        }

             }

 

I also had to change the doReceive method in JdbcMessageDispatcher as it ignores the transaction at the minute:

 

    protected UMOMessage doReceive(UMOImmutableEndpoint endpoint, long timeout) throws Exception

    {

        if (logger.isDebugEnabled())

        {

            logger.debug("Trying to receive a message with a timeout of " + timeout);

        }

 

        String[] stmts = this.connector.getReadAndAckStatements(endpoint);

        String readStmt = stmts[0];

        String ackStmt = stmts[1];

        List readParams = new ArrayList();

        List ackParams = new ArrayList();

        readStmt = connector.parseStatement(readStmt, readParams);

        ackStmt = connector.parseStatement(ackStmt, ackParams);

 

        Connection con = null;

        long t0 = System.currentTimeMillis();

 

        // SM Get the transaction

        UMOTransaction tx = TransactionCoordination.getInstance().getTransaction();

        try

        {

            con = this.connector.getConnection();

            if (timeout < 0)

            {

                timeout = Long.MAX_VALUE;

            }

            Object result;

            do

            {

                result = connector.createQueryRunner().query(con, readStmt,

                    connector.getParams(endpoint, readParams, null), connector.createResultSetHandler());

                if (result != null)

                {

                    if (logger.isDebugEnabled())

                    {

                        logger.debug("Received: " + result);

                    }

                    break;

                }

                long sleep = Math.min(this.connector.getPollingFrequency(),

                    timeout - (System.currentTimeMillis() - t0));

                if (sleep > 0)

                {

                    if (logger.isDebugEnabled())

                    {

                        logger.debug("No results, sleeping for " + sleep);

                    }

                    Thread.sleep(sleep);

                }

                else

                {

                    logger.debug("Timeout");

                    return null;

                }

            }

            while (true);

            if (ackStmt != null)

            {

                int nbRows = connector.createQueryRunner().update(con, ackStmt,

                    connector.getParams(endpoint, ackParams, result));

                if (nbRows != 1)

                {

                    logger.warn("Row count for ack should be 1 and not " + nbRows);

                }

            }

            UMOMessageAdapter msgAdapter = this.connector.getMessageAdapter(result);

            UMOMessage message = new MuleMessage(msgAdapter);

           

            // SM Check if we’re in a transaction

            if (tx == null)

            {

                        JdbcUtils.commitAndClose(con);

            }

            return message;

        }

        catch (Exception e)

        {

            // SM Check if we’re in a transaction

            if (tx == null)

            {

                        JdbcUtils.rollbackAndClose(con);

            }

            throw e;

        }

    }

 

 

Stephen