[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