Merge bug24579

Author: Simon MacMullen

src/com/rabbitmq/client/ConnectionFactory.java

@@ -21,7 +21,6 @@
 import java.security.NoSuchAlgorithmException;
 import java.util.Map;
 import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Executors;
 
 import java.net.InetSocketAddress;
 import java.net.Socket;
@@ -43,8 +42,9 @@
  */
 
 public class ConnectionFactory implements Cloneable {
-    
+
     /** Default Executor threads */
+    @Deprecated
     public static final int    DEFAULT_NUM_CONSUMER_THREADS = 5;
     /** Default user name */
     public static final String DEFAULT_USER = "guest";
@@ -76,7 +76,6 @@ public class ConnectionFactory implements Cloneable {
     /** The default SSL protocol */
     private static final String DEFAULT_SSL_PROTOCOL = "SSLv3";
 
-    private int numConsumerThreads                = DEFAULT_NUM_CONSUMER_THREADS;
     private String username                       = DEFAULT_USER;
     private String password                       = DEFAULT_PASS;
     private String virtualHost                    = DEFAULT_VHOST;
@@ -91,13 +90,15 @@ public class ConnectionFactory implements Cloneable {
     private SaslConfig saslConfig                 = DefaultSaslConfig.PLAIN;
 
     /** @return number of consumer threads in default {@link ExecutorService} */
+    @Deprecated
     public int getNumConsumerThreads() {
-        return numConsumerThreads;
+        return DEFAULT_NUM_CONSUMER_THREADS;
     }
 
     /** @param numConsumerThreads threads in created private executor service */
+    @Deprecated
     public void setNumConsumerThreads(int numConsumerThreads) {
-        this.numConsumerThreads = numConsumerThreads;
+        throw new IllegalArgumentException("setNumConsumerThreads not supported -- create explicit ExecutorService instead.");
     }
 
     /** @return the default host to use for connections */
@@ -472,8 +473,7 @@ protected void configureSocket(Socket socket) throws IOException{
      * @throws IOException if it encounters a problem
      */
     public Connection newConnection(Address[] addrs) throws IOException {
-        return newConnection(Executors.newFixedThreadPool(this.numConsumerThreads),
-            addrs);
+        return newConnection(null, addrs);
     }
 
     /**
@@ -490,7 +490,7 @@ public Connection newConnection(ExecutorService executor, Address[] addrs)
         for (Address addr : addrs) {
             try {
                 FrameHandler frameHandler = createFrameHandler(addr);
-                AMQConnection conn = 
+                AMQConnection conn =
                     new AMQConnection(username,
                                       password,
                                       frameHandler,
@@ -518,7 +518,7 @@ public Connection newConnection(ExecutorService executor, Address[] addrs)
      * @throws IOException if it encounters a problem
      */
     public Connection newConnection() throws IOException {
-        return newConnection(Executors.newFixedThreadPool(this.numConsumerThreads),
+        return newConnection(null,
                              new Address[] {new Address(getHost(), getPort())}
                             );
     }

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

@@ -107,8 +107,7 @@ public class ChannelN extends AMQChannel implements com.rabbitmq.client.Channel
      * @see Connection#createChannel
      * @param connection The connection associated with this channel
      * @param channelNumber The channel number to be associated with this channel
-     * @param workPool pool in which this channel's consumer work is stored
-     * @param executor service which executes the work in the workPool
+     * @param workService service for managing this channel's consumer callbacks
      */
     public ChannelN(AMQConnection connection, int channelNumber,
                     ConsumerWorkService workService) {
@@ -651,7 +650,7 @@ public Exchange.DeleteOk exchangeDelete(String exchange)
     public Exchange.BindOk exchangeBind(String destination, String source,
             String routingKey, Map<String, Object> arguments)
             throws IOException {
-        return (Exchange.BindOk) 
+        return (Exchange.BindOk)
                exnWrappingRpc(new Exchange.Bind.Builder()
                                .destination(destination)
                                .source(source)

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

@@ -18,16 +18,25 @@
 import java.util.ArrayList;
 import java.util.List;
 import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
 
 import com.rabbitmq.client.Channel;
 
 final class ConsumerWorkService {
     private static final int MAX_RUNNABLE_BLOCK_SIZE = 16;
+    private static final int DEFAULT_NUM_THREADS = 5;
     private final ExecutorService executor;
+    private final boolean privateExecutor;
     private final WorkPool<Channel, Runnable> workPool;
 
     public ConsumerWorkService(ExecutorService executor) {
-        this.executor = executor;
+        if (executor == null) {
+            privateExecutor = true;
+            this.executor = Executors.newFixedThreadPool(DEFAULT_NUM_THREADS);
+        } else {
+            privateExecutor = false;
+            this.executor = executor;
+        }
         this.workPool = new WorkPool<Channel, Runnable>();
     }
 
@@ -36,7 +45,8 @@ public ConsumerWorkService(ExecutorService executor) {
      */
     public void shutdown() {
         this.workPool.unregisterAllKeys();
-        this.executor.shutdown();
+        if (privateExecutor)
+            this.executor.shutdown();
     }
 
     /**