diff options
Diffstat (limited to 'java')
| -rw-r--r-- | java/client/src/main/java/org/apache/qpid/client/AMQSession_0_10.java | 16 |
1 files changed, 13 insertions, 3 deletions
diff --git a/java/client/src/main/java/org/apache/qpid/client/AMQSession_0_10.java b/java/client/src/main/java/org/apache/qpid/client/AMQSession_0_10.java index dc79555171..34457d745f 100644 --- a/java/client/src/main/java/org/apache/qpid/client/AMQSession_0_10.java +++ b/java/client/src/main/java/org/apache/qpid/client/AMQSession_0_10.java @@ -135,7 +135,7 @@ public class AMQSession_0_10 extends AMQSession<BasicMessageConsumer_0_10, Basic { try { - flushAcknowledgments(); + flushAcknowledgments(true); } catch (Throwable t) { @@ -236,12 +236,17 @@ public class AMQSession_0_10 extends AMQSession<BasicMessageConsumer_0_10, Basic void flushAcknowledgments() { + flushAcknowledgments(false); + } + + void flushAcknowledgments(boolean setSyncBit) + { synchronized (unacked) { if (unackedCount > 0) { messageAcknowledge - (unacked, _acknowledgeMode != org.apache.qpid.jms.Session.NO_ACKNOWLEDGE); + (unacked, _acknowledgeMode != org.apache.qpid.jms.Session.NO_ACKNOWLEDGE,setSyncBit); clearUnacked(); } } @@ -249,6 +254,11 @@ public class AMQSession_0_10 extends AMQSession<BasicMessageConsumer_0_10, Basic void messageAcknowledge(RangeSet ranges, boolean accept) { + messageAcknowledge(ranges,accept,false); + } + + void messageAcknowledge(RangeSet ranges, boolean accept,boolean setSyncBit) + { Session ssn = getQpidSession(); for (Range range : ranges) { @@ -257,7 +267,7 @@ public class AMQSession_0_10 extends AMQSession<BasicMessageConsumer_0_10, Basic ssn.flushProcessed(accept ? BATCH : NONE); if (accept) { - ssn.messageAccept(ranges, UNRELIABLE); + ssn.messageAccept(ranges, UNRELIABLE,setSyncBit? SYNC : NONE); } } |