diff options
Diffstat (limited to 'java/client/src')
| -rw-r--r-- | java/client/src/main/java/org/apache/qpid/client/AMQSession_0_10.java | 7 | ||||
| -rw-r--r-- | java/client/src/main/java/org/apache/qpid/client/BasicMessageConsumer_0_10.java | 27 |
2 files changed, 30 insertions, 4 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 34457d745f..1515a07507 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 @@ -396,10 +396,9 @@ public class AMQSession_0_10 extends AMQSession<BasicMessageConsumer_0_10, Basic public void releaseForRollback() { - if (_dispatcher != null) - { - _dispatcher.rollback(); - } + startDispatcherIfNecessary(); + syncDispatchQueue(); + _dispatcher.rollback(); getQpidSession().messageRelease(_txRangeSet, Option.SET_REDELIVERED); _txRangeSet.clear(); _txSize = 0; 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 8b17dcf91f..1b2d6876bd 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 @@ -430,4 +430,31 @@ public class BasicMessageConsumer_0_10 extends BasicMessageConsumer<UnprocessedM return receiveNoWait(); } + @Override public void rollbackPendingMessages() + { + if (_synchronousQueue.size() > 0) + { + RangeSet ranges = new RangeSet(); + Iterator iterator = _synchronousQueue.iterator(); + while (iterator.hasNext()) + { + + Object o = iterator.next(); + if (o instanceof AbstractJMSMessage) + { + ranges.add((int) ((AbstractJMSMessage) o).getDeliveryTag()); + iterator.remove(); + } + else + { + _logger.error("Queue contained a :" + o.getClass() + + " unable to reject as it is not an AbstractJMSMessage. Will be cleared"); + iterator.remove(); + } + } + + _0_10session.getQpidSession().messageRelease(ranges, Option.SET_REDELIVERED); + clearReceiveQueue(); + } + } } |