merge bug14587 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.FrameHandler;
 import com.rabbitmq.client.impl.FrameHandlerFactory;
@@ -89,6 +90,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 boolean automaticRecovery             = false;
@@ -426,6 +428,24 @@ public void setSharedExecutor(ExecutorService executor) {
         this.sharedExecutor = executor;
     }
 
+    /**
+     * 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 to use for newly created connections.
+     * @see com.rabbitmq.client.ThreadFactory
+     * @see com.rabbitmq.client.impl.DefaultThreadFactory
+     */
+    public void setThreadFactory(ThreadFactory threadFactory) {
+        this.threadFactory = threadFactory;
+    }
+
     public boolean isSSL(){
         return getSocketFactory() instanceof SSLSocketFactory;
     }
@@ -544,6 +564,7 @@ public Connection newConnection(ExecutorService executor, Address[] addrs)
                 try {
                     FrameHandler handler = fhFactory.create(addr);
                     AMQConnection conn = new AMQConnection(params, handler);
+                    conn.setThreadFactory(this.threadFactory);
                     conn.start();
                     return conn;
                 } catch (IOException e) {

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

@@ -44,6 +44,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;
 
@@ -61,6 +62,8 @@ 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 Thread mainLoopThread;
+    private ThreadFactory threadFactory = new DefaultThreadFactory();
 
     /**
      * Retrieve a copy of the default table of client properties that
@@ -216,7 +219,7 @@ public AMQConnection(ConnectionParams params, FrameHandler frameHandler)
         this._workService  = new ConsumerWorkService(params.getExecutor());
         this._channelManager = null;
 
-        this._heartbeatSender = new HeartbeatSender(frameHandler);
+        this._heartbeatSender = new HeartbeatSender(frameHandler, threadFactory);
         this._brokerInitiatedShutdown = false;
 
         this._inConnectionNegotiation = true; // we start out waiting for the first protocol response
@@ -264,7 +267,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 +342,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 +379,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 +432,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 +507,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 +651,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 +805,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/DefaultThreadFactory.java

@@ -0,0 +1,18 @@
+package com.rabbitmq.client.impl;
+
+import com.rabbitmq.client.ThreadFactory;
+
+/**
+ * Default thread factory that instantiates {@link java.lang.Thread}s directly.
+ */
+public class DefaultThreadFactory implements ThreadFactory {
+    @Override
+    public Thread newThread(Runnable r, String threadName) {
+        return new Thread(r, threadName);
+    }
+
+    @Override
+    public Thread newThread(Runnable r) {
+        return new Thread(r);
+    }
+}

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

@@ -0,0 +1,18 @@
+package com.rabbitmq.client.impl;
+
+/**
+ * Infers information about the execution environment, e.g.
+ * security permissions.
+ */
+class Environment {
+    public static boolean isAllowedToModifyThreads() {
+        SecurityManager sm = new SecurityManager();
+        try {
+            sm.checkPermission(new RuntimePermission("modifyThread"));
+            sm.checkPermission(new RuntimePermission("modifyThreadGroup"));
+            return true;
+        } catch (SecurityException se) {
+            return false;
+        }
+    }
+}

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

@@ -22,6 +22,7 @@
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.Executors;
+import java.util.concurrent.ThreadFactory;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.ScheduledFuture;
 import java.io.IOException;
@@ -39,6 +40,7 @@ final class HeartbeatSender {
     private final Object monitor = new Object();
 
     private final FrameHandler frameHandler;
+    private final ThreadFactory threadFactory;
 
     private ScheduledExecutorService executor;
 
@@ -48,8 +50,9 @@ final class HeartbeatSender {
 
     private volatile long lastActivityTime;
 
-    HeartbeatSender(FrameHandler frameHandler) {
+    HeartbeatSender(FrameHandler frameHandler, ThreadFactory threadFactory) {
         this.frameHandler = frameHandler;
+        this.threadFactory = threadFactory;
     }
 
     public void signalActivity() {
@@ -86,7 +89,7 @@ public void setHeartbeat(int heartbeatSeconds) {
     private ScheduledExecutorService createExecutorIfNecessary() {
         synchronized (this.monitor) {
             if (this.executor == null) {
-                this.executor = Executors.newSingleThreadScheduledExecutor();
+                this.executor = Executors.newSingleThreadScheduledExecutor(threadFactory);
             }
             return this.executor;
         }

src/com/rabbitmq/client/impl/recovery/RecoveryAwareAMQConnection.java

@@ -1,5 +1,6 @@
 package com.rabbitmq.client.impl.recovery;
 
+import com.rabbitmq.client.ThreadFactory;
 import com.rabbitmq.client.impl.AMQConnection;
 import com.rabbitmq.client.impl.ConnectionParams;
 import com.rabbitmq.client.impl.FrameHandler;
@@ -14,7 +15,7 @@ public RecoveryAwareAMQConnection(ConnectionParams params, FrameHandler handler)
     }
 
     @Override
-    protected RecoveryAwareChannelManager instantiateChannelManager(int channelMax) {
+    protected RecoveryAwareChannelManager instantiateChannelManager(int channelMax, ThreadFactory threadFactory) {
         return new RecoveryAwareChannelManager(super._workService, channelMax);
     }
 }