[jboss-cvs] JBoss Messaging SVN: r2617 - trunk/tests/src/org/jboss/test/messaging/jms.

jboss-cvs-commits at lists.jboss.org jboss-cvs-commits at lists.jboss.org
Tue May 1 21:01:33 EDT 2007


Author: clebert.suconic at jboss.com
Date: 2007-05-01 21:01:33 -0400 (Tue, 01 May 2007)
New Revision: 2617

Modified:
   trunk/tests/src/org/jboss/test/messaging/jms/XATestBase.java
Log:
http://jira.jboss.org/jira/browse/JBMESSAGING-946 - fixing testcase due to new transaction behavior

Modified: trunk/tests/src/org/jboss/test/messaging/jms/XATestBase.java
===================================================================
--- trunk/tests/src/org/jboss/test/messaging/jms/XATestBase.java	2007-05-01 00:41:56 UTC (rev 2616)
+++ trunk/tests/src/org/jboss/test/messaging/jms/XATestBase.java	2007-05-02 01:01:33 UTC (rev 2617)
@@ -33,6 +33,11 @@
 import javax.jms.TextMessage;
 import javax.jms.XAConnection;
 import javax.jms.XASession;
+import javax.jms.MessageListener;
+import javax.jms.XAConnectionFactory;
+import javax.jms.ServerSessionPool;
+import javax.jms.ServerSession;
+import javax.jms.JMSException;
 import javax.management.ObjectName;
 import javax.naming.InitialContext;
 import javax.transaction.Transaction;
@@ -44,10 +49,14 @@
 
 import org.jboss.jms.client.JBossConnection;
 import org.jboss.jms.client.JBossConnectionFactory;
+import org.jboss.jms.client.JBossSession;
 import org.jboss.jms.client.delegate.ClientConnectionDelegate;
+import org.jboss.jms.client.delegate.DelegateSupport;
 import org.jboss.jms.client.state.ConnectionState;
+import org.jboss.jms.client.state.SessionState;
 import org.jboss.jms.tx.MessagingXAResource;
 import org.jboss.jms.tx.ResourceManager;
+import org.jboss.jms.tx.LocalTx;
 import org.jboss.messaging.core.tx.MessagingXid;
 import org.jboss.test.messaging.MessagingTestCase;
 import org.jboss.test.messaging.tools.ServerManagement;
@@ -55,11 +64,13 @@
 import org.jboss.test.messaging.tools.jndi.InVMInitialContextFactory;
 import org.jboss.tm.TransactionManagerLocator;
 import org.jboss.tm.TxUtils;
+import org.jboss.logging.Logger;
 
 import com.arjuna.ats.internal.jta.transaction.arjunacore.TransactionManagerImple;
+import java.util.ArrayList;
 
 /**
- * 
+ *
  * A XATestBase
  *
  * @author <a href="mailto:tim.fox at jboss.com">Tim Fox</a>
@@ -73,21 +84,21 @@
    // Constants -----------------------------------------------------
 
    // Static --------------------------------------------------------
-   
+
    // Attributes ----------------------------------------------------
 
    protected InitialContext initialContext;
-   
+
    protected JBossConnectionFactory cf;
-   
+
    protected Destination queue;
-   
+
    protected TransactionManager tm;
-   
+
    protected Transaction suspendedTx;
-   
+
    protected ServiceContainer sc;
-   
+
    protected boolean useArjuna;
 
 
@@ -97,23 +108,23 @@
    {
       super(name);
    }
-   
-   
+
+
    // TestCase overrides -------------------------------------------
 
    public void setUp() throws Exception
    {
       super.setUp();
-      
+
       if (useArjuna)
       {
          ServerManagement.start("all,-transaction,jbossjta");
       }
       else
-      {      
+      {
          ServerManagement.start("all");
       }
-      
+
       //Also need a local tx mgr if test is running remote
       if (ServerManagement.isRemote())
       {
@@ -125,19 +136,19 @@
          {
             sc = new ServiceContainer("transaction");
          }
-         
+
          //Don't drop the tables again!
          sc.start(false);
       }
-      
+
       initialContext = new InitialContext(ServerManagement.getJNDIEnvironment());
-      
+
       InitialContext localIc = new InitialContext(InVMInitialContextFactory.getJNDIEnvironment());
-      
+
       cf = (JBossConnectionFactory)initialContext.lookup("/ConnectionFactory");
-            
+
       tm = (TransactionManager)localIc.lookup(ServiceContainer.TRANSACTION_MANAGER_JNDI_NAME);
-      
+
       if (useArjuna)
       {
          log.info("tm is " + tm.getClass().getName());
@@ -147,13 +158,13 @@
       {
          assertTrue(tm instanceof org.jboss.tm.TxManager);
       }
-    
+
       ServerManagement.undeployQueue("Queue");
-      
+
       ServerManagement.deployQueue("Queue");
-      
+
       queue = (Destination)initialContext.lookup("/queue/Queue");
-      
+
       drainDestination(cf, queue);
 
       if (!ServerManagement.isRemote())
@@ -165,7 +176,7 @@
    public void tearDown() throws Exception
    {
       ServerManagement.undeployQueue("Queue");
-      
+
       if (TxUtils.isUncommitted(tm))
       {
          //roll it back
@@ -183,37 +194,46 @@
          Transaction tx = tm.suspend();
          if (tx != null)
             log.warn("Transaction still associated with thread " + tx + " at status " + TxUtils.getStatusAsString(tx.getStatus()));
-      }    
-      
+      }
+
       if (suspendedTx != null)
       {
          tm.resume(suspendedTx);
       }
-      
+
       if (ServerManagement.isRemote())
       {
          sc.stop();
       }
-      
+
       ServerManagement.stop();
 
       super.tearDown();
    }
-   
-   
 
 
+
+
    // Public --------------------------------------------------------
-   
-   
+
+
    /* If there is no global tx present the send must behave as non transacted
     * See http://www.jboss.com/index.html?module=bb&op=viewtopic&t=98577&postdays=0&postorder=asc&start=0
-    * http://jira.jboss.com/jira/browse/JBMESSAGING-410 
+    * http://jira.jboss.com/jira/browse/JBMESSAGING-410
     * http://jira.jboss.com/jira/browse/JBMESSAGING-721
     * http://jira.jboss.org/jira/browse/JBMESSAGING-946
     */
    public void testSendNoGlobalTransaction() throws Exception
    {
+
+      if (ServerManagement.isRemote())
+      {
+         log.info("************************************************************************************************************************");
+         log.info("testSendNoGlobalTransaction is being ignored on remote runs, as this requires access to the JCA layer on java:/");
+         log.info("************************************************************************************************************************");
+         return;
+      }
+
       Transaction suspended = null;
 
       try
@@ -269,13 +289,21 @@
 
    /* If there is no global tx present the send must behave as non transacted
     * See http://www.jboss.com/index.html?module=bb&op=viewtopic&t=98577&postdays=0&postorder=asc&start=0
-    * http://jira.jboss.com/jira/browse/JBMESSAGING-410 
+    * http://jira.jboss.com/jira/browse/JBMESSAGING-410
     * http://jira.jboss.com/jira/browse/JBMESSAGING-721
     * http://jira.jboss.org/jira/browse/JBMESSAGING-946
     */
    public void testSendNoGlobalTransaction2() throws Exception
    {
 
+      if (ServerManagement.isRemote())
+      {
+         log.info("************************************************************************************************************************");
+         log.info("testSendNoGlobalTransaction2 is being ignored on remote runs, as this requires access to the JCA layer on java:/");
+         log.info("************************************************************************************************************************");
+         return;
+      }
+
       Transaction suspended = TransactionManagerLocator.getInstance().locate().suspend();
 
       try
@@ -350,29 +378,284 @@
          }
       }
    }
