Merge bug22236 ⇨ default

Author: Matthew Sackman

src/com/rabbitmq/client/QueueingConsumer.java

@@ -37,28 +37,37 @@
 import java.util.concurrent.TimeUnit;
 
 import com.rabbitmq.client.AMQP.BasicProperties;
-import com.rabbitmq.utility.ValueOrException;
+import com.rabbitmq.utility.Utility;
 
 /**
  * Convenience class: an implementation of {@link Consumer} with straightforward blocking semantics
  */
 public class QueueingConsumer extends DefaultConsumer {
-    public BlockingQueue<ValueOrException<Delivery, ShutdownSignalException>> _queue;
+    private final BlockingQueue<Delivery> _queue;
+
+    // When this is non-null the queue is in shutdown mode and nextDelivery should
+    // throw a shutdown signal exception.
+    private volatile ShutdownSignalException _shutdown;
+
+    // Marker object used to signal the queue is in shutdown mode. 
+    // It is only there to wake up consumers. The canonical representation
+    // of shutting down is the presence of _shutdown. 
+    // Invariant: This is never on _queue unless _shutdown != null.
+    private static final Delivery POISON = new Delivery(null, null, null);
 
     public QueueingConsumer(Channel ch) {
-        this(ch,
-             new LinkedBlockingQueue<ValueOrException<Delivery, ShutdownSignalException>>());
+        this(ch, new LinkedBlockingQueue<Delivery>());
     }
 
-    public QueueingConsumer(Channel ch,
-                            BlockingQueue<ValueOrException<Delivery, ShutdownSignalException>> q)
+    public QueueingConsumer(Channel ch, BlockingQueue<Delivery> q)
     {
         super(ch);
         this._queue = q;
     }
 
     @Override public void handleShutdownSignal(String consumerTag, ShutdownSignalException sig) {
-        this._queue.add(ValueOrException. <Delivery, ShutdownSignalException> makeException(sig));
+        _shutdown = sig; 
+        _queue.add(POISON);
     }
 
     @Override public void handleDelivery(String consumerTag,
@@ -67,8 +76,8 @@ public QueueingConsumer(Channel ch,
                                byte[] body)
         throws IOException
     {
-        this._queue.add(ValueOrException. <Delivery, ShutdownSignalException> makeValue
-                        (new Delivery(envelope, properties, body)));
+        checkShutdown();
+        this._queue.add(new Delivery(envelope, properties, body));
     }
 
     /**
@@ -110,6 +119,27 @@ public byte[] getBody() {
         }
     }
 
+    /**
+     * Check if we are in shutdown mode and if so throw an exception.
+     */
+    private void checkShutdown(){
+      if(_shutdown != null) throw Utility.fixStackTrace(_shutdown);
+    }
+
+    /**
+     * If this is a non-POISON non-null delivery simply return it.
+     * If this is POISON we are in shutdown mode, throw _shutdown
+     * If this is null, we may be in shutdown mode. Check and see.
+     */
+    private Delivery handle(Delivery delivery)
+    {
+      if(delivery == POISON || (delivery == null && _shutdown != null)){
+        if(delivery == POISON) _queue.add(POISON);
+        throw Utility.fixStackTrace(_shutdown);
+      }
+      return delivery;
+    }
+
     /**
      * Main application-side API: wait for the next message delivery and return it.
      * @return the next message
@@ -119,7 +149,7 @@ public byte[] getBody() {
     public Delivery nextDelivery()
         throws InterruptedException, ShutdownSignalException
     {
-        return _queue.take().getValue();
+        return handle(_queue.take());
     }
 
     /**
@@ -132,16 +162,7 @@ public Delivery nextDelivery()
     public Delivery nextDelivery(long timeout)
         throws InterruptedException, ShutdownSignalException
     {
-        ValueOrException<Delivery, ShutdownSignalException> r =
-            _queue.poll(timeout, TimeUnit.MILLISECONDS);
-        return r == null ? null : r.getValue();
-    }
-
-    /**
-     * Retrieve the underlying blocking queue.
-     * @return the queue where incoming messages are stored
-     */
-    public BlockingQueue<ValueOrException<Delivery, ShutdownSignalException>> getQueue() {
-        return _queue;
+        checkShutdown();
+        return handle(_queue.poll(timeout, TimeUnit.MILLISECONDS));
     }
 }

test/src/com/rabbitmq/client/test/ClientTests.java

@@ -47,6 +47,7 @@ public static TestSuite suite() {
         suite.addTestSuite(Bug20004Test.class);
         suite.addTestSuite(CloseInMainLoop.class);
         suite.addTestSuite(ChannelNumberAllocationTests.class);
+        suite.addTestSuite(QueueingConsumerShutdownTests.class);
         return suite;
     }
 }

test/src/com/rabbitmq/client/test/QueueingConsumerShutdownTests.java

@@ -0,0 +1,76 @@
+//   The contents of this file are subject to the Mozilla Public License
+//   Version 1.1 (the "License"); you may not use this file except in
+//   compliance with the License. You may obtain a copy of the License at
+//   http://www.mozilla.org/MPL/
+//
+//   Software distributed under the License is distributed on an "AS IS"
+//   basis, WITHOUT WARRANTY OF ANY KIND, either express or implied. See the
+//   License for the specific language governing rights and limitations
+//   under the License.
+//
+//   The Original Code is RabbitMQ.
+//
+//   The Initial Developers of the Original Code are LShift Ltd,
+//   Cohesive Financial Technologies LLC, and Rabbit Technologies Ltd.
+//
+//   Portions created before 22-Nov-2008 00:00:00 GMT by LShift Ltd,
+//   Cohesive Financial Technologies LLC, or Rabbit Technologies Ltd
+//   are Copyright (C) 2007-2008 LShift Ltd, Cohesive Financial
+//   Technologies LLC, and Rabbit Technologies Ltd.
+//
+//   Portions created by LShift Ltd are Copyright (C) 2007-2009 LShift
+//   Ltd. Portions created by Cohesive Financial Technologies LLC are
+//   Copyright (C) 2007-2009 Cohesive Financial Technologies
+//   LLC. Portions created by Rabbit Technologies Ltd are Copyright
+//   (C) 2007-2009 Rabbit Technologies Ltd.
+//
+//   All Rights Reserved.
+//
+//   Contributor(s): ______________________________________.
+//
+package com.rabbitmq.client.test;
+
+import com.rabbitmq.client.*;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+public class QueueingConsumerShutdownTests extends BrokerTestCase{
+  static final String QUEUE = "some-queue";
+  static final int THREADS = 5;
+
+  public void testNThreadShutdown() throws Exception{
+    Channel channel = connection.createChannel();
+    final QueueingConsumer c = new QueueingConsumer(channel);
+    channel.queueDeclare(QUEUE);
+    channel.basicConsume(QUEUE, c);
+    final AtomicInteger count = new AtomicInteger(THREADS);
+    final CountDownLatch latch = new CountDownLatch(THREADS);
+
+    for(int i = 0; i < THREADS; i++){
+      new Thread(){
+        @Override public void run(){
+          try {
+            while(true){
+              c.nextDelivery();
+            }
+          } catch (ShutdownSignalException sig) {
+            count.decrementAndGet();
+          } catch (Exception e) {
+            throw new RuntimeException(e);
+          } finally {
+            latch.countDown();
+          }
+        }
+      }.start();
+    }
+
+    connection.close();
+
+    // Far longer than this could reasonably take
+    assertTrue(latch.await(5, TimeUnit.SECONDS));
+    assertEquals(0, count.get());
+  }
+
+
+}