merge default into bug25999

Author: michaelklishin

src/com/rabbitmq/client/ConnectionFactory.java

@@ -32,6 +32,7 @@
 import javax.net.ssl.TrustManager;
 
 import com.rabbitmq.client.impl.AMQConnection;
+import com.rabbitmq.client.impl.DefaultThreadFactory;
 import com.rabbitmq.client.impl.ConnectionParams;
 import com.rabbitmq.client.impl.DefaultExceptionHandler;
 import com.rabbitmq.client.impl.FrameHandler;
@@ -90,6 +91,7 @@ public class ConnectionFactory implements Cloneable {
     private SocketFactory factory                 = SocketFactory.getDefault();
     private SaslConfig saslConfig                 = DefaultSaslConfig.PLAIN;
     private ExecutorService sharedExecutor;
+    private ThreadFactory threadFactory           = new DefaultThreadFactory();
     private SocketConfigurator socketConf         = new DefaultSocketConfigurator();
     private ExceptionHandler exceptionHandler     = new DefaultExceptionHandler();
 
@@ -429,20 +431,40 @@ public void setSharedExecutor(ExecutorService executor) {
     }
 
     /**
-     * Get the exception handler.
-     *
-     * @see com.rabbitmq.client.ExceptionHandler
+     * Retrieve the thread factory used to instantiate new threads.
+     * @see com.rabbitmq.client.ThreadFactory
+     * @see com.rabbitmq.client.impl.DefaultThreadFactory
+     */
+    public ThreadFactory getThreadFactory() {
+        return threadFactory;
+    }
+
+    /**
+     * Set the thread factory used to instantiate new threads.
+     * @see com.rabbitmq.client.ThreadFactory
+     * @see com.rabbitmq.client.impl.DefaultThreadFactory
      */
+    public void setThreadFactory(ThreadFactory threadFactory) {
+        this.threadFactory = threadFactory;
+    }
+
+    /**
+    * Get the exception handler.
+    *
+    * @see com.rabbitmq.client.ExceptionHandler
+    */
     public ExceptionHandler getExceptionHandler() {
         return exceptionHandler;
     }
 
     /**
+     * Set the thread factory to use for newly created connections.
+     * @see com.rabbitmq.client.ThreadFactory
+     * @see com.rabbitmq.client.impl.DefaultThreadFactory
      * Set the exception handler to use for newly created connections.
      *
      * @see com.rabbitmq.client.ExceptionHandler
      */
-
     public void setExceptionHandler(ExceptionHandler exceptionHandler) {
         if (exceptionHandler == null) {
           throw new IllegalArgumentException("exception handler cannot be null!");
@@ -583,7 +605,7 @@ public Connection newConnection(ExecutorService executor, Address[] addrs)
     public ConnectionParams params(ExecutorService executor) {
         return new ConnectionParams(username, password, executor, virtualHost, getClientProperties(),
                                     requestedFrameMax, requestedChannelMax, requestedHeartbeat, saslConfig,
-                                    networkRecoveryInterval, topologyRecovery, exceptionHandler);
+                                    networkRecoveryInterval, topologyRecovery, exceptionHandler, threadFactory);
     }
 
     /**

src/com/rabbitmq/client/ThreadFactory.java

@@ -0,0 +1,20 @@
+package com.rabbitmq.client;
+
+/**
+ * Extends {@link java.util.concurrent.ThreadFactory} to make it possible to specify
+ * thread name.
+ *
+ * In environments with restricted thread management (e.g. Google App Engine), developers
+ * can provide a custom factory to control how network I/O thread is created.
+ */
+public interface ThreadFactory extends java.util.concurrent.ThreadFactory {
+    /**
+     * Like {@link java.util.concurrent.ThreadFactory#newThread(Runnable)} but also takes
+     * a thread name.
+     *
+     * @param r runnable to execute
+     * @param threadName thread name
+     * @return a new thread
+     */
+    Thread newThread(Runnable r, String threadName);
+}

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;
@@ -45,6 +46,7 @@
 import com.rabbitmq.client.SaslConfig;
 import com.rabbitmq.client.SaslMechanism;
 import com.rabbitmq.client.ShutdownSignalException;
+import com.rabbitmq.client.ThreadFactory;
 import com.rabbitmq.client.impl.AMQChannel.BlockingRpcContinuation;
 import com.rabbitmq.utility.BlockingCell;
 
@@ -62,6 +64,9 @@ 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();
 
     /**
      * Retrieve a copy of the default table of client properties that
@@ -102,7 +107,7 @@ public static final Map<String, Object> defaultClientProperties() {
         }
     };
 
-    protected final ConsumerWorkService _workService;
+    protected ConsumerWorkService _workService = null;
 
     /** Frame source/sink */
     private final FrameHandler _frameHandler;
@@ -125,7 +130,7 @@ public static final Map<String, Object> defaultClientProperties() {
     private volatile boolean _inConnectionNegotiation;
 
     /** Manages heart-beat sending for this connection */
-    private final HeartbeatSender _heartbeatSender;
+    private HeartbeatSender _heartbeatSender;
 
     private final String _virtualHost;
     private final Map<String, Object> _clientProperties;
@@ -212,17 +217,25 @@ public AMQConnection(ConnectionParams params, FrameHandler frameHandler)
         this.requestedChannelMax = params.getRequestedChannelMax();
         this.requestedHeartbeat = params.getRequestedHeartbeat();
         this.saslConfig = params.getSaslConfig();
+        this.executor = params.getExecutor();
+        this.threadFactory = params.getThreadFactory();
 
-        this._workService  = new ConsumerWorkService(params.getExecutor());
         this._channelManager = null;
 
-        this._heartbeatSender = new HeartbeatSender(frameHandler);
         this._brokerInitiatedShutdown = false;
 
         this._inConnectionNegotiation = true; // we start out waiting for the first protocol response
     }
 
-  /**
+    private void initializeConsumerWorkService() {
+        this._workService  = new ConsumerWorkService(executor, threadFactory);
+    }
+
+    private void initializeHeartbeatSender() {
+        this._heartbeatSender = new HeartbeatSender(_frameHandler, threadFactory);
+    }
+
+    /**
      * Start up the connection, including the MainLoop thread.
      * Sends the protocol
      * version negotiation header, and runs through
@@ -241,6 +254,8 @@ public AMQConnection(ConnectionParams params, FrameHandler frameHandler)
     public void start()
         throws IOException
     {
+        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
@@ -264,7 +279,9 @@ public void start()
         }
 
         // start the main loop going
-        new MainLoop("AMQP Connection " + getHostAddress() + ":" + getPort()).start();
+        MainLoop loop = new MainLoop();
+        mainLoopThread = threadFactory.newThread(loop, "AMQP Connection " + getHostAddress() + ":" + getPort());
+        mainLoopThread.start();
         // after this point clear-up of MainLoop is triggered by closing the frameHandler.
 
         AMQP.Connection.Start connStart = null;
@@ -337,7 +354,7 @@ public void start()
             int channelMax =
                 negotiateChannelMax(this.requestedChannelMax,
                                     connTune.getChannelMax());
-            _channelManager = instantiateChannelManager(channelMax);
+            _channelManager = instantiateChannelManager(channelMax, threadFactory);
 
             int frameMax =
                 negotiatedMaxValue(this.requestedFrameMax,
@@ -374,8 +391,8 @@ public void start()
         return;
     }
 
-    protected ChannelManager instantiateChannelManager(int channelMax) {
-        return new ChannelManager(this._workService, channelMax);
+    protected ChannelManager instantiateChannelManager(int channelMax, ThreadFactory threadFactory) {
+        return new ChannelManager(this._workService, channelMax, threadFactory);
     }
 
     /**
@@ -427,6 +444,23 @@ public void setHeartbeat(int heartbeat) {
         }
     }
 
+    /**
+     * Makes it possible to override thread factory that is used
+     * to instantiate connection network I/O loop. Only necessary
+     * in the environments with restricted
+     * @param threadFactory
+     */
+    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);
     }