-   
-   
+
    /*
     * If there is no global tx present messages consumed must consumed as if they were in a
     * local tx. Note this behaviour differs from messages sent
     * This is so we can support transacted delivery of messags in an MDB as mentioned
-    * 
+    *
+    * However to accomodate TCK tests, when Sessions/Consumers are being used without transaction
+    * enlist, we will process those cases as NonTransactional, AutoAck.
+    *
     * See http://www.jboss.com/index.html?module=bb&op=viewtopic&t=98577&postdays=0&postorder=asc&start=0
-    * http://jira.jboss.com/jira/browse/JBMESSAGING-410 
+    * http://jira.jboss.com/jira/browse/JBMESSAGING-410
     * http://jira.jboss.com/jira/browse/JBMESSAGING-721
     * http://jira.jboss.org/jira/browse/JBMESSAGING-946
-    * 
+    *
     * For transactional delivery the receipt of the message should be in a transaction but by the time
     * the mdb container is invoked the message has already been received it is too late - the message
     * has already been received and passed on (see page 199 (chapter 5 JMS and Transactions, section "Application Server Integration"
     * of Mark Little's book Java Transaction processing
     * for a discussion of how different app serves deal with this)
     * The way jboss messaging (and jboss mq) deals with this is to convert any work done
-    * prior to when the xasession is enlisted in the tx, into work done in the xa tx
+    * prior to when the xasession is enlisted in the tx, into work done in the xa tx.
     *
+    *      NOTE: To accomodate TCK tests where Session/Consumers are being used without transaction enlisting
+    *            we are processing those cases as nonTransactional/AutoACK, however if the session is being used
+    *            to process MDBs we will consider the LocalTransaction convertion and process those as the comment above
+    *            This was done as per: http://jira.jboss.org/jira/browse/JBMESSAGING-946
+    *
     */
