Stop channel consumer work pool after channel.close()

Author: Steve Powell

src/com/rabbitmq/client/impl/ChannelManager.java

@@ -59,7 +59,7 @@ public ChannelManager(ConsumerWorkService workService, int channelMax) {
         }
         _channelMax = channelMax;
         channelNumberAllocator = new IntAllocator(1, channelMax);
-        
+
         this.workService = workService;
     }
 

src/com/rabbitmq/client/impl/ConsumerDispatcher.java

@@ -27,12 +27,11 @@
 import java.util.concurrent.CountDownLatch;
 
 /**
- * Dispatches notifications to a {@link Consumer} on an
- * internally-managed executor service and work pool.
+ * Dispatches notifications to a {@link Consumer} on an internally-managed executor service and work
+ * pool.
  * <p/>
- * Each {@link Channel} has a single {@link ConsumerDispatcher},
- * but the executor service and work pool may be shared with other channels, typically those on the same
- * {@link Connection}.
+ * Each {@link Channel} has a single {@link ConsumerDispatcher}, but the executor service and work
+ * pool may be shared with other channels, typically those on the same {@link AMQConnection}.
  */
 final class ConsumerDispatcher {
 
@@ -147,6 +146,7 @@ public CountDownLatch handleShutdownSignal(final Map<String, Consumer> consumers
                 public void run() {
                     notifyConsumersOfShutdown(consumers, signal);
                     ConsumerDispatcher.this.shutdown(signal);
+                    ConsumerDispatcher.this.workService.stopWork(ConsumerDispatcher.this.channel);
                     latch.countDown();
                 }
             });

src/com/rabbitmq/client/impl/ConsumerWorkService.java

@@ -51,6 +51,7 @@ public void shutdown() {
 
     /**
      * Stop executing all consumer work for a particular channel
+     * @param channel to stop consumer work for
      */
     public void stopWork(Channel channel) {
         this.workPool.unregisterKey(channel);