diff options
| -rw-r--r-- | java/client/src/main/java/org/apache/qpid/client/BasicMessageConsumer.java | 25 | ||||
| -rw-r--r-- | java/client/src/main/java/org/apache/qpid/client/BasicMessageConsumer_0_10.java | 4 |
2 files changed, 15 insertions, 14 deletions
diff --git a/java/client/src/main/java/org/apache/qpid/client/BasicMessageConsumer.java b/java/client/src/main/java/org/apache/qpid/client/BasicMessageConsumer.java index c4245d4fc8..8288581538 100644 --- a/java/client/src/main/java/org/apache/qpid/client/BasicMessageConsumer.java +++ b/java/client/src/main/java/org/apache/qpid/client/BasicMessageConsumer.java @@ -290,6 +290,7 @@ public abstract class BasicMessageConsumer<H, B> extends Closeable implements Me } _session.setInRecovery(false); + preDeliver(jmsMsg); } /** @@ -642,7 +643,7 @@ public abstract class BasicMessageConsumer<H, B> extends Closeable implements Me // if (!_closed.get()) { - preDeliver(jmsMessage); + //preDeliver(jmsMessage); notifyMessage(jmsMessage); } @@ -724,6 +725,17 @@ public abstract class BasicMessageConsumer<H, B> extends Closeable implements Me // to send out the appropriate frame msg.setAMQSession(_session); break; + case Session.SESSION_TRANSACTED: + if (isNoConsume()) + { + _session.acknowledgeMessage(msg.getDeliveryTag(), false); + } + else + { + _session.addDeliveredMessage(msg.getDeliveryTag()); + } + + break; } } @@ -750,17 +762,6 @@ public abstract class BasicMessageConsumer<H, B> extends Closeable implements Me } break; - case Session.SESSION_TRANSACTED: - if (isNoConsume()) - { - _session.acknowledgeMessage(msg.getDeliveryTag(), false); - } - else - { - _session.addDeliveredMessage(msg.getDeliveryTag()); - } - - break; } } diff --git a/java/client/src/main/java/org/apache/qpid/client/BasicMessageConsumer_0_10.java b/java/client/src/main/java/org/apache/qpid/client/BasicMessageConsumer_0_10.java index 6103d92fd8..9230225bd5 100644 --- a/java/client/src/main/java/org/apache/qpid/client/BasicMessageConsumer_0_10.java +++ b/java/client/src/main/java/org/apache/qpid/client/BasicMessageConsumer_0_10.java @@ -261,11 +261,11 @@ public class BasicMessageConsumer_0_10 extends BasicMessageConsumer<Struct[], By @Override protected void preApplicationProcessing(AbstractJMSMessage jmsMsg) throws JMSException { - if (!_session.getTransacted()) + super.preApplicationProcessing(jmsMsg); + if (!_session.getTransacted() && _session.getAcknowledgeMode() != org.apache.qpid.jms.Session.CLIENT_ACKNOWLEDGE) { _session.addUnacknowledgedMessage(jmsMsg.getDeliveryTag()); } - _session.setInRecovery(false); } @Override public AbstractJMSMessage createJMSMessageFromUnprocessedMessage( |