diff options
| author | Robert Godfrey <rgodfrey@apache.org> | 2012-10-26 08:44:43 +0000 |
|---|---|---|
| committer | Robert Godfrey <rgodfrey@apache.org> | 2012-10-26 08:44:43 +0000 |
| commit | 0e8d54276e57032e92c24f1da26e0f185e142eb2 (patch) | |
| tree | c57a5746c610dfc7c3b8b299d66f8b0d764b339a /java | |
| parent | 4a5b8e0f01c253500ff31457b0531b789a8eb570 (diff) | |
| download | qpid-python-0e8d54276e57032e92c24f1da26e0f185e142eb2.tar.gz | |
QPID-4395 : Fix client ack in onMessage in JMS AMQP 1.0 client
git-svn-id: https://svn.apache.org/repos/asf/qpid/trunk/qpid@1402430 13f79535-47bb-0310-9956-ffa450edef68
Diffstat (limited to 'java')
2 files changed, 9 insertions, 5 deletions
diff --git a/java/amqp-1-0-client-jms/src/main/java/org/apache/qpid/amqp_1_0/jms/impl/MessageConsumerImpl.java b/java/amqp-1-0-client-jms/src/main/java/org/apache/qpid/amqp_1_0/jms/impl/MessageConsumerImpl.java index e0402cd0a7..eb34cead08 100644 --- a/java/amqp-1-0-client-jms/src/main/java/org/apache/qpid/amqp_1_0/jms/impl/MessageConsumerImpl.java +++ b/java/amqp-1-0-client-jms/src/main/java/org/apache/qpid/amqp_1_0/jms/impl/MessageConsumerImpl.java @@ -316,9 +316,9 @@ public class MessageConsumerImpl implements MessageConsumer, QueueReceiver, Topi _lastUnackedMessage = deliveryTag;
}
- void preReceiveAction(final org.apache.qpid.amqp_1_0.client.Message msg) throws IllegalStateException
+ void preReceiveAction(final org.apache.qpid.amqp_1_0.client.Message msg)
{
- final int acknowledgeMode = _session.getAcknowledgeMode();
+ int acknowledgeMode = _session.getAckModeEnum().ordinal();
if(acknowledgeMode == Session.AUTO_ACKNOWLEDGE
|| acknowledgeMode == Session.DUPS_OK_ACKNOWLEDGE
diff --git a/java/amqp-1-0-client-jms/src/main/java/org/apache/qpid/amqp_1_0/jms/impl/SessionImpl.java b/java/amqp-1-0-client-jms/src/main/java/org/apache/qpid/amqp_1_0/jms/impl/SessionImpl.java index d14aeedf23..cedf9a180a 100644 --- a/java/amqp-1-0-client-jms/src/main/java/org/apache/qpid/amqp_1_0/jms/impl/SessionImpl.java +++ b/java/amqp-1-0-client-jms/src/main/java/org/apache/qpid/amqp_1_0/jms/impl/SessionImpl.java @@ -804,6 +804,10 @@ public class SessionImpl implements Session, QueueSession, TopicSession if(message != null)
{
+ if(_acknowledgeMode == AcknowledgeMode.CLIENT_ACKNOWLEDGE)
+ {
+ consumer.setLastUnackedMessage(msg.getDeliveryTag());
+ }
_currentConsumer = consumer;
_currentMessage = msg;
try
@@ -816,11 +820,11 @@ public class SessionImpl implements Session, QueueSession, TopicSession _currentMessage = null;
}
- if((_recoveredMessage == null) && (_acknowledgeMode == AcknowledgeMode.AUTO_ACKNOWLEDGE
- || _acknowledgeMode == AcknowledgeMode.DUPS_OK_ACKNOWLEDGE))
+ if(_recoveredMessage == null)
{
- consumer.acknowledge(msg);
+ consumer.preReceiveAction(msg);
}
+
}
}
|