@@ -485,14 +519,7 @@ private static final int negotiatedMaxValue(int clientValue, int serverValue) {
             Math.min(clientValue, serverValue);
     }
 
-    private class MainLoop extends Thread {
-
-        /**
-         * @param name of thread
-         */
-        MainLoop(String name) {
-            super(name);
-        }
+    private class MainLoop implements Runnable {
 
         /**
          * Channel reader thread main loop. Reads a frame, and if it is
@@ -636,13 +663,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) {
@@ -789,7 +817,7 @@ public void close(int closeCode,
                       boolean abort)
         throws IOException
     {
-        boolean sync = !(Thread.currentThread() instanceof MainLoop);
+        boolean sync = !(Thread.currentThread() == mainLoopThread);
 
         try {
             AMQP.Connection.Close reason =

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 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,14 +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) { /*ignored*/ }
                 }
                 ssWorkService.shutdown();
-            }}, "ConsumerWorkServiceShutdown");
-        shutdownThread.setDaemon(true);
+            }
+        };
+        Thread shutdownThread = threadFactory.newThread(target, "ConsumerWorkService shutdown monitor");
+        if(Environment.isAllowedToModifyThreads()) {
+            shutdownThread.setDaemon(true);
+        }
         shutdownThread.start();
     }
 

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

