From a7484073368b0334cd174074bc4576f031a5ebe1 Mon Sep 17 00:00:00 2001 From: "Rafael H. Schloming" Date: Wed, 25 Feb 2009 23:21:13 +0000 Subject: Made the various receive variants check that the server queue is empty before returning null. Also modified AMQQueueBrowser to use receiveNoWait() when browsing queues using 0-10. These changes uncovered numerous second order bugs, mostly in failover. These are also fixed. This fixes QPID-1642 and QPID-1643. git-svn-id: https://svn.apache.org/repos/asf/qpid/trunk@747963 13f79535-47bb-0310-9956-ffa450edef68 --- qpid/java/08ExcludeList | 1 + qpid/java/08ExcludeList-nonvm | 1 + .../qpid/example/amqpexample/direct/Listener.java | 2 + .../qpid/example/amqpexample/fanout/Listener.java | 2 + .../qpid/example/amqpexample/headers/Listener.java | 2 + .../qpid/example/amqpexample/lvq/Listener.java | 2 + .../example/amqpexample/pubsub/TopicListener.java | 2 + .../qpid/client/AMQConnectionDelegate_0_10.java | 6 -- .../org/apache/qpid/client/AMQQueueBrowser.java | 4 +- .../java/org/apache/qpid/client/AMQSession.java | 52 ++++++++--- .../org/apache/qpid/client/AMQSession_0_10.java | 101 ++++++++++++--------- .../org/apache/qpid/client/AMQSession_0_8.java | 6 +- .../apache/qpid/client/BasicMessageConsumer.java | 26 ++---- .../qpid/client/BasicMessageConsumer_0_10.java | 64 ++++++++----- .../qpid/client/BasicMessageConsumer_0_8.java | 20 ++-- .../qpid/test/unit/message/TestAMQSession.java | 7 +- .../src/main/java/org/apache/qpid/ToyClient.java | 2 + .../main/java/org/apache/qpid/transport/Echo.java | 2 + .../java/org/apache/qpid/transport/Session.java | 42 +++++++-- .../org/apache/qpid/transport/SessionListener.java | 2 + .../main/java/org/apache/qpid/transport/Sink.java | 2 + .../org/apache/qpid/transport/ConnectionTest.java | 3 + .../management/domain/services/QpidService.java | 2 + .../apache/qpid/client/MessageListenerTest.java | 8 ++ .../qpid/test/client/failover/FailoverTest.java | 6 -- .../main/java/org/apache/qpid/tools/QpidBench.java | 2 + 26 files changed, 232 insertions(+), 137 deletions(-) (limited to 'qpid/java') diff --git a/qpid/java/08ExcludeList b/qpid/java/08ExcludeList index 88eb754950..1d8950dddf 100644 --- a/qpid/java/08ExcludeList +++ b/qpid/java/08ExcludeList @@ -6,3 +6,4 @@ org.apache.qpid.test.testcases.FailoverTest#* org.apache.qpid.test.client.failover.FailoverTest#test4MinuteFailover // Those tests are written against the 0.10 path org.apache.qpid.test.unit.message.UTF8Test#* +org.apache.qpid.client.MessageListenerTest#testSynchronousRecieveNoWait diff --git a/qpid/java/08ExcludeList-nonvm b/qpid/java/08ExcludeList-nonvm index eb6c60b225..546dc01f5b 100644 --- a/qpid/java/08ExcludeList-nonvm +++ b/qpid/java/08ExcludeList-nonvm @@ -27,3 +27,4 @@ org.apache.qpid.server.security.acl.SimpleACLTest#* // Those tests are written against the 0.10 path org.apache.qpid.test.unit.message.UTF8Test#* +org.apache.qpid.client.MessageListenerTest#testSynchronousRecieveNoWait diff --git a/qpid/java/client/example/src/main/java/org/apache/qpid/example/amqpexample/direct/Listener.java b/qpid/java/client/example/src/main/java/org/apache/qpid/example/amqpexample/direct/Listener.java index 93bb097268..1ac3e85f7a 100755 --- a/qpid/java/client/example/src/main/java/org/apache/qpid/example/amqpexample/direct/Listener.java +++ b/qpid/java/client/example/src/main/java/org/apache/qpid/example/amqpexample/direct/Listener.java @@ -42,6 +42,8 @@ public class Listener implements SessionListener public void opened(Session ssn) {} + public void resumed(Session ssn) {} + public void message(Session ssn, MessageTransfer xfr) { System.out.println("Message: " + xfr); diff --git a/qpid/java/client/example/src/main/java/org/apache/qpid/example/amqpexample/fanout/Listener.java b/qpid/java/client/example/src/main/java/org/apache/qpid/example/amqpexample/fanout/Listener.java index 4c72ce75a5..21f9c43cd2 100755 --- a/qpid/java/client/example/src/main/java/org/apache/qpid/example/amqpexample/fanout/Listener.java +++ b/qpid/java/client/example/src/main/java/org/apache/qpid/example/amqpexample/fanout/Listener.java @@ -42,6 +42,8 @@ public class Listener implements SessionListener public void opened(Session ssn) {} + public void resumed(Session ssn) {} + public void message(Session ssn, MessageTransfer xfr) { System.out.println("Message: " + xfr); diff --git a/qpid/java/client/example/src/main/java/org/apache/qpid/example/amqpexample/headers/Listener.java b/qpid/java/client/example/src/main/java/org/apache/qpid/example/amqpexample/headers/Listener.java index aa1c9b0a41..dff49228a1 100644 --- a/qpid/java/client/example/src/main/java/org/apache/qpid/example/amqpexample/headers/Listener.java +++ b/qpid/java/client/example/src/main/java/org/apache/qpid/example/amqpexample/headers/Listener.java @@ -32,6 +32,8 @@ public class Listener implements SessionListener public void opened(Session ssn) {} + public void resumed(Session ssn) {} + public void message(Session ssn, MessageTransfer xfr) { String body = xfr.getBodyString(); diff --git a/qpid/java/client/example/src/main/java/org/apache/qpid/example/amqpexample/lvq/Listener.java b/qpid/java/client/example/src/main/java/org/apache/qpid/example/amqpexample/lvq/Listener.java index b930062813..e17d3eef9f 100644 --- a/qpid/java/client/example/src/main/java/org/apache/qpid/example/amqpexample/lvq/Listener.java +++ b/qpid/java/client/example/src/main/java/org/apache/qpid/example/amqpexample/lvq/Listener.java @@ -44,6 +44,8 @@ public class Listener implements SessionListener public void opened(Session ssn) {} + public void resumed(Session ssn) {} + public void message(Session ssn, MessageTransfer xfr) { String body = xfr.getBodyString(); diff --git a/qpid/java/client/example/src/main/java/org/apache/qpid/example/amqpexample/pubsub/TopicListener.java b/qpid/java/client/example/src/main/java/org/apache/qpid/example/amqpexample/pubsub/TopicListener.java index 5e6d3c6f69..dd9307ca84 100755 --- a/qpid/java/client/example/src/main/java/org/apache/qpid/example/amqpexample/pubsub/TopicListener.java +++ b/qpid/java/client/example/src/main/java/org/apache/qpid/example/amqpexample/pubsub/TopicListener.java @@ -40,6 +40,8 @@ public class TopicListener implements SessionListener public void opened(Session ssn) {} + public void resumed(Session ssn) {} + public void message(Session ssn, MessageTransfer xfr) { DeliveryProperties dp = xfr.getHeader().get(DeliveryProperties.class); diff --git a/qpid/java/client/src/main/java/org/apache/qpid/client/AMQConnectionDelegate_0_10.java b/qpid/java/client/src/main/java/org/apache/qpid/client/AMQConnectionDelegate_0_10.java index 29f1aec2f5..c2fb05d94e 100644 --- a/qpid/java/client/src/main/java/org/apache/qpid/client/AMQConnectionDelegate_0_10.java +++ b/qpid/java/client/src/main/java/org/apache/qpid/client/AMQConnectionDelegate_0_10.java @@ -239,12 +239,6 @@ public class AMQConnectionDelegate_0_10 implements AMQConnectionDelegate, Connec { _conn.failoverPrep(); _qpidConnection.resume(); - - if (_conn.firePreResubscribe()) - { - _conn.resubscribeSessions(); - } - _conn.fireFailoverComplete(); return; } diff --git a/qpid/java/client/src/main/java/org/apache/qpid/client/AMQQueueBrowser.java b/qpid/java/client/src/main/java/org/apache/qpid/client/AMQQueueBrowser.java index 08fd49286b..d7a54cad4c 100644 --- a/qpid/java/client/src/main/java/org/apache/qpid/client/AMQQueueBrowser.java +++ b/qpid/java/client/src/main/java/org/apache/qpid/client/AMQQueueBrowser.java @@ -97,7 +97,7 @@ public class AMQQueueBrowser implements QueueBrowser return new Enumeration() { - Message _nextMessage = consumer == null ? null : consumer.receive(1000); + Message _nextMessage = consumer == null ? null : consumer.receiveBrowse(); public boolean hasMoreElements() { @@ -111,7 +111,7 @@ public class AMQQueueBrowser implements QueueBrowser try { _logger.info("QB:nextElement about to receive"); - _nextMessage = consumer.receive(1000); + _nextMessage = consumer.receiveBrowse(); _logger.info("QB:nextElement received:" + _nextMessage); } catch (JMSException e) diff --git a/qpid/java/client/src/main/java/org/apache/qpid/client/AMQSession.java b/qpid/java/client/src/main/java/org/apache/qpid/client/AMQSession.java index 733bee2d81..9012632adf 100644 --- a/qpid/java/client/src/main/java/org/apache/qpid/client/AMQSession.java +++ b/qpid/java/client/src/main/java/org/apache/qpid/client/AMQSession.java @@ -574,13 +574,20 @@ public abstract class AMQSession(new FailoverProtectedOperation()*/ new FailoverNoopSupport(new FailoverProtectedOperation() { public Object execute() throws AMQException, FailoverException { - sendQueueBind(queueName, routingKey, arguments, exchangeName, destination); + sendQueueBind(queueName, routingKey, arguments, exchangeName, destination, nowait); return null; } }, _connection).execute(); @@ -595,7 +602,8 @@ public abstract class AMQSession(*/ @@ -2296,14 +2320,15 @@ public abstract class AMQSession false - // We need to sync so that we get notify of an error. - getQpidSession().sync(); - getCurrentException(); + if (!nowait) + { + // We need to sync so that we get notify of an error. + getQpidSession().sync(); + getCurrentException(); + } return res; } @@ -609,7 +626,8 @@ public class AMQSession_0_10 extends AMQSession(*/ @@ -736,34 +770,11 @@ public class AMQSession_0_10 extends AMQSession extends Closeable implements Messa o = _synchronousQueue.take(); } return o; - } + } + + abstract Message receiveBrowse() throws JMSException; public Message receiveNoWait() throws JMSException { @@ -1037,23 +1039,6 @@ public abstract class BasicMessageConsumer extends Closeable implements Messa _synchronousQueue.clear(); } - public void start() - { - // do nothing as this is a 0_10 feature - } - - - public void stop() - { - // do nothing as this is a 0_10 feature - } - - public boolean isStrated() - { - // do nothing as this is a 0_10 feature - return false; - } - public AMQShortString getQueuename() { return _queuename; @@ -1070,10 +1055,13 @@ public abstract class BasicMessageConsumer extends Closeable implements Messa } /** to be called when a failover has occured */ - public void failedOver() + public void failedOverPre() { clearReceiveQueue(); // TGM FIXME: think this should just be removed // clearUnackedMessages(); } + + public void failedOverPost() {} + } diff --git a/qpid/java/client/src/main/java/org/apache/qpid/client/BasicMessageConsumer_0_10.java b/qpid/java/client/src/main/java/org/apache/qpid/client/BasicMessageConsumer_0_10.java index 7d535643c0..9db2007e1a 100644 --- a/qpid/java/client/src/main/java/org/apache/qpid/client/BasicMessageConsumer_0_10.java +++ b/qpid/java/client/src/main/java/org/apache/qpid/client/BasicMessageConsumer_0_10.java @@ -31,6 +31,7 @@ import org.apache.qpid.filter.JMSSelectorFilter; import javax.jms.InvalidSelectorException; import javax.jms.JMSException; +import javax.jms.Message; import javax.jms.MessageListener; import java.util.Iterator; import java.util.concurrent.atomic.AtomicBoolean; @@ -148,7 +149,8 @@ public class BasicMessageConsumer_0_10 extends BasicMessageConsumer 0) + if (expiry > 0 && !m.isUnreliable()) { commands[mod(next, commands.length)] = m; commandBytes += m.getBodySize(); diff --git a/qpid/java/common/src/main/java/org/apache/qpid/transport/SessionListener.java b/qpid/java/common/src/main/java/org/apache/qpid/transport/SessionListener.java index 63690177f9..eb650eb9ed 100644 --- a/qpid/java/common/src/main/java/org/apache/qpid/transport/SessionListener.java +++ b/qpid/java/common/src/main/java/org/apache/qpid/transport/SessionListener.java @@ -31,6 +31,8 @@ public interface SessionListener void opened(Session session); + void resumed(Session session); + void message(Session ssn, MessageTransfer xfr); void exception(Session session, SessionException exception); diff --git a/qpid/java/common/src/main/java/org/apache/qpid/transport/Sink.java b/qpid/java/common/src/main/java/org/apache/qpid/transport/Sink.java index 622993effb..88870284f6 100644 --- a/qpid/java/common/src/main/java/org/apache/qpid/transport/Sink.java +++ b/qpid/java/common/src/main/java/org/apache/qpid/transport/Sink.java @@ -87,6 +87,8 @@ public class Sink implements SessionListener public void opened(Session ssn) {} + public void resumed(Session ssn) {} + public void message(Session ssn, MessageTransfer xfr) { count++; diff --git a/qpid/java/common/src/test/java/org/apache/qpid/transport/ConnectionTest.java b/qpid/java/common/src/test/java/org/apache/qpid/transport/ConnectionTest.java index dca6264367..3d634bfb70 100644 --- a/qpid/java/common/src/test/java/org/apache/qpid/transport/ConnectionTest.java +++ b/qpid/java/common/src/test/java/org/apache/qpid/transport/ConnectionTest.java @@ -74,6 +74,8 @@ public class ConnectionTest extends TestCase implements SessionListener public void opened(Session ssn) {} + public void resumed(Session ssn) {} + public void message(final Session ssn, MessageTransfer xfr) { if (queue) @@ -277,6 +279,7 @@ public class ConnectionTest extends TestCase implements SessionListener class TestSessionListener implements SessionListener { public void opened(Session s) {} + public void resumed(Session s) {} public void exception(Session s, SessionException e) {} public void message(Session s, MessageTransfer xfr) { diff --git a/qpid/java/management/client/src/main/java/org/apache/qpid/management/domain/services/QpidService.java b/qpid/java/management/client/src/main/java/org/apache/qpid/management/domain/services/QpidService.java index a12993d40e..ee41beaf50 100644 --- a/qpid/java/management/client/src/main/java/org/apache/qpid/management/domain/services/QpidService.java +++ b/qpid/java/management/client/src/main/java/org/apache/qpid/management/domain/services/QpidService.java @@ -86,6 +86,8 @@ public class QpidService implements SessionListener public void opened(Session ssn) {} + public void resumed(Session ssn) {} + public void message(Session ssn, MessageTransfer xfr) { MessagePartListenerAdapter l = _listeners.get(xfr.getDestination()); diff --git a/qpid/java/systests/src/main/java/org/apache/qpid/client/MessageListenerTest.java b/qpid/java/systests/src/main/java/org/apache/qpid/client/MessageListenerTest.java index 4c1d5ee9c1..ffec6c7a29 100644 --- a/qpid/java/systests/src/main/java/org/apache/qpid/client/MessageListenerTest.java +++ b/qpid/java/systests/src/main/java/org/apache/qpid/client/MessageListenerTest.java @@ -106,6 +106,14 @@ public class MessageListenerTest extends QpidTestCase implements MessageListener } } + public void testSynchronousRecieveNoWait() throws Exception + { + for (int msg = 0; msg < MSG_COUNT; msg++) + { + assertTrue(_consumer.receiveNoWait() != null); + } + } + public void testAsynchronousRecieve() throws Exception { _consumer.setMessageListener(this); diff --git a/qpid/java/systests/src/main/java/org/apache/qpid/test/client/failover/FailoverTest.java b/qpid/java/systests/src/main/java/org/apache/qpid/test/client/failover/FailoverTest.java index cd921f0971..4c4ef0320c 100644 --- a/qpid/java/systests/src/main/java/org/apache/qpid/test/client/failover/FailoverTest.java +++ b/qpid/java/systests/src/main/java/org/apache/qpid/test/client/failover/FailoverTest.java @@ -230,12 +230,6 @@ public class FailoverTest extends FailoverBaseCase implements ConnectionListener causeFailure(DEFAULT_FAILOVER_TIME); - if (!CLUSTERED) - { - msg = consumer.receive(500); - assertNull("Should not have received message from new broker!", msg); - } - // Check that you produce and consume the rest of messages. _logger.debug("=================="); _logger.debug("Sending " + (totalMessages-toProduce) + " messages"); diff --git a/qpid/java/tools/src/main/java/org/apache/qpid/tools/QpidBench.java b/qpid/java/tools/src/main/java/org/apache/qpid/tools/QpidBench.java index 4bba7b113d..9770adceb0 100644 --- a/qpid/java/tools/src/main/java/org/apache/qpid/tools/QpidBench.java +++ b/qpid/java/tools/src/main/java/org/apache/qpid/tools/QpidBench.java @@ -696,6 +696,8 @@ public class QpidBench public void opened(org.apache.qpid.transport.Session ssn) {} + public void resumed(org.apache.qpid.transport.Session ssn) {} + public void exception(org.apache.qpid.transport.Session ssn, SessionException exc) { -- cgit v1.2.1