+   public void testConnectionConsumer() throws Exception
+   {
+      ServerManagement.deployQueue("MyQueue2");
+
+
+      try
+      {
+         // send a message to the queue
+
+         ConnectionFactory cf = (ConnectionFactory) initialContext.lookup("/ConnectionFactory");
+         Queue queue = (Queue) initialContext.lookup("queue/MyQueue2");
+         Connection conn = cf.createConnection();
+         Session s = conn.createSession(false, Session.AUTO_ACKNOWLEDGE);
+         MessageProducer p = s.createProducer(queue);
+         p.setDeliveryMode(DeliveryMode.PERSISTENT);
+         Message m = s.createTextMessage("one");
+         p.send(m);
+         conn.close();
+
+         // make sure there's no active JTA transaction
+
+         Transaction suspended = tm.suspend();
+
+         XAConnection xaconn = null;
+         try
+         {
+
+            ObjectName queueMBean = new ObjectName("jboss.messaging.destination:service=Queue,name=MyQueue2");
+            Integer count = (Integer) ServerManagement.getAttribute(queueMBean, "MessageCount");
+            assertEquals(1, count.intValue());
+
+            // using XA with a ConnectionConsumer (testing the transaction behavior under MDBs)
+
+            XAConnectionFactory xacf = (XAConnectionFactory) cf;
+            xaconn = xacf.createXAConnection();
+            xaconn.start();
+            XASession xasession = xaconn.createXASession();
+            DummyListener listener = new DummyListener();
+            xasession.setMessageListener(listener);
+
+            ServerSessionPool pool = new MockServerSessionPool(xasession);
+
+            xaconn.createConnectionConsumer(queue, null, pool, 1);
+
+            Thread.sleep(1000);
+            assertEquals(1, listener.messages.size());
+
+            // Message should still be on server
+            count = (Integer) ServerManagement.getAttribute(queueMBean, "MessageCount");
+            assertEquals(1, count.intValue());
+
+            XAResource resource = xasession.getXAResource();
+
+            // Starts a new transaction
+            tm.begin();
+
+            Transaction trans = tm.getTransaction();
+
+            JBossSession session = (JBossSession)xasession;
+            SessionState state = (SessionState)((DelegateSupport)session.getDelegate()).getState();
+
+            // Validates TX convertion
+            assertTrue(state.getCurrentTxId() instanceof LocalTx);
+
+            // Enlist the transaction... as supposed to be happening on JBossAS with the
+            // default listener (enlist happening after message is received)
+            trans.enlistResource(resource);
+
+            // Validates TX convertion
+            assertFalse(state.getCurrentTxId() instanceof LocalTx);
+
+            trans.delistResource(resource, XAResource.TMSUCCESS);
+
+            trans.commit();
+
+
+            // After commit the message should be consumed
+            count = (Integer) ServerManagement.getAttribute(queueMBean, "MessageCount");
+            assertEquals(0, count.intValue());
+         }
+         finally
+         {
+            if (xaconn != null)
+            {
+               xaconn.close();
+            }
+            if (suspended != null)
+            {
+               TransactionManagerLocator.getInstance().locate().resume(suspended);
+            }
+         }
+      }
+      finally
+      {
+         ServerManagement.undeployQueue("MyQueue2");
+      }
+
+
+   }
+
+   /*
+    * If there is no global tx present messages consumed must consumed as if they were in a
+    * local tx. Note this behaviour differs from messages sent
+    * This is so we can support transacted delivery of messags in an MDB as mentioned
+    *
+    * However to accomodate TCK tests, when Sessions/Consumers are being used without transaction
+    * enlist, we will process those cases as NonTransactional, AutoAck.
+    *
+    * See http://www.jboss.com/index.html?module=bb&op=viewtopic&t=98577&postdays=0&postorder=asc&start=0
+    * http://jira.jboss.com/jira/browse/JBMESSAGING-410
+    * http://jira.jboss.com/jira/browse/JBMESSAGING-721
+    * http://jira.jboss.org/jira/browse/JBMESSAGING-946
+    *
+    * For transactional delivery the receipt of the message should be in a transaction but by the time
+    * the mdb container is invoked the message has already been received it is too late - the message
+    * has already been received and passed on (see page 199 (chapter 5 JMS and Transactions, section "Application Server Integration"
+    * of Mark Little's book Java Transaction processing
+    * for a discussion of how different app serves deal with this)
+    * The way jboss messaging (and jboss mq) deals with this is to convert any work done
+    * prior to when the xasession is enlisted in the tx, into work done in the xa tx.
+    *
+    *      NOTE: To accomodate TCK tests where Session/Consumers are being used without transaction enlisting
+    *            we are processing those cases as nonTransactional/AutoACK, however if the session is being used
+    *            to process MDBs we will consider the LocalTransaction convertion and process those as the comment above
+    *            This was done as per: http://jira.jboss.org/jira/browse/JBMESSAGING-946
+    *
+    */
+   public void testReceiveGlobalTransaction() throws Exception
+   {
+      XAConnection xaconn = null;
+
+      try
+      {
+         ServerManagement.deployQueue("MyQueue2");
+
+         // send a message to the queue
+
+         ConnectionFactory cf = (ConnectionFactory)initialContext.lookup("/ConnectionFactory");
+         Queue queue = (Queue)initialContext.lookup("queue/MyQueue2");
+         Connection conn = cf.createConnection();
+         Session s = conn.createSession(false, Session.AUTO_ACKNOWLEDGE);
+         MessageProducer p = s.createProducer(queue);
+         p.setDeliveryMode(DeliveryMode.PERSISTENT);
+         Message m = s.createTextMessage("one");
+         p.send(m);
+         conn.close();
+
+         ObjectName queueMBean = new ObjectName("jboss.messaging.destination:service=Queue,name=MyQueue2");
+         Integer count = (Integer)ServerManagement.getAttribute(queueMBean, "MessageCount");
+         assertEquals(1, count.intValue());
+
+         tm.begin();
+
+         Transaction trans = tm.getTransaction();
+
+         XAConnectionFactory xacf = (XAConnectionFactory)cf;
+
+         xaconn = xacf.createXAConnection();
+
+         xaconn.start();
+
+         XASession xasession = xaconn.createXASession();
+
+
+         XAResource resouce = xasession.getXAResource();
+         trans.enlistResource(resouce);
+
+         MessageConsumer consumer = xasession.createConsumer(queue);
+
+         TextMessage messageReceived = (TextMessage)consumer.receive(1000);
+
+         assertNotNull(messageReceived);
+
+         assertEquals("one", messageReceived.getText());
+
+         assertNull(consumer.receive(1000));
+
+         count = (Integer)ServerManagement.getAttribute(queueMBean, "MessageCount");
+         assertEquals(1, count.intValue());
+
+         trans.delistResource(resouce, XAResource.TMSUCCESS);
+
+         tm.rollback();
+
+         tm.begin();
+         trans = tm.getTransaction();
+         trans.enlistResource(resouce);
+
+         messageReceived = (TextMessage)consumer.receive(1000);
+
+         assertNotNull(messageReceived);
+
+         assertEquals("one", messageReceived.getText());
+
+         count = (Integer)ServerManagement.getAttribute(queueMBean, "MessageCount");
+         assertEquals(1, count.intValue());
+
+         trans.commit();
+
+         count = (Integer)ServerManagement.getAttribute(queueMBean, "MessageCount");
+         assertEquals(0, count.intValue());
+
+      }
+      finally
+      {
+         if (xaconn != null)
+         {
+            xaconn.close();
+         }
+         ServerManagement.undeployQueue("MyQueue2");
+      }
+   }
+
+   /*
+    * If there is no global tx present messages consumed must consumed as if they were in a
+    * local tx. Note this behaviour differs from messages sent
+    * This is so we can support transacted delivery of messags in an MDB as mentioned
+    *
+    * However to accomodate TCK tests, when Sessions/Consumers are being used without transaction
+    * enlist, we will process those cases as NonTransactional, AutoAck.
+    *
+    * See http://www.jboss.com/index.html?module=bb&op=viewtopic&t=98577&postdays=0&postorder=asc&start=0
+    * http://jira.jboss.com/jira/browse/JBMESSAGING-410
+    * http://jira.jboss.com/jira/browse/JBMESSAGING-721
+    * http://jira.jboss.org/jira/browse/JBMESSAGING-946
+    *
+    * For transactional delivery the receipt of the message should be in a transaction but by the time
+    * the mdb container is invoked the message has already been received it is too late - the message
+    * has already been received and passed on (see page 199 (chapter 5 JMS and Transactions, section "Application Server Integration"
+    * of Mark Little's book Java Transaction processing
+    * for a discussion of how different app serves deal with this)
+    * The way jboss messaging (and jboss mq) deals with this is to convert any work done
+    * prior to when the xasession is enlisted in the tx, into work done in the xa tx.
+    *
+    *      NOTE: To accomodate TCK tests where Session/Consumers are being used without transaction enlisting
+    *            we are processing those cases as nonTransactional/AutoACK, however if the session is being used
+    *            to process MDBs we will consider the LocalTransaction convertion and process those as the comment above
+    *            This was done as per: http://jira.jboss.org/jira/browse/JBMESSAGING-946
+    *
+    */
    public void testReceiveNoGlobalTransaction() throws Exception
    {
+      if (ServerManagement.isRemote())
+      {
+         log.info("************************************************************************************************************************");
+         log.info("testReceiveNoGlobalTransaction is being ignored on remote runs, as this requires access to the JCA layer on java:/");
+         log.info("************************************************************************************************************************");
+         return;
+      }
+
       try
       {
          ServerManagement.deployQueue("MyQueue2");
@@ -395,18 +678,24 @@
 
          try
          {
-            // using a JCA wrapper
 
+            ObjectName queueMBean = new ObjectName("jboss.messaging.destination:service=Queue,name=MyQueue2");
+            Integer count = (Integer)ServerManagement.getAttribute(queueMBean, "MessageCount");
+            assertEquals(1, count.intValue());
+
+            // using a JCA wrapper without a transaction
+
             ConnectionFactory mcf =
-               (ConnectionFactory)initialContext.lookup("java:/JCAConnectionFactory");
+                  (ConnectionFactory)initialContext.lookup("java:/JCAConnectionFactory");
             conn = mcf.createConnection();
             conn.start();
 
             // no active JTA transaction here
 
             s = conn.createSession(false, Session.AUTO_ACKNOWLEDGE);
+
             MessageConsumer c = s.createConsumer(queue);
-            
+
             // the messge should be store unacked in the local session
             TextMessage rm = (TextMessage)c.receive(1000);
 
@@ -414,10 +703,9 @@
 
             conn.close();
 
-            // the messsage should still be in the queue
-            ObjectName on = new ObjectName("jboss.messaging.destination:service=Queue,name=MyQueue2");
-            Integer count = (Integer)ServerManagement.getAttribute(on, "MessageCount");
-            assertEquals(1, count.intValue());
+            // messages should be consumed
+            count = (Integer)ServerManagement.getAttribute(queueMBean, "MessageCount");
+            assertEquals(0, count.intValue());
          }
          finally
          {
@@ -433,60 +721,60 @@
          ServerManagement.undeployQueue("MyQueue2");
       }
    }
-   
-   
+
+
    //See http://jira.jboss.com/jira/browse/JBMESSAGING-638
    public void testResourceManagerMemoryLeakOnCommit() throws Exception
    {
 
       XAConnection xaConn = null;
-      
+
       try
       {
          xaConn = cf.createXAConnection();
-         
+
          JBossConnection jbConn = (JBossConnection)xaConn;
-         
+
          ClientConnectionDelegate del = (ClientConnectionDelegate)jbConn.getDelegate();
-         
+
          ConnectionState state = (ConnectionState)del.getState();
-         
+
          ResourceManager rm = state.getResourceManager();
-         
+
          XASession xaSession = xaConn.createXASession();
-         
+
          xaConn.start();
-         
+
          XAResource res = xaSession.getXAResource();
-         
+
          XAResource dummy = new DummyXAResource();
-         
+
          for (int i = 0; i < 100; i++)
          {
-            
+
             tm.begin();
-                     
+
             Transaction tx = tm.getTransaction();
-            
+
             tx.enlistResource(res);
-            
+
             tx.enlistResource(dummy);
-            
+
             assertEquals(1, rm.size());
-            
+
             tx.delistResource(res, XAResource.TMSUCCESS);
-            
+
             tx.delistResource(dummy, XAResource.TMSUCCESS);
-            
+
             tm.commit();
-         }                  
-         
+         }
+
          assertEquals(1, rm.size());
-         
+
          xaConn.close();
-         
+
          xaConn = null;
-         
+
          assertEquals(0, rm.size());
 
       }
@@ -498,57 +786,57 @@
          }
       }
    }
