You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@qpid.apache.org by kp...@apache.org on 2007/01/30 22:35:15 UTC
svn commit: r501577 - in /incubator/qpid/branches/qpid.0-9/java/client/src:
main/java/org/apache/qpid/client/AMQSession.java
test/java/org/apache/qpid/test/unit/transacted/TransactedTest.java
Author: kpvdr
Date: Tue Jan 30 13:35:14 2007
New Revision: 501577
URL: http://svn.apache.org/viewvc?view=rev&rev=501577
Log:
Fixed TransactedTest and a bug in the rollback handling in the client
Modified:
incubator/qpid/branches/qpid.0-9/java/client/src/main/java/org/apache/qpid/client/AMQSession.java
incubator/qpid/branches/qpid.0-9/java/client/src/test/java/org/apache/qpid/test/unit/transacted/TransactedTest.java
Modified: incubator/qpid/branches/qpid.0-9/java/client/src/main/java/org/apache/qpid/client/AMQSession.java
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/qpid.0-9/java/client/src/main/java/org/apache/qpid/client/AMQSession.java?view=diff&rev=501577&r1=501576&r2=501577
==============================================================================
--- incubator/qpid/branches/qpid.0-9/java/client/src/main/java/org/apache/qpid/client/AMQSession.java (original)
+++ incubator/qpid/branches/qpid.0-9/java/client/src/main/java/org/apache/qpid/client/AMQSession.java Tue Jan 30 13:35:14 2007
@@ -550,6 +550,7 @@
checkTransacted();
try
{
+ _unacknowledged.clear();
// AMQP version change: Hardwire the version to 0-9 (major=0, minor=9)
// TODO: Connect this to the session version obtained from ProtocolInitiation for this session.
// Be aware of possible changes to parameter order as versions change.
@@ -1599,8 +1600,8 @@
_logger.debug("Message received in session with channel id " + _channelId);
}
- _queue.add(message);
_unacknowledged.offer(message.deliveryTag);
+ _queue.add(message);
}
/**
Modified: incubator/qpid/branches/qpid.0-9/java/client/src/test/java/org/apache/qpid/test/unit/transacted/TransactedTest.java
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/qpid.0-9/java/client/src/test/java/org/apache/qpid/test/unit/transacted/TransactedTest.java?view=diff&rev=501577&r1=501576&r2=501577
==============================================================================
--- incubator/qpid/branches/qpid.0-9/java/client/src/test/java/org/apache/qpid/test/unit/transacted/TransactedTest.java (original)
+++ incubator/qpid/branches/qpid.0-9/java/client/src/test/java/org/apache/qpid/test/unit/transacted/TransactedTest.java Tue Jan 30 13:35:14 2007
@@ -24,6 +24,8 @@
import org.apache.qpid.client.AMQQueue;
import org.apache.qpid.client.AMQSession;
import org.apache.qpid.testutil.VMBrokerSetup;
+import org.apache.qpid.exchange.ExchangeDefaults;
+import org.apache.log4j.Logger;
import javax.jms.*;
@@ -47,11 +49,12 @@
private Session testSession;
private MessageConsumer testConsumer1;
private MessageConsumer testConsumer2;
+ private static final Logger _logger = Logger.getLogger(TransactedTest.class);
protected void setUp() throws Exception
{
super.setUp();
- queue1 = new AMQQueue("Q1", false);
+ queue1 = new AMQQueue("Q1", "Q1", false, true);
queue2 = new AMQQueue("Q2", false);
con = new AMQConnection("vm://:1", "guest", "guest", "TransactedTest", "/test");
@@ -67,29 +70,22 @@
prepSession = prepCon.createSession(false, AMQSession.NO_ACKNOWLEDGE);
prepProducer1 = prepSession.createProducer(queue1);
prepCon.start();
-
-
- //add some messages
- prepProducer1.send(prepSession.createTextMessage("A"));
- prepProducer1.send(prepSession.createTextMessage("B"));
- prepProducer1.send(prepSession.createTextMessage("C"));
-
- testCon = new AMQConnection("vm://:1", "guest", "guest", "TestConnection", "/test");
- testSession = testCon.createSession(false, AMQSession.NO_ACKNOWLEDGE);
- testConsumer2 = testSession.createConsumer(queue2);
- testCon.start();
}
protected void tearDown() throws Exception
{
con.close();
- testCon.close();
prepCon.close();
super.tearDown();
}
public void testCommit() throws Exception
{
+ //add some messages
+ prepProducer1.send(prepSession.createTextMessage("A"));
+ prepProducer1.send(prepSession.createTextMessage("B"));
+ prepProducer1.send(prepSession.createTextMessage("C"));
+
//send and receive some messages
producer2.send(session.createTextMessage("X"));
producer2.send(session.createTextMessage("Y"));
@@ -102,17 +98,35 @@
session.commit();
//ensure sent messages can be received and received messages are gone
+ testCon = new AMQConnection("vm://:1", "guest", "guest", "TestConnection", "/test");
+ testSession = testCon.createSession(false, AMQSession.NO_ACKNOWLEDGE);
+ testConsumer1 = testSession.createConsumer(queue1);
+ testConsumer2 = testSession.createConsumer(queue2);
+ testCon.start();
+
expect("X", testConsumer2.receive(1000));
expect("Y", testConsumer2.receive(1000));
expect("Z", testConsumer2.receive(1000));
- testConsumer1 = testSession.createConsumer(queue1);
assertTrue(null == testConsumer1.receive(1000));
assertTrue(null == testConsumer2.receive(1000));
+ testCon.close();
+ }
+
+ // This checks that queue Q1 is in fact empty and does not have any stray
+ // messages left over from the last test (which can affect later tests)...
+ public void testEmpty1() throws Exception
+ {
+ assertTrue(null == consumer1.receive(1000));
}
public void testRollback() throws Exception
{
+ //add some messages
+ prepProducer1.send(prepSession.createTextMessage("A"));
+ prepProducer1.send(prepSession.createTextMessage("B"));
+ prepProducer1.send(prepSession.createTextMessage("C"));
+
producer2.send(session.createTextMessage("X"));
producer2.send(session.createTextMessage("Y"));
producer2.send(session.createTextMessage("Z"));
@@ -128,9 +142,105 @@
expect("B", consumer1.receive(1000));
expect("C", consumer1.receive(1000));
+ //commit
+ session.commit();
+
+ testCon = new AMQConnection("vm://:1", "guest", "guest", "TestConnection", "/test");
+ testSession = testCon.createSession(false, AMQSession.NO_ACKNOWLEDGE);
testConsumer1 = testSession.createConsumer(queue1);
+ testConsumer2 = testSession.createConsumer(queue2);
+ testCon.start();
assertTrue(null == testConsumer1.receive(1000));
assertTrue(null == testConsumer2.receive(1000));
+ testCon.close();
+ }
+
+ // This checks that queue Q1 is in fact empty and does not have any stray
+ // messages left over from the last test (which can affect later tests)...
+ public void testEmpty2() throws Exception
+ {
+ assertTrue(null == consumer1.receive(1000));
+ }
+
+ public void testResendsMsgsAfterSessionClose() throws Exception
+ {
+ Connection con = new AMQConnection("vm://:1", "guest", "guest", "consumer1", "/test");
+
+ Session consumerSession = con.createSession(true, Session.CLIENT_ACKNOWLEDGE);
+ AMQQueue queue3 = new AMQQueue("Q3", false);
+ MessageConsumer consumer = consumerSession.createConsumer(queue3);
+ //force synch to ensure the consumer has resulted in a bound queue
+ ((AMQSession) consumerSession).declareExchangeSynch(ExchangeDefaults.DIRECT_EXCHANGE_NAME, ExchangeDefaults.DIRECT_EXCHANGE_CLASS);
+
+ Connection con2 = new AMQConnection("vm://:1", "guest", "guest", "producer1", "/test");
+ Session producerSession = con2.createSession(true, Session.CLIENT_ACKNOWLEDGE);
+ MessageProducer producer = producerSession.createProducer(queue3);
+
+ _logger.info("Sending four messages");
+ producer.send(producerSession.createTextMessage("msg1"));
+ producer.send(producerSession.createTextMessage("msg2"));
+ producer.send(producerSession.createTextMessage("msg3"));
+ producer.send(producerSession.createTextMessage("msg4"));
+
+ producerSession.commit();
+
+
+ _logger.info("Starting connection");
+ con.start();
+ TextMessage tm = (TextMessage) consumer.receive();
+
+ tm.acknowledge();
+ consumerSession.commit();
+
+ _logger.info("Received and acknowledged first message");
+ tm = (TextMessage) consumer.receive(1000);
+ assertNotNull(tm);
+ tm = (TextMessage) consumer.receive(1000);
+ assertNotNull(tm);
+ tm = (TextMessage) consumer.receive(1000);
+ assertNotNull(tm);
+ _logger.info("Received all four messages. Closing connection with three outstanding messages");
+
+ consumerSession.close();
+
+ consumerSession = con.createSession(true, Session.CLIENT_ACKNOWLEDGE);
+
+ consumer = consumerSession.createConsumer(queue3);
+
+ // no ack for last three messages so when I call recover I expect to get three messages back
+
+ tm = (TextMessage) consumer.receive(3000);
+ assertNotNull(tm);
+ assertEquals("msg2", tm.getText());
+
+ tm = (TextMessage) consumer.receive(3000);
+ assertNotNull(tm);
+ assertEquals("msg3", tm.getText());
+
+ tm = (TextMessage) consumer.receive(3000);
+ assertNotNull(tm);
+ assertEquals("msg4", tm.getText());
+
+ _logger.info("Received redelivery of three messages. Acknowledging last message");
+ tm.acknowledge();
+ consumerSession.commit();
+ _logger.info("Calling acknowledge with no outstanding messages");
+ // all acked so no messages to be delivered
+
+
+ tm = (TextMessage) consumer.receiveNoWait();
+ assertNull(tm);
+ _logger.info("No messages redelivered as is expected");
+
+ con.close();
+ con2.close();
+ }
+
+ // This checks that queue Q1 is in fact empty and does not have any stray
+ // messages left over from the last test (which can affect later tests)...
+ public void testEmpty3() throws Exception
+ {
+ assertTrue(null == consumer1.receive(1000));
}
private void expect(String text, Message msg) throws JMSException