From 385956302c8f65a0b1908efdbb8bc5df9dbffcc3 Mon Sep 17 00:00:00 2001
From: Rajith Muditha Attapattu <rajith@apache.org>
Date: Tue, 24 Jan 2012 23:26:46 +0000
Subject: QPID-3604 Once message stop is issued for each subscriber, the client
 now drains the internal queues of each subscriber. It also drains the
 dispatch queue. These messages are then released without marking them as
 redelivered. Messages that were given to the application but were not acked
 are also released, but are marked as redelivered. All messages received upto
 that point are marked as completed.

git-svn-id: https://svn.apache.org/repos/asf/qpid/trunk/qpid@1235550 13f79535-47bb-0310-9956-ffa450edef68
---
 .../java/org/apache/qpid/client/AMQSession.java    | 54 +++++++++++++++++++++-
 .../org/apache/qpid/client/AMQSession_0_10.java    | 40 ++++++++++++++++
 2 files changed, 93 insertions(+), 1 deletion(-)

(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 48c4e3e3e6..90e3100690 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;
@@ -2247,6 +2247,58 @@ public abstract class AMQSession<C extends BasicMessageConsumer, P extends Basic
         }
     }
 
+    void drainDispatchQueue()
+    {
+        if (Thread.currentThread() == _dispatcherThread)
+        {
+            while (!_closed.get() && !_queue.isEmpty())
+            {
+                Dispatchable disp;
+                try
+                {
+                    disp = (Dispatchable) _queue.take();
+                }
+                catch (InterruptedException e)
+                {
+                    throw new RuntimeException(e);
+                }
+
+                // Check just in case _queue becomes empty, it shouldn't but
+                // better than an NPE.
+                if (disp == null)
+                {
+                    _logger.debug("_queue became empty during sync.");
+                    break;
+                }
+
+                disp.dispatch(AMQSession.this);
+            }
+        }
+        else
+        {
+            startDispatcherIfNecessary(false);
+
+            final CountDownLatch signal = new CountDownLatch(1);
+
+            _queue.add(new Dispatchable()
+            {
+                public void dispatch(AMQSession ssn)
+                {
+                    signal.countDown();
+                }
+            });
+
+            try
+            {
+                signal.await();
+            }
+            catch (InterruptedException e)
+            {
+                throw new RuntimeException(e);
+            }
+        }
+    }
+
     /**
      * Resubscribes all producers and consumers. This is called when performing failover.
      *
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 8395c8f4b7..dcabfecac3 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
@@ -1354,5 +1354,45 @@ public class AMQSession_0_10 extends AMQSession<BasicMessageConsumer_0_10, Basic
         super.resubscribe();
         getQpidSession().sync();
     }
+
+    @Override
+    void stop() throws AMQException
+    {
+        super.stop();
+        synchronized (getMessageDeliveryLock())
+        {
+	        for (BasicMessageConsumer consumer : _consumers.values())
+	        {
+	            List<Long> tags = consumer.drainReceiverQueueAndRetrieveDeliveryTags();
+	            _prefetchedMessageTags.addAll(tags);
+	        }
+        }
+        _usingDispatcherForCleanup = true;
+        drainDispatchQueue();
+        _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();
+    }
+
 }
 
-- 
cgit v1.2.1