summaryrefslogtreecommitdiff
diff options
context:
space:
mode:
authorRafael H. Schloming <rhs@apache.org>2008-08-15 08:21:01 +0000
committerRafael H. Schloming <rhs@apache.org>2008-08-15 08:21:01 +0000
commit41f6548417f398ee9777a8127467bb220a8051eb (patch)
treee05c4a5eaed6722650386ce4a9f32128079456ae
parent21a55f9e4236ccf30d1e5dd3810bc91f4bfc856c (diff)
downloadqpid-python-41f6548417f398ee9777a8127467bb220a8051eb.tar.gz
QPID-1245: use notifyMessage rather than onMessage in setMessageListener so that messages from the synchronous queue actually get acked
git-svn-id: https://svn.apache.org/repos/asf/incubator/qpid/trunk@686172 13f79535-47bb-0310-9956-ffa450edef68
-rw-r--r--qpid/java/client/src/main/java/org/apache/qpid/client/BasicMessageConsumer.java2
-rw-r--r--qpid/java/systests/src/main/java/org/apache/qpid/client/MessageListenerTest.java10
2 files changed, 11 insertions, 1 deletions
diff --git a/qpid/java/client/src/main/java/org/apache/qpid/client/BasicMessageConsumer.java b/qpid/java/client/src/main/java/org/apache/qpid/client/BasicMessageConsumer.java
index 01bb68c23e..dfd228370c 100644
--- a/qpid/java/client/src/main/java/org/apache/qpid/client/BasicMessageConsumer.java
+++ b/qpid/java/client/src/main/java/org/apache/qpid/client/BasicMessageConsumer.java
@@ -280,7 +280,7 @@ public abstract class BasicMessageConsumer<U> extends Closeable implements Messa
Object o = _synchronousQueue.poll();
while (o != null)
{
- messageListener.onMessage((Message) o);
+ notifyMessage((AbstractJMSMessage) o);
o = _synchronousQueue.poll();
}
}
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 e1c0f0ccef..4c1d5ee9c1 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
@@ -152,6 +152,16 @@ public class MessageListenerTest extends QpidTestCase implements MessageListener
// Should have recieved all async messages
assertEquals(MSG_COUNT, receivedCount);
+ _clientConnection.close();
+
+ Connection conn = getConnection("guest", "guest");
+ Session clientSession = conn.createSession(false, Session.AUTO_ACKNOWLEDGE);
+ Queue queue = clientSession.createQueue("message-listener-test-queue");
+ MessageConsumer cons = clientSession.createConsumer(queue);
+ conn.start();
+
+ // check that the messages were actually dequeued
+ assertTrue(cons.receive(2000) == null);
}
public void onMessage(Message message)