@@ -2,6 +2,7 @@
 
 import com.rabbitmq.client.ExceptionHandler;
 import com.rabbitmq.client.SaslConfig;
+import com.rabbitmq.client.ThreadFactory;
 
 import java.util.Map;
 import java.util.concurrent.ExecutorService;
@@ -20,8 +21,9 @@ public class ConnectionParams {
     private final boolean topologyRecovery;
 
     private ExceptionHandler exceptionHandler;
+  private ThreadFactory threadFactory;
 
-    /**
+  /**
      * @param username name used to establish connection
      * @param password for <code><b>username</b></code>
      * @param executor thread pool service for consumer threads for channels on this connection
@@ -33,13 +35,17 @@ public class ConnectionParams {
      * @param saslConfig sasl configuration hook
      * @param networkRecoveryInterval interval used when recovering from network failure
      * @param topologyRecovery should topology (queues, exchanges, bindings, consumers) recovery be performed?
+<<<<<<< local
+     * @param threadFactory
+=======
      * @param exceptionHandler
+>>>>>>> other
      */
     public ConnectionParams(String username, String password, ExecutorService executor,
                             String virtualHost, Map<String, Object> clientProperties,
                             int requestedFrameMax, int requestedChannelMax, int requestedHeartbeat,
                             SaslConfig saslConfig, int networkRecoveryInterval,
-                            boolean topologyRecovery, ExceptionHandler exceptionHandler) {
+                            boolean topologyRecovery, ExceptionHandler exceptionHandler, ThreadFactory threadFactory) {
         this.username = username;
         this.password = password;
         this.executor = executor;
@@ -52,6 +58,7 @@ public ConnectionParams(String username, String password, ExecutorService execut
         this.networkRecoveryInterval = networkRecoveryInterval;
         this.topologyRecovery = topologyRecovery;
         this.exceptionHandler = exceptionHandler;
+        this.threadFactory = threadFactory;
     }
 
     public String getUsername() {
@@ -101,4 +108,8 @@ public int getNetworkRecoveryInterval() {
     public boolean isTopologyRecoveryEnabled() {
         return topologyRecovery;
     }
+
+  public ThreadFactory getThreadFactory() {
+    return threadFactory;
+  }
 }