-   
+
    //See http://jira.jboss.com/jira/browse/JBMESSAGING-638
    public void testResourceManagerMemoryLeakOnRollback() throws Exception
-   { 
+   {
       XAConnection xaConn = null;
-      
+
       try
       {
          xaConn = cf.createXAConnection();
-         
+
          JBossConnection jbConn = (JBossConnection)xaConn;
-         
+
          ClientConnectionDelegate del = (ClientConnectionDelegate)jbConn.getDelegate();
-         
+
          ConnectionState state = (ConnectionState)del.getState();
-         
+
          ResourceManager rm = state.getResourceManager();
-         
+
          XASession xaSession = xaConn.createXASession();
-         
+
          xaConn.start();
-         
+
          XAResource res = xaSession.getXAResource();
-         
+
          XAResource dummy = new DummyXAResource();
-         
+
          for (int i = 0; i < 100; i++)
-         {            
+         {
             tm.begin();
-                     
+
             Transaction tx = tm.getTransaction();
-            
+
             tx.enlistResource(res);
-            
+
             tx.enlistResource(dummy);
-            
+
             assertEquals(1, rm.size());
-            
+
             tx.delistResource(res, XAResource.TMSUCCESS);
-            
+
             tx.delistResource(dummy, XAResource.TMSUCCESS);
-            
+
             tm.rollback();
-         }                  
-         
+         }
+
          assertEquals(1, rm.size());
-         
+
          xaConn.close();
-         
+
          xaConn = null;
-         
+
          assertEquals(0, rm.size());
 
       }
@@ -560,70 +848,85 @@
          }
       }
    }
-   
 
