Make ConsumerWorkService use connection's ThreadFactory

Author: michaelklishin

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

@@ -26,6 +26,7 @@
 import java.util.HashMap;
 import java.util.Map;
 import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.ExecutorService;
 import java.util.concurrent.TimeoutException;
 
 import com.rabbitmq.client.AMQP;
@@ -62,6 +63,7 @@ final class Copyright {
 public class AMQConnection extends ShutdownNotifierComponent implements Connection, NetworkConnection {
     /** Timeout used while waiting for AMQP handshaking to complete (milliseconds) */
     public static final int HANDSHAKE_TIMEOUT = 10000;
+    private final ExecutorService executor;
     private Thread mainLoopThread;
     private ThreadFactory threadFactory = new DefaultThreadFactory();
 
@@ -104,7 +106,7 @@ public static final Map<String, Object> defaultClientProperties() {
         }
     };
 
-    protected final ConsumerWorkService _workService;
+    protected ConsumerWorkService _workService = null;
 
     /** Frame source/sink */
     private final FrameHandler _frameHandler;
@@ -215,17 +217,21 @@ public AMQConnection(ConnectionParams params, FrameHandler frameHandler)
         this.requestedChannelMax = params.getRequestedChannelMax();
         this.requestedHeartbeat = params.getRequestedHeartbeat();
         this.saslConfig = params.getSaslConfig();
+        this.executor = params.getExecutor();
 
-        this._workService  = new ConsumerWorkService(params.getExecutor());
         this._channelManager = null;
 
         this._brokerInitiatedShutdown = false;
 
         this._inConnectionNegotiation = true; // we start out waiting for the first protocol response
     }
 
-    private void initializeHeartbeatSender(FrameHandler frameHandler) {
-        this._heartbeatSender = new HeartbeatSender(frameHandler, threadFactory);
+    private void initializeConsumerWorkService() {
+        this._workService  = new ConsumerWorkService(executor, threadFactory);
+    }
+
+    private void initializeHeartbeatSender() {
+        this._heartbeatSender = new HeartbeatSender(_frameHandler, threadFactory);
     }
 
     /**
@@ -247,7 +253,8 @@ private void initializeHeartbeatSender(FrameHandler frameHandler) {
     public void start()
         throws IOException
     {
-        initializeHeartbeatSender(_frameHandler);
+        initializeConsumerWorkService();
+        initializeHeartbeatSender();
         this._running = true;
         // Make sure that the first thing we do is to send the header,
         // which should cause any socket errors to show up for us, rather

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

@@ -19,6 +19,7 @@
 import java.util.List;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
+import java.util.concurrent.ThreadFactory;
 
 import com.rabbitmq.client.Channel;
 
@@ -29,9 +30,9 @@ final public class ConsumerWorkService {
     private final boolean privateExecutor;
     private final WorkPool<Channel, Runnable> workPool;
 
-    public ConsumerWorkService(ExecutorService executor) {
+    public ConsumerWorkService(ExecutorService executor, ThreadFactory threadFactory) {
         this.privateExecutor = (executor == null);
-        this.executor = (executor == null) ? Executors.newFixedThreadPool(DEFAULT_NUM_THREADS)
+        this.executor = (executor == null) ? Executors.newFixedThreadPool(DEFAULT_NUM_THREADS, threadFactory)
                                            : executor;
         this.workPool = new WorkPool<Channel, Runnable>();
     }

test/src/com/rabbitmq/client/test/server/ChannelLimitNegotiation.java

@@ -7,6 +7,7 @@
 import com.rabbitmq.client.ShutdownSignalException;
 import com.rabbitmq.client.impl.AMQConnection;
 import com.rabbitmq.client.impl.ChannelN;
+import com.rabbitmq.client.impl.DefaultThreadFactory;
 import com.rabbitmq.client.impl.SocketFrameHandler;
 import com.rabbitmq.client.impl.ConsumerWorkService;
 import com.rabbitmq.client.test.BrokerTestCase;
@@ -80,7 +81,7 @@ public void testOpeningTooManyChannels() throws Exception {
 
             // Construct a channel directly
             final ChannelN ch = new ChannelN((AMQConnection) conn, n + 1,
-                                             new ConsumerWorkService(Executors.newSingleThreadExecutor()));
+                                             new ConsumerWorkService(Executors.newSingleThreadExecutor(), new DefaultThreadFactory()));
             conn.addShutdownListener(new ShutdownListener() {
                 public void shutdownCompleted(ShutdownSignalException cause) {
                     // make sure channel.open continuation is released