From 921a1c732a4a73a4f10064f0293e342f2bc93973 Mon Sep 17 00:00:00 2001
From: Rajith Muditha Attapattu <rajith@apache.org>
Date: Wed, 11 Jan 2012 02:12:38 +0000
Subject: QPID-3604 The code now drains individual consumer queues as well as
 the dispatch queue (via syncDipatchQueue method) and releases both unacked
 and prefetched messages, while only the former being marked redelivered. Also
 all of these transfers are being marked as completed to ensure credits don't
 dry up.

git-svn-id: https://svn.apache.org/repos/asf/qpid/trunk/qpid@1229857 13f79535-47bb-0310-9956-ffa450edef68
---
 .../java/org/apache/qpid/client/AMQSession.java    |  2 +-
 .../org/apache/qpid/client/AMQSession_0_10.java    | 42 +++++++++++++++++++---
 2 files changed, 38 insertions(+), 6 deletions(-)

(limited to 'java/client/src')

diff --git a/java/client/src/main/java/org/apache/qpid/client/AMQSession.java b/java/client/src/main/java/org/apache/qpid/client/AMQSession.java
index 2ae7a17af2..784b75af10 100644
--- a/java/client/src/main/java/org/apache/qpid/client/AMQSession.java
+++ b/java/client/src/main/java/org/apache/qpid/client/AMQSession.java
@@ -371,7 +371,7 @@ public abstract class AMQSession<C extends BasicMessageConsumer, P extends Basic
      * Set when the dispatcher should direct incoming messages straight into the UnackedMessage list instead of
      * to the syncRecieveQueue or MessageListener. Used during cleanup, e.g. in Session.recover().
      */
-    private volatile boolean _usingDispatcherForCleanup;
+    protected volatile boolean _usingDispatcherForCleanup;
 
     /** Used to indicates that the connection to which this session belongs, has been stopped. */
     private boolean _connectionStopped;
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 8dc36673dc..49b77dcc7b 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
@@ -795,11 +795,43 @@ public class AMQSession_0_10 extends AMQSession<BasicMessageConsumer_0_10, Basic
     {
         if (suspend)
         {
-            for (BasicMessageConsumer consumer : _consumers.values())
-            {
-                getQpidSession().messageStop(String.valueOf(consumer.getConsumerTag()),
-                                             Option.UNRELIABLE);
-            }
+                synchronized (getMessageDeliveryLock())
+                {
+                    for (BasicMessageConsumer consumer : _consumers.values())
+	            {
+	                getQpidSession().messageStop(String.valueOf(consumer.getConsumerTag()),
+	                                             Option.UNRELIABLE);
+	                sync();
+	                List<Long> tags = consumer.drainReceiverQueueAndRetrieveDeliveryTags();
+	                _prefetchedMessageTags.addAll(tags);
+	            }
+                }
+
+                _usingDispatcherForCleanup = true;
+                syncDispatchQueue();
+                _usingDispatcherForCleanup = false;
+
+                RangeSet delivered = gatherRangeSet(_unacknowledgedMessageTags);
+		RangeSet prefetched = gatherRangeSet(_prefetchedMessageTags);
+		RangeSet all = RangeSetFactory.createRangeSet(delivered.size()
+					+ prefetched.size());
+
+		for (Iterator<Range> deliveredIter = delivered.iterator(); deliveredIter.hasNext();)
+		{
+			Range range = deliveredIter.next();
+			all.add(range);
+		}
+
+		for (Iterator<Range> prefetchedIter = prefetched.iterator(); prefetchedIter.hasNext();)
+		{
+			Range range = prefetchedIter.next();
+			all.add(range);
+		}
+
+		flushProcessed(all, false);
+		getQpidSession().messageRelease(delivered,Option.SET_REDELIVERED);
+		getQpidSession().messageRelease(prefetched);
+		sync();
         }
         else
         {
-- 
cgit v1.2.1