-   //http://jira.jboss.com/jira/browse/JBMESSAGING-721
+
+   /**
+   * //http://jira.jboss.com/jira/browse/JBMESSAGING-721
+   *      NOTE: To accomodate TCK tests where Session/Consumers are being used without transaction enlisting
+   *            we are processing those cases as nonTransactional/AutoACK, however if the session is being used
+   *            to process MDBs we will consider the LocalTransaction convertion and process those as the comment above
+   *            This was done as per: http://jira.jboss.org/jira/browse/JBMESSAGING-946
+   *
+   */
+
    public void testConvertFromLocalTx() throws Exception
    {
       Connection conn = null;
-      
+
       XAConnection xaConn = null;
-      
+
+
       try
       {
-      
+
          //First send some messages to a queue
-         
+
+         ObjectName queueMBean = new ObjectName("jboss.messaging.destination:service=Queue,name=Queue");
+         Integer count = (Integer) ServerManagement.getAttribute(queueMBean, "MessageCount");
+         assertEquals(0, count.intValue());
+
          conn = cf.createConnection();
-         
+
          Session sessSend = conn.createSession(false, Session.AUTO_ACKNOWLEDGE);
-         
+
          MessageProducer prod = sessSend.createProducer(queue);
-         
+
          TextMessage tm1 = sessSend.createTextMessage("message1");
-         
+
          TextMessage tm2 = sessSend.createTextMessage("message2");
-         
+
          prod.send(tm1);
-         
+
          prod.send(tm2);
-         
-         
+
+         count = (Integer) ServerManagement.getAttribute(queueMBean, "MessageCount");
+         assertEquals(2, count.intValue());
+
          xaConn = cf.createXAConnection();
-         
+
          XASession xaSession = xaConn.createXASession();
-         
+
          xaConn.start();
-         
-         MessageConsumer cons = xaSession.createConsumer(queue);
-         
-         //Receive the two messages outside of a transaction
-         
-         TextMessage rm1 = (TextMessage)cons.receive(1000);
-         
-         assertNotNull(rm1);
-         
-         assertEquals("message1", rm1.getText());
-         
-         TextMessage rm2 = (TextMessage)cons.receive(1000);
-         
-         assertNotNull(rm2);
-         
-         assertEquals("message2", rm2.getText());
-         
-         Message rm3 = cons.receive(1000);
-         
-         assertNull(rm3);
-         
+
+         DummyListener listener = new DummyListener();
+
+         xaSession.setMessageListener(listener);
+
+         ServerSessionPool pool = new MockServerSessionPool(xaSession);
+
+         xaConn.createConnectionConsumer(queue, null, pool, 1);
+
+         Thread.sleep(1000);
+
+         assertEquals(2, listener.messages.size());
+
+         assertEquals("message1", ((TextMessage)(listener.messages.get(0))).getText());
+         assertEquals("message2", ((TextMessage)(listener.messages.get(1))).getText());
+
+         count = (Integer) ServerManagement.getAttribute(queueMBean, "MessageCount");
+         assertEquals(2, count.intValue());
+
+         listener.messages.clear();
+
          //Now we enlist the session in an xa transaction
-         
+
          XAResource res = xaSession.getXAResource();
-         
+
          tm.begin();
-         
+
          Transaction tx = tm.getTransaction();
          tx.enlistResource(res);
-         
+
          //This should cause the work done previously to be converted into work done in the xa transaction
          //this is what an MDB does
          //There is a difficulty in transactional delivery with an MDB.
@@ -636,147 +939,169 @@
          //for a discussion of how different app serves deal with this)
          //The way jboss messaging (and jboss mq) deals with this is to convert any work done
          //prior to when the xasession is enlisted in the tx, into work done in the xa tx
-         
+
          tx.delistResource(res, XAResource.TMSUCCESS);
-         
+
          //Now rollback the tx - this should cause redelivery of the two messages
-         tx.rollback();
-         
-         rm1 = (TextMessage)cons.receive(1000);
-         
-         assertNotNull(rm1);
-         
-         assertEquals("message1", rm1.getText());
-         
-         rm2 = (TextMessage)cons.receive(1000);
-         
-         assertNotNull(rm2);
-         
-         assertEquals("message2", rm2.getText());
-         
-         rm3 = cons.receive(1000);
-         
-         assertNull(rm3);
+         tm.rollback();
+
+         count = (Integer) ServerManagement.getAttribute(queueMBean, "MessageCount");
+         assertEquals(2, count.intValue());
+
+         Thread.sleep(1000);
+
+         assertEquals(2, listener.messages.size());
+
+         listener.messages.clear();
+
+         tm.begin();
+
+         tx = tm.getTransaction();
+         tx.enlistResource(res);
+
+         tm.commit();
+
+         Thread.sleep(1000);
+
+         assertEquals(0, listener.messages.size());
+
+         count = (Integer) ServerManagement.getAttribute(queueMBean, "MessageCount");
+         assertEquals(0, count.intValue());
+
+         assertNull(tm.getTransaction());
+
+
       }
       finally
-      {         
+      {
          if (conn != null)
          {
             conn.close();
          }
-         
+
          if (xaConn != null)
          {
             xaConn.close();
          }
+
+        /* if (suspended != null)
+         {
+            tm.resume(suspended);
+         }*/
       }
    }
-   
+
    //http://jira.jboss.com/jira/browse/JBMESSAGING-721
+   // Note: The behavior of this test was changed after http://jira.jboss.com/jira/browse/JBMESSAGING-946
+   // When you have a XASession without a transaction enlisted we will behave the same way as non transactedSession, AutoAck
    public void testTransactionIdSetAfterCommit() throws Exception
    {
       Connection conn = null;
-      
+
       XAConnection xaConn = null;
-      
+
       try
       {
-      
+
          //First send some messages to a queue
-         
+
          conn = cf.createConnection();
-         
+
          Session sessSend = conn.createSession(false, Session.AUTO_ACKNOWLEDGE);
-         
+
          MessageProducer prod = sessSend.createProducer(queue);
-         
+
          TextMessage tm1 = sessSend.createTextMessage("message1");
-         
+
          TextMessage tm2 = sessSend.createTextMessage("message2");
-         
+
          prod.send(tm1);
-         
+
          prod.send(tm2);
-         
-         
+
+
          xaConn = cf.createXAConnection();
-         
+
          XASession xaSession = xaConn.createXASession();
-         
+
          xaConn.start();
-         
+
          MessageConsumer cons = xaSession.createConsumer(queue);
-         
+
          //Now we enlist the session in an xa transaction
-         
+
          XAResource res = xaSession.getXAResource();
-         
+
          tm.begin();
-         
+
          Transaction tx = tm.getTransaction();
          tx.enlistResource(res);
-         
+
          tx.delistResource(res, XAResource.TMSUCCESS);
-         
+
          //Then we do a commit
          tm.commit();
-                              
+
+         // I have changed where this begin was originally set
+         // as when you don't have a resource enlisted, XASessions will act as
+         // non transacted + AutoAck
+
+         //And enlist again - this should convert the work done in the local tx
+         //into the global branch
+
+         tx = tm.getTransaction();
+
+         tm.begin();
+
+         tx = tm.getTransaction();
+         tx.enlistResource(res);
+
          //Then we receive the messages outside the tx
-         
+
          TextMessage rm1 = (TextMessage)cons.receive(1000);
-         
+
          assertNotNull(rm1);
-         
+
          assertEquals("message1", rm1.getText());
-         
+
          TextMessage rm2 = (TextMessage)cons.receive(1000);
-         
+
          assertNotNull(rm2);
-         
+
          assertEquals("message2", rm2.getText());
-         
+
          Message rm3 = cons.receive(1000);
-         
+
          assertNull(rm3);
-         
-         //And enlist again - this should convert the work done in the local tx
-         //into the global branch
-         
-         tx = tm.getTransaction();
-         
-         tm.begin();
-         
-         tx = tm.getTransaction();
-         tx.enlistResource(res);
-         
-         tx.delistResource(res, XAResource.TMSUCCESS);         
-               
+
+         tx.delistResource(res, XAResource.TMSUCCESS);
+
          //Now rollback the tx - this should cause redelivery of the two messages
          tx.rollback();
-         
+
          rm1 = (TextMessage)cons.receive(1000);
-         
+
          assertNotNull(rm1);
-         
+
          assertEquals("message1", rm1.getText());
-         
+
          rm2 = (TextMessage)cons.receive(1000);
-         
+
          assertNotNull(rm2);
-         
+
          assertEquals("message2", rm2.getText());
-         
+
          rm3 = cons.receive(1000);
-         
+
          assertNull(rm3);
       }
       finally
-      {         
+      {
          if (conn != null)
          {
             conn.close();
          }
-         
+
          if (xaConn != null)
          {
             xaConn.close();
@@ -784,109 +1109,114 @@
       }
 
    }
-   
+
    //http://jira.jboss.com/jira/browse/JBMESSAGING-721
    public void testTransactionIdSetAfterRollback() throws Exception
    {
       Connection conn = null;
-      
+
       XAConnection xaConn = null;
-      
+
       try
       {
-      
+
          //First send some messages to a queue
-         
+
          conn = cf.createConnection();
-         
+
          Session sessSend = conn.createSession(false, Session.AUTO_ACKNOWLEDGE);
-         
+
          MessageProducer prod = sessSend.createProducer(queue);
-         
+
          TextMessage tm1 = sessSend.createTextMessage("message1");
-         
+
          TextMessage tm2 = sessSend.createTextMessage("message2");
-         
+
          prod.send(tm1);
-         
+
          prod.send(tm2);
-         
-         
+
+
          xaConn = cf.createXAConnection();
-         
+
          XASession xaSession = xaConn.createXASession();
-         
+
          xaConn.start();
-         
+
          MessageConsumer cons = xaSession.createConsumer(queue);
-         
+
          //Now we enlist the session in an xa transaction
-         
+
          XAResource res = xaSession.getXAResource();
-         
+
          tm.begin();
-         
+
          Transaction tx = tm.getTransaction();
          tx.enlistResource(res);
          tx.delistResource(res, XAResource.TMSUCCESS);
-         
+
          //Then we do a rollback
-         tm.rollback();                 
-         
+         tm.rollback();
+
+
+         tm.begin();
+
+         //And enlist again - the work should then be converted into the global tx branch
+
+
+         // I have changed where this begin was originally set
+         // as when you don't have a resource enlisted, XASessions will act as
+         // non transacted + AutoAck
+
+         tx = tm.getTransaction();
+
+         tx.enlistResource(res);
+
          //Then we receive the messages outside the global tx
-         
+
          TextMessage rm1 = (TextMessage)cons.receive(1000);
-         
+
          assertNotNull(rm1);
-         
+
          assertEquals("message1", rm1.getText());
-         
+
          TextMessage rm2 = (TextMessage)cons.receive(1000);
-         
+
          assertNotNull(rm2);
-         
+
          assertEquals("message2", rm2.getText());
-         
+
          Message rm3 = cons.receive(1000);
-         
+
          assertNull(rm3);
-         
-         tm.begin();
-         
-         //And enlist again - the work should then be converted into the global tx branch
-         
-         tx = tm.getTransaction();
-         
-         tx.enlistResource(res);
-         
          tx.delistResource(res, XAResource.TMSUCCESS);
-               
+
          //Now rollback the tx - this should cause redelivery of the two messages
          tx.rollback();
-         
+
          rm1 = (TextMessage)cons.receive(1000);
-         
+
          assertNotNull(rm1);
-         
+
          assertEquals("message1", rm1.getText());
-         
+
          rm2 = (TextMessage)cons.receive(1000);
-         
+
          assertNotNull(rm2);
-         
+
          assertEquals("message2", rm2.getText());
-         
+
          rm3 = cons.receive(1000);
-         
+
          assertNull(rm3);
       }
       finally
-      {         
+      {
          if (conn != null)
          {
             conn.close();
          }
-         
+
          if (xaConn != null)
          {
             xaConn.close();
@@ -894,7 +1224,7 @@
       }
 
    }
-   
+
    // See http://jira.jboss.org/jira/browse/JBMESSAGING-825
    // Need to test that ids with trailing zeros are dealt with properly - sybase has the habit
    // of truncating trailing zeros in varbinary columns
@@ -904,64 +1234,64 @@
       {
          return;
       }
-      
+
       XAConnection conn1 = null;
-      
+
       try
-      {      
+      {
          conn1 = cf.createXAConnection();
-   
+
          XASession sess1 = conn1.createXASession();
-   
+
          XAResource res1 = sess1.getXAResource();
-         
+
          byte[] branchQualifier = new byte[] { 1, 2, 3, 4, 5, 6, 0, 0, 0, 0 };
-         
+
          byte[] globalTxId = new byte[] { 6, 5, 4, 3, 2, 1, 0, 0, 0, 0 };
-                  
+
          Xid trailing = new MessagingXid(branchQualifier, 12435, globalTxId);
-         
+
          res1.start(trailing, XAResource.TMNOFLAGS);
-   
+
          MessageProducer prod1 = sess1.createProducer(queue);
-   
+
          TextMessage tm1 = sess1.createTextMessage("testing1");
-   
+
          prod1.send(tm1);
-   
+
          res1.end(trailing, XAResource.TMSUCCESS);
-   
-   
+
+
          res1.prepare(trailing);
 
-   
+
          //Now "crash" the server
-   
+
          ServerManagement.stopServerPeer();
-   
+
          ServerManagement.startServerPeer();
-   
+
          ServerManagement.deployQueue("Queue");
-   
-   
+
+
          XAResource res = cf.createXAConnection().createXASession().getXAResource();
-   
+
          Xid[] xids = res.recover(XAResource.TMSTARTRSCAN);
          assertEquals(1, xids.length);
-   
+
          Xid[] xids2 = res.recover(XAResource.TMENDRSCAN);
          assertEquals(0, xids2.length);
-         
+
          Xid trailing2 = xids[0];
-         
+
          assertTrue(trailing.getFormatId() == trailing2.getFormatId());
-         
+
          assertEqualByteArrays(trailing.getGlobalTransactionId(), trailing2.getGlobalTransactionId());
-         
+
          assertEqualByteArrays(trailing.getBranchQualifier(), trailing2.getBranchQualifier());
-   
+
          res.commit(trailing, false);
-            
+
          if (checkNoMessageData())
          {
             fail("Data remains in database");
@@ -979,44 +1309,44 @@
             {
                //Ignore
             }
-         }                  
+         }
       }
    }
-   
+
    public void test2PCSendCommit1PCOptimization() throws Exception
    {
       //Since both resources have some RM, TM will probably use 1PC optimization
-      
+
       XAConnection conn = null;
       Connection conn2 = null;
-      
+
       try
-      {      
+      {
          conn = cf.createXAConnection();
-         
+
          tm.begin();
-         
+
          XASession sess = conn.createXASession();
          XAResource res = sess.getXAResource();
-         
+
          XAResource res2 = new DummyXAResource();
-         
+
          Transaction tx = tm.getTransaction();
          tx.enlistResource(res);
          tx.enlistResource(res2);
-         
+
          MessageProducer prod = sess.createProducer(queue);
          prod.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
          Message m = sess.createTextMessage("XATest1");
          prod.send(queue, m);
          m = sess.createTextMessage("XATest2");
          prod.send(queue, m);
-         
+
          tx.delistResource(res, XAResource.TMSUCCESS);
          tx.delistResource(res2, XAResource.TMSUCCESS);
-         
+
          tm.commit();
-         
+
          conn2 = cf.createConnection();
          conn2.start();
          Session sessReceiver = conn2.createSession(false, Session.AUTO_ACKNOWLEDGE);
@@ -1074,7 +1404,7 @@
          prod.send(queue, m);
          m = sess.createTextMessage("XATest2");
          prod.send(queue, m);
-         
+
          tx.delistResource(res, XAResource.TMSUCCESS);
          tx.delistResource(res2, XAResource.TMSUCCESS);
 
@@ -1133,7 +1463,7 @@
          prod.send(queue, m);
          m = sess.createTextMessage("XATest2");
          prod.send(queue, m);
-         
+
          tx.delistResource(res, XAResource.TMSUCCESS);
          tx.delistResource(res2, XAResource.TMSUCCESS);
 
@@ -1198,7 +1528,7 @@
          tx.delistResource(res2, XAResource.TMSUCCESS);
          tx.delistResource(res3, XAResource.TMSUCCESS);
          tx.delistResource(res4, XAResource.TMSUCCESS);
-         
+
          try
          {
             tm.commit();
@@ -1259,7 +1589,7 @@
          prod.send(queue, m);
          m = sess.createTextMessage("XATest2");
          prod.send(queue, m);
-         
+
          tx.delistResource(res, XAResource.TMSUCCESS);
          tx.delistResource(res2, XAResource.TMSUCCESS);
 
@@ -1330,7 +1660,7 @@
 
          assertNotNull(m2);
          assertEquals("XATest2", m2.getText());
-         
+
          tx.delistResource(res, XAResource.TMSUCCESS);
          tx.delistResource(res2, XAResource.TMSUCCESS);
 
@@ -1345,7 +1675,7 @@
          Message m3 = cons.receive(MIN_TIMEOUT);
 
          assertNull(m3);
-         
+
          tx.delistResource(res, XAResource.TMSUCCESS);
          tx.delistResource(res2, XAResource.TMSUCCESS);
 
@@ -1410,7 +1740,7 @@
 
          assertNotNull(m2);
          assertEquals("XATest2", m2.getText());
-         
+
          tx.delistResource(res, XAResource.TMSUCCESS);
          tx.delistResource(res2, XAResource.TMSUCCESS);
 
@@ -1428,7 +1758,7 @@
 
          tx.delistResource(res, XAResource.TMSUCCESS);
          tx.delistResource(res2, XAResource.TMSUCCESS);
-         
+
          tm.commit();
 
 
@@ -1489,7 +1819,7 @@
          m2 = (TextMessage)cons.receive(MAX_TIMEOUT);
          assertNotNull(m2);
          assertEquals("XATest2", m2.getText());
-         
+
          tx.delistResource(res, XAResource.TMSUCCESS);
          tx.delistResource(res2, XAResource.TMSUCCESS);
 
@@ -1511,7 +1841,7 @@
          assertEquals("XATest2", m3.getText());
 
          assertTrue(m3.getJMSRedelivered());
-         
+
          tx.delistResource(res, XAResource.TMSUCCESS);
          tx.delistResource(res2, XAResource.TMSUCCESS);
 
@@ -1573,7 +1903,7 @@
          m2 = (TextMessage)cons.receive(MAX_TIMEOUT);
          assertNotNull(m2);
          assertEquals("XATest2", m2.getText());
-         
+
          tx.delistResource(res, XAResource.TMSUCCESS);
          tx.delistResource(res2, XAResource.TMSUCCESS);
 
@@ -1595,7 +1925,7 @@
          assertEquals("XATest2", m3.getText());
 
          assertTrue(m3.getJMSRedelivered());
-         
+
          tx.delistResource(res, XAResource.TMSUCCESS);
          tx.delistResource(res2, XAResource.TMSUCCESS);
 
@@ -1645,7 +1975,7 @@
          prod.send(queue, m);
 
          tx.delistResource(res, XAResource.TMSUCCESS);
-         
+
          tm.commit();
 
          conn2 = cf.createConnection();
@@ -1698,7 +2028,7 @@
          prod.send(queue, m);
 
          tx.delistResource(res, XAResource.TMSUCCESS);
-         
+
          tm.rollback();
 
          conn2 = cf.createConnection();
@@ -1760,7 +2090,7 @@
 
          assertNotNull(m2);
          assertEquals("XATest2", m2.getText());
-         
+
          tx.delistResource(res, XAResource.TMSUCCESS);
 
          tm.commit();
@@ -1773,7 +2103,7 @@
          Message m3 = cons.receive(MIN_TIMEOUT);
 
          assertNull(m3);
-         
+
          tx.delistResource(res, XAResource.TMSUCCESS);
 
          tm.commit();
@@ -1835,7 +2165,7 @@
          assertEquals("XATest2", m2.getText());
 
          tx.delistResource(res, XAResource.TMSUCCESS);
-         
+
          tm.rollback();
 
          //Message should be redelivered
@@ -1858,7 +2188,7 @@
          assertTrue(m3.getJMSRedelivered());
 
          tx.delistResource(res, XAResource.TMSUCCESS);
-         
+
          tm.commit();
 
       }
@@ -1927,7 +2257,7 @@
 
          tx.delistResource(res1, XAResource.TMSUCCESS);
          tx.delistResource(res2, XAResource.TMSUCCESS);
-         
+
          //commit
          tm.commit();
 
@@ -2004,7 +2334,7 @@
 
          tx.delistResource(res1, XAResource.TMSUCCESS);
          tx.delistResource(res2, XAResource.TMSUCCESS);
-         
+
          //commit
          tm.commit();
 
@@ -2097,27 +2427,27 @@
          cons2.close();
 
          //rollback
-         
+
          tx.delistResource(res1, XAResource.TMSUCCESS);
          tx.delistResource(res2, XAResource.TMSUCCESS);
 
          tm.rollback();
-           
+
          //Rollback causes cancel which is asynch
          Thread.sleep(1000);
-         
+
          //We cannot assume anything about the order in which the transaction manager rollsback
          //the sessions - this is implementation dependent
 
          Session sess = conn2.createSession(false, Session.AUTO_ACKNOWLEDGE);
          MessageConsumer cons = sess.createConsumer(queue);
-         conn2.start();         
-         
+         conn2.start();
+
          TextMessage r = (TextMessage)cons.receive(MAX_TIMEOUT);
          assertNotNull(r);
-         
+
          boolean session1First = false;
-         
+
          if (r.getText().equals("jellyfish1"))
          {
             session1First = true;
@@ -2130,52 +2460,52 @@
          {
             fail("Unexpected message");
          }
-         
+
          if (session1First)
          {
             r = (TextMessage)cons.receive(MAX_TIMEOUT);
-            
+
             assertNotNull(r);
-            
+
             assertEquals("jellyfish2", r.getText());
-            
+
             r = (TextMessage)cons.receive(MAX_TIMEOUT);
-            
+
             assertNotNull(r);
-            
+
             assertEquals("jellyfish3", r.getText());
-            
+
             r = (TextMessage)cons.receive(MAX_TIMEOUT);
-            
+
             assertNotNull(r);
-            
+
             assertEquals("jellyfish4", r.getText());
-            
-            
+
+
          }
          else
          {
             r = (TextMessage)cons.receive(MAX_TIMEOUT);
-            
+
             assertNotNull(r);
-            
+
             assertEquals("jellyfish4", r.getText());
-            
+
             r = (TextMessage)cons.receive(MAX_TIMEOUT);
-            
+
             assertNotNull(r);
-            
+
             assertEquals("jellyfish1", r.getText());
-            
+
             r = (TextMessage)cons.receive(MAX_TIMEOUT);
-            
+
             assertNotNull(r);
-            
+
             assertEquals("jellyfish2", r.getText());
          }
-         
+
          r = (TextMessage)cons.receive(MIN_TIMEOUT);
-         
+
          assertNull(r);
 
       }
@@ -2244,7 +2574,7 @@
          assertEquals("jellyfish2", r1.getText());
 
          cons1.close();
-         
+
          //Cancel is asynch
          Thread.sleep(500);
 
@@ -2262,15 +2592,15 @@
          //rollback
 
          cons2.close();
-         
+
          tx.delistResource(res1, XAResource.TMSUCCESS);
          tx.delistResource(res2, XAResource.TMSUCCESS);
 
          tm.rollback();
-         
+
          // Rollback causes cancel which is asynch
          Thread.sleep(1000);
-         
+
          //We cannot assume anything about the order in which the transaction manager rollsback
          //the sessions - this is implementation dependent
 
@@ -2281,9 +2611,9 @@
 
          TextMessage r = (TextMessage)cons.receive(MAX_TIMEOUT);
          assertNotNull(r);
-         
+
          boolean session1First = false;
-         
+
          if (r.getText().equals("jellyfish1"))
          {
             session1First = true;
@@ -2296,52 +2626,52 @@
          {
             fail("Unexpected message");
          }
-         
+
          if (session1First)
          {
             r = (TextMessage)cons.receive(MAX_TIMEOUT);
-            
+
             assertNotNull(r);
-            
+
             assertEquals("jellyfish2", r.getText());
-            
+
             r = (TextMessage)cons.receive(MAX_TIMEOUT);
-            
+
             assertNotNull(r);
-            
+
             assertEquals("jellyfish3", r.getText());
-            
+
             r = (TextMessage)cons.receive(MAX_TIMEOUT);
-            
+
             assertNotNull(r);
-            
+
             assertEquals("jellyfish4", r.getText());
-            
-            
+
+
          }
          else
          {
             r = (TextMessage)cons.receive(MAX_TIMEOUT);
-            
+
             assertNotNull(r);
-            
+
             assertEquals("jellyfish4", r.getText());
-            
+
             r = (TextMessage)cons.receive(MAX_TIMEOUT);
-            
+
             assertNotNull(r);
-            
+
             assertEquals("jellyfish1", r.getText());
-            
+
             r = (TextMessage)cons.receive(MAX_TIMEOUT);
-            
+
             assertNotNull(r);
-            
+
             assertEquals("jellyfish2", r.getText());
          }
-         
+
          r = (TextMessage)cons.receive(MIN_TIMEOUT);
-         
+
          assertNull(r);
 
 
@@ -2371,7 +2701,7 @@
          conn2 = cf.createConnection();
          Session sessProducer = conn2.createSession(false, Session.AUTO_ACKNOWLEDGE);
          MessageProducer prod  = sessProducer.createProducer(queue);
-         
+
          Message m = sessProducer.createTextMessage("jellyfish1");
          prod.send(m);
          m = sessProducer.createTextMessage("jellyfish2");
@@ -2383,7 +2713,7 @@
 
          conn = cf.createXAConnection();
          conn.start();
-         
+
          tm.begin();
 
          XASession sess1 = conn.createXASession();
@@ -2415,9 +2745,9 @@
 
          assertNotNull(r1);
          assertEquals("jellyfish4", r1.getText());
-         
+
          r1 = (TextMessage)cons1.receive(1000);
-         
+
          assertNull(r1);
 
          cons1.close();
@@ -2425,10 +2755,10 @@
 
          //try and commit - and we're going to make the dummyxaresource throw an exception on commit,
          //which should cause rollback to be called on the other resource
-         
+
          tx.delistResource(res1, XAResource.TMSUCCESS);
          tx.delistResource(res2, XAResource.TMSUCCESS);
-         
+
          //rollback will cause an attemp to deliver messages locally to the original consumers.
          //the original consumer has closed, so it will cancelled to the server
          //the server cancel is asynch, so we need to sleep for a bit to make sure it completes
@@ -2442,40 +2772,40 @@
          {
             //We should expect this
          }
-         
+
          Thread.sleep(1000);
-         
-         
+
+
          Session sess = conn2.createSession(false, Session.AUTO_ACKNOWLEDGE);
          MessageConsumer cons = sess.createConsumer(queue);
          conn2.start();
 
          TextMessage r = (TextMessage)cons.receive(MAX_TIMEOUT);
-         
+
          assertNotNull(r);
-         
+
          assertEquals("jellyfish1", r.getText());
-         
+
          r = (TextMessage)cons.receive(MAX_TIMEOUT);
-         
+
          assertNotNull(r);
-         
+
          assertEquals("jellyfish2", r.getText());
-         
+
          r = (TextMessage)cons.receive(MAX_TIMEOUT);
-         
+
          assertNotNull(r);
-         
+
          assertEquals("jellyfish3", r.getText());
-         
+
          r = (TextMessage)cons.receive(MAX_TIMEOUT);
-         
+
          assertNotNull(r);
-         
-         assertEquals("jellyfish4", r.getText());         
-         
+
+         assertEquals("jellyfish4", r.getText());
+
          r = (TextMessage)cons.receive(MIN_TIMEOUT);
-         
+
          assertNull(r);
       }
       finally
@@ -2527,7 +2857,7 @@
 
          tx.delistResource(res1, XAResource.TMSUCCESS);
          tx.delistResource(res2, XAResource.TMSUCCESS);
-         
+
          //commit
          tm.commit();
 
@@ -2600,7 +2930,7 @@
 
          tx.delistResource(res1, XAResource.TMSUCCESS);
          tx.delistResource(res2, XAResource.TMSUCCESS);
-         
+
          //commit
          tm.commit();
 
@@ -2672,7 +3002,7 @@
 
          tx.delistResource(res1, XAResource.TMSUCCESS);
          tx.delistResource(res2, XAResource.TMSUCCESS);
-         
+
          //rollback
          tm.rollback();
 
@@ -2739,7 +3069,7 @@
 
          tx.delistResource(res1, XAResource.TMSUCCESS);
          tx.delistResource(res2, XAResource.TMSUCCESS);
-         
+
          //rollback
          tm.rollback();
 
@@ -2822,7 +3152,7 @@
          assertEquals("jellyfish2", r2.getText());
 
          tx2.delistResource(res1, XAResource.TMSUCCESS);
-         
+
          //commit this transaction
          tm.commit();
 
@@ -2836,9 +3166,9 @@
 
          //now resume the first tx and then commit it
          tm.resume(suspended);
-         
+
          tx1.delistResource(res1, XAResource.TMSUCCESS);
-         
+
          tm.commit();
 
 
@@ -2910,7 +3240,7 @@
          assertEquals("jellyfish2", r2.getText());
 
          cons1.close();
-         
+
          tx1.delistResource(res1, XAResource.TMSUCCESS);
 
          //rollback this transaction
@@ -2996,7 +3326,7 @@
          prod1.send(sess1.createTextMessage("kangaroo2"));
 
          tx2.delistResource(res1, XAResource.TMSUCCESS);
-         
+
          //commit this transaction
          tm.commit();
 
@@ -3013,9 +3343,9 @@
 
          //now resume the first tx and then commit it
          tm.resume(suspended);
-         
+
          tx1.delistResource(res1, XAResource.TMSUCCESS);
-         
+
          tm.commit();
 
          //verify that the first text message is received
@@ -3078,7 +3408,7 @@
          prod1.send(sess1.createTextMessage("kangaroo2"));
 
          tx2.delistResource(res1, XAResource.TMSUCCESS);
-         
+
          //rollback this transaction
          tm.rollback();
 
@@ -3094,9 +3424,9 @@
 
          //now resume the first tx and then commit it
          tm.resume(suspended);
-         
+
          tx1.delistResource(res1, XAResource.TMSUCCESS);
-         
+
          tm.commit();
 
          //verify that the first text message is received
@@ -3121,20 +3451,20 @@
    }
 
    // Package protected ---------------------------------------------
-   
+
    // Protected -----------------------------------------------------
-   
+
    // Private -------------------------------------------------------
-   
+
    private void assertEqualByteArrays(byte[] b1, byte[] b2)
    {
       log.info("b1 length: " + b1.length + " b2 length " + b2.length);
-      
+
       if (b1.length != b2.length)
       {
          fail("Lengths not the same");
       }
-      
+
       for (int i = 0; i < b1.length; i++)
       {
          if (b1[i] != b2[i])
@@ -3143,24 +3473,78 @@
          }
       }
    }
-   
+
    // Inner classes -------------------------------------------------
-   
+
+
+   static class DummyListener implements MessageListener
+   {
+
+      protected Logger log = Logger.getLogger(getClass());
+
+      public ArrayList messages = new ArrayList();
+
+      public void onMessage(Message message)
+      {
+         log.info("Message received on DummyListener " + message);
+         messages.add(message);
+      }
+   }
+
+   static class MockServerSessionPool implements ServerSessionPool
+   {
+      private ServerSession serverSession;
+
+      MockServerSessionPool(Session sess)
+      {
+         serverSession = new MockServerSession(sess);
+      }
+
+      public ServerSession getServerSession() throws JMSException
+      {
+         return serverSession;
+      }
+   }
+
+   static class MockServerSession implements ServerSession
+   {
+      Session session;
+
+      MockServerSession(Session sess)
+      {
+         this.session = sess;
+      }
+
+
+      public Session getSession() throws JMSException
+      {
+         return session;
+      }
+
+      public void start() throws JMSException
+      {
+         session.run();
+      }
+
+   }
+
+
+
    static class DummyXAResource implements XAResource
    {
       boolean failOnPrepare;
-      
+
       DummyXAResource()
-      {         
+      {
       }
-      
+
       DummyXAResource(boolean failOnPrepare)
       {
          this.failOnPrepare = failOnPrepare;
       }
 
       public void commit(Xid arg0, boolean arg1) throws XAException
-      {         
+      {
       }
 
       public void end(Xid arg0, int arg1) throws XAException
@@ -3208,7 +3592,7 @@
       {
 
       }
-      
+
    }
 
 }




More information about the jboss-cvs-commits mailing list