Don't instantiate threads directly in two more places

Author: michaelklishin

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

@@ -389,7 +389,7 @@ public void start()
             int channelMax =
                 negotiateChannelMax(this.requestedChannelMax,
                                     connTune.getChannelMax());
-            _channelManager = new ChannelManager(this._workService, channelMax);
+            _channelManager = new ChannelManager(this._workService, channelMax, threadFactory);
 
             int frameMax =
                 negotiatedMaxValue(this.requestedFrameMax,
@@ -485,6 +485,13 @@ public void setThreadFactory(ThreadFactory threadFactory) {
         this.threadFactory = threadFactory;
     }
 
+    /**
+     * @return Thread factory used by this connection.
+     */
+    public ThreadFactory getThreadFactory() {
+        return threadFactory;
+    }
+
     public Map<String, Object> getClientProperties() {
         return new HashMap<String, Object>(_clientProperties);
     }
@@ -687,13 +694,14 @@ public void handleConnectionClose(Command closeCommand) {
             _channel0.quiescingTransmit(new AMQP.Connection.CloseOk.Builder().build());
         } catch (IOException _) { } // ignore
         _brokerInitiatedShutdown = true;
-        Thread scw = new SocketCloseWait(sse);
-        scw.setName("AMQP Connection Closing Monitor " +
+        SocketCloseWait scw = new SocketCloseWait(sse);
+        Thread waiter = threadFactory.newThread(scw);
+        waiter.setName("AMQP Connection Closing Monitor " +
                 getHostAddress() + ":" + getPort());
-        scw.start();
+        waiter.start();
     }
 
-    private class SocketCloseWait extends Thread {
+    private class SocketCloseWait implements Runnable {
         private final ShutdownSignalException cause;
 
         public SocketCloseWait(ShutdownSignalException sse) {

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

@@ -25,6 +25,7 @@
 import java.util.concurrent.TimeUnit;
 
 import com.rabbitmq.client.ShutdownSignalException;
+import com.rabbitmq.client.ThreadFactory;
 import com.rabbitmq.utility.IntAllocator;
 
 /**
@@ -45,12 +46,17 @@ public final class ChannelManager {
 
     /** Maximum channel number available on this connection. */
     private final int _channelMax;
+    private final ThreadFactory threadFactory;
 
     public int getChannelMax(){
       return _channelMax;
     }
 
     public ChannelManager(ConsumerWorkService workService, int channelMax) {
+        this(workService, channelMax, new DefaultThreadFactory());
+    }
+
+    public ChannelManager(ConsumerWorkService workService, int channelMax, ThreadFactory threadFactory) {
         if (channelMax == 0) {
             // The framing encoding only allows for unsigned 16-bit integers
             // for the channel number
@@ -60,6 +66,7 @@ public ChannelManager(ConsumerWorkService workService, int channelMax) {
         channelNumberAllocator = new IntAllocator(1, channelMax);
 
         this.workService = workService;
+        this.threadFactory = threadFactory;
     }
 
     /**
@@ -97,13 +104,18 @@ public void handleSignal(ShutdownSignalException signal) {
     private void scheduleShutdownProcessing() {
         final Set<CountDownLatch> sdSet = new HashSet<CountDownLatch>(shutdownSet);
         final ConsumerWorkService ssWorkService = workService;
-        Thread shutdownThread = new Thread( new Runnable() {
+        Runnable target = new Runnable() {
             public void run() {
                 for (CountDownLatch latch : sdSet) {
-                    try { latch.await(SHUTDOWN_TIMEOUT_SECONDS, TimeUnit.SECONDS); } catch (Throwable e) { }
+                    try {
+                        latch.await(SHUTDOWN_TIMEOUT_SECONDS, TimeUnit.SECONDS);
+                    } catch (Throwable e) {
+                    }
                 }
                 ssWorkService.shutdown();
-            }}, "ConsumerWorkServiceShutdown");
+            }
+        };
+        Thread shutdownThread = threadFactory.newThread(target, "ConsumerWorkService shutdown monitor");
         shutdownThread.setDaemon(true);
         shutdownThread.start();
     }