merge stable into default

Author: rade

src/com/rabbitmq/client/Channel.java

@@ -26,7 +26,6 @@
 import com.rabbitmq.client.AMQP.Tx;
 import com.rabbitmq.client.AMQP.Basic;
 import com.rabbitmq.client.AMQP.Confirm;
-import com.rabbitmq.client.AMQP.Channel.FlowOk;
 
 /**
  * Public API: Interface to an AMQ channel. See the <a href="http://www.amqp.org/">spec</a> for details.
@@ -90,17 +89,11 @@ public interface Channel extends ShutdownNotifier {
     void close(int closeCode, String closeMessage) throws IOException;
 
     /**
-     * Set flow on the channel
-     *
-     * @param active if true, the server is asked to start sending. If false, the server is asked to stop sending.
-     * @throws IOException
-     */
-    FlowOk flow(boolean active) throws IOException;
-
-    /**
-     * Return the current Channel.Flow settings.
+     * Indicates whether the server has asked this client to stop
+     * sending content-bearing commands (such as basic.publish) by
+     * issueing a channel.flow{active=false}.
      */
-    FlowOk getFlow();
+    boolean flowBlocked();
 
     /**
      * Abort this channel with the {@link com.rabbitmq.client.AMQP#REPLY_SUCCESS} close code

src/com/rabbitmq/client/ConnectionFactory.java

@@ -88,6 +88,7 @@ public class ConnectionFactory implements Cloneable {
     private Map<String, Object> _clientProperties = AMQConnection.defaultClientProperties();
     private SocketFactory factory                 = SocketFactory.getDefault();
     private SaslConfig saslConfig                 = DefaultSaslConfig.PLAIN;
+    private ExecutorService sharedExecutor;
 
     /** @return number of consumer threads in default {@link ExecutorService} */
     @Deprecated
@@ -384,6 +385,19 @@ public void setSocketFactory(SocketFactory factory) {
         this.factory = factory;
     }
 
+    /**
+     * Set the executor to use by default for newly created connections.
+     * All connections that use this executor share it.
+     *
+     * It's developer's responsibility to shut down the executor
+     * when it is no longer needed.
+     *
+     * @param executor
+     */
+    public void setSharedExecutor(ExecutorService executor) {
+        this.sharedExecutor = executor;
+    }
+
     public boolean isSSL(){
         return getSocketFactory() instanceof SSLSocketFactory;
     }
@@ -485,7 +499,7 @@ protected void configureSocket(Socket socket) throws IOException{
      * @throws IOException if it encounters a problem
      */
     public Connection newConnection(Address[] addrs) throws IOException {
-        return newConnection(null, addrs);
+        return newConnection(this.sharedExecutor, addrs);
     }
 
     /**
@@ -530,7 +544,7 @@ public Connection newConnection(ExecutorService executor, Address[] addrs)
      * @throws IOException if it encounters a problem
      */
     public Connection newConnection() throws IOException {
-        return newConnection(null,
+        return newConnection(this.sharedExecutor,
                              new Address[] {new Address(getHost(), getPort())}
                             );
     }

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

@@ -47,7 +47,6 @@
 import com.rabbitmq.client.ShutdownSignalException;
 import com.rabbitmq.client.impl.AMQChannel.BlockingRpcContinuation;
 import com.rabbitmq.utility.BlockingCell;
-import com.rabbitmq.utility.Utility;
 
 final class Copyright {
     final static String COPYRIGHT="Copyright (C) 2007-2013 GoPivotal, Inc.";
@@ -383,8 +382,8 @@ public void start()
 
         try {
             int channelMax =
-                negotiatedMaxValue(this.requestedChannelMax,
-                                   connTune.getChannelMax());
+                negotiateChannelMax(this.requestedChannelMax,
+                                    connTune.getChannelMax());
             _channelManager = new ChannelManager(this._workService, channelMax);
 
             int frameMax =
@@ -422,6 +421,13 @@ public void start()
         return;
     }
 
+    /**
+     * Private API, allows for easier simulation of bogus clients.
+     */
+    protected int negotiateChannelMax(int requestedChannelMax, int serverMax) {
+        return negotiatedMaxValue(requestedChannelMax, serverMax);
+    }
+
     /**
      * Private API - check required preconditions and protocol invariants
      */
@@ -475,6 +481,16 @@ public ExceptionHandler getExceptionHandler() {
         return _exceptionHandler;
     }
 
+
+    /** Public API
+     *
+     * @return true if this work service instance uses its own executor (as opposed to a shared one)
+     */
+    public boolean willShutDownConsumerExecutor() {
+        return this._workService.usesPrivateExecutor();
+    }
+
+
     /** Public API - {@inheritDoc} */
     public Channel createChannel(int channelNumber) throws IOException {
         ensureIsOpen();

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

@@ -1057,13 +1057,8 @@ public Confirm.SelectOk confirmSelect()
     }
 
     /** Public API - {@inheritDoc} */
-    public Channel.FlowOk flow(final boolean a) throws IOException {
-        return (Channel.FlowOk) exnWrappingRpc(new Channel.Flow(a)).getMethod();
-    }
-
-    /** Public API - {@inheritDoc} */
-    public Channel.FlowOk getFlow() {
-        return new Channel.FlowOk(!_blockContent);
+    public boolean flowBlocked() {
+        return _blockContent;
     }
 
     /** Public API - {@inheritDoc} */

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

@@ -22,7 +22,7 @@
 
 import com.rabbitmq.client.Channel;
 
-final class ConsumerWorkService {
+final public class ConsumerWorkService {
     private static final int MAX_RUNNABLE_BLOCK_SIZE = 16;
     private static final int DEFAULT_NUM_THREADS = 5;
     private final ExecutorService executor;
@@ -63,6 +63,14 @@ public void addWork(Channel channel, Runnable runnable) {
         }
     }
 
+    /**
+     * @return true if executor used by this work service is managed
+     *              by it and wasn't provided by the user
+     */
+    public boolean usesPrivateExecutor() {
+        return privateExecutor;
+    }
+
     private final class WorkPoolRunnable implements Runnable {
 
         public void run() {

test/src/com/rabbitmq/client/test/ClientTests.java

@@ -39,6 +39,7 @@ public static TestSuite suite() {
         suite.addTestSuite(AMQBuilderApiTest.class);
         suite.addTestSuite(AmqpUriTest.class);
         suite.addTestSuite(JSONReadWriteTest.class);
+        suite.addTestSuite(SharedThreadPoolTest.class);
         return suite;
     }
 }

test/src/com/rabbitmq/client/test/SharedThreadPoolTest.java

@@ -0,0 +1,30 @@
+package com.rabbitmq.client.test;
+
+import com.rabbitmq.client.ConnectionFactory;
+import com.rabbitmq.client.impl.AMQConnection;
+
+import java.io.IOException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+
+public class SharedThreadPoolTest extends BrokerTestCase {
+    public void testWillShutDownExecutor() throws IOException {
+        ConnectionFactory cf = new ConnectionFactory();
+        ExecutorService executor = Executors.newFixedThreadPool(8);
+        cf.setSharedExecutor(executor);
+
+        AMQConnection conn1 = (AMQConnection)cf.newConnection();
+        assertFalse(conn1.willShutDownConsumerExecutor());
+
+        AMQConnection conn2 = (AMQConnection)cf.newConnection(Executors.newSingleThreadExecutor());
+        assertFalse(conn2.willShutDownConsumerExecutor());
+
+        AMQConnection conn3 = (AMQConnection)cf.newConnection((ExecutorService)null);
+        assertTrue(conn3.willShutDownConsumerExecutor());
+
+        cf.setSharedExecutor(null);
+
+        AMQConnection conn4 = (AMQConnection)cf.newConnection();
+        assertTrue(conn4.willShutDownConsumerExecutor());
+    }
+}

test/src/com/rabbitmq/client/test/functional/QosTests.java

@@ -344,43 +344,6 @@ public void testLimitInheritsUnackedCount()
         drain(c, 1);
     }
 
-    public void testFlow() throws IOException
-    {
-        QueueingConsumer c = new QueueingConsumer(channel);
-        declareBindConsume(c);
-        fill(1);
-        drain(c, 1);
-        channel.flow(false);
-        fill(1);
-        drain(c, 0);
-        channel.flow(true);
-        drain(c, 1);
-    }
-
-    public void testLimitAndFlow() throws IOException
-    {
-        channel.basicQos(1);
-        QueueingConsumer c = new QueueingConsumer(channel);
-        declareBindConsume(c);
-        channel.flow(false);
-        fill(3);
-        drain(c, 0);
-        channel.flow(true);
-        ack(drain(c, 1), false);
-        drain(c, 1);
-        channel.basicQos(0);
-        drain(c, 1);
-    }
-
-    public void testNoConsumers() throws Exception {
-        String q = declareBind(channel);
-        fill(1);
-        channel.flow(false);
-        QueueingConsumer c = new QueueingConsumer(channel);
-        channel.basicConsume(q, c);
-        drain(c, 0);
-    }
-
     public void testRecoverReducesLimit() throws Exception {
         channel.basicQos(2);
         QueueingConsumer c = new QueueingConsumer(channel);

test/src/com/rabbitmq/client/test/performance/QosScaling.java

@@ -129,9 +129,7 @@ public long run() throws IOException {
         channel.basicQos(1);
         QueueingConsumer consumer = new QueueingConsumer(channel);
         try {
-            channel.flow(false);
             publish(consume(consumer));
-            channel.flow(true);
             return drain(consumer);
         } finally {
             connection.abort();

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

@@ -0,0 +1,110 @@
+package com.rabbitmq.client.test.server;
+
+import com.rabbitmq.client.AMQP;
+import com.rabbitmq.client.Connection;
+import com.rabbitmq.client.ConnectionFactory;
+import com.rabbitmq.client.ShutdownListener;
+import com.rabbitmq.client.ShutdownSignalException;
+import com.rabbitmq.client.impl.AMQConnection;
+import com.rabbitmq.client.impl.ChannelN;
+import com.rabbitmq.client.impl.DefaultExceptionHandler;
+import com.rabbitmq.client.impl.SocketFrameHandler;
+import com.rabbitmq.client.impl.ConsumerWorkService;
+import com.rabbitmq.client.test.BrokerTestCase;
+import com.rabbitmq.tools.Host;
+
+import javax.net.SocketFactory;
+import java.io.IOException;
+import java.util.concurrent.Executors;
+
+public class ChannelLimitNegotiation extends BrokerTestCase {
+    class SpecialConnection extends AMQConnection {
+        private final int channelMax;
+
+        public SpecialConnection(int channelMax) throws Exception {
+            this(new ConnectionFactory(), channelMax);
+        }
+
+        private SpecialConnection(ConnectionFactory factory, int channelMax) throws Exception {
+            super(factory.getUsername(),
+                  factory.getPassword(),
+                  new SocketFrameHandler(SocketFactory.getDefault().createSocket("localhost", AMQP.PROTOCOL.PORT)),
+                  Executors.newFixedThreadPool(1),
+                  factory.getVirtualHost(),
+                  factory.getClientProperties(),
+                  factory.getRequestedFrameMax(),
+                  channelMax,
+                  factory.getRequestedHeartbeat(),
+                  factory.getSaslConfig(),
+                  new DefaultExceptionHandler());
+
+            this.channelMax = channelMax;
+        }
+
+        /**
+         * Private API, allows for easier simulation of bogus clients.
+         */
+        @Override
+        protected int negotiateChannelMax(int requestedChannelMax, int serverMax) {
+            return this.channelMax;
+        }
+    }
+
+    public void testChannelMaxLowerThanServerMinimum() throws Exception {
+        int n = 64;
+        ConnectionFactory cf = new ConnectionFactory();
+        cf.setRequestedChannelMax(n);
+
+        Connection conn = cf.newConnection();
+        assertEquals(n, conn.getChannelMax());
+    }
+
+    public void testChannelMaxGreaterThanServerValue() throws Exception {
+        try {
+            Host.rabbitmqctl("eval 'application:set_env(rabbit, channel_max, 2048).'");
+
+            SpecialConnection connection = new SpecialConnection(4096);
+            try {
+                connection.start();
+                fail("expected failure during connection negotiation");
+            } catch (IOException e) {
+                // expected
+            }
+        } finally {
+            Host.rabbitmqctl("eval 'application:set_env(rabbit, channel_max, 0).'");
+        }
+    }
+
+    public void testOpeningTooManyChannels() throws Exception {
+        int n = 48;
+
+        try {
+            Host.rabbitmqctl("eval 'application:set_env(rabbit, channel_max, " + n + ").'");
+            ConnectionFactory cf = new ConnectionFactory();
+            Connection conn = cf.newConnection();
+            assertEquals(n, conn.getChannelMax());
+
+            for (int i = 1; i <= n; i++) {
+                assertNotNull(conn.createChannel(i));
+            }
+            // ChannelManager guards against channel.open being sent
+            assertNull(conn.createChannel(n + 1));
+
+            // Construct a channel directly
+            final ChannelN ch = new ChannelN((AMQConnection) conn, n + 1,
+                                             new ConsumerWorkService(Executors.newSingleThreadExecutor()));
+            conn.addShutdownListener(new ShutdownListener() {
+                public void shutdownCompleted(ShutdownSignalException cause) {
+                    // make sure channel.open continuation is released
+                    ch.processShutdownSignal(cause, true, true);
+                }
+            });
+            ch.open();
+            fail("expected channel.open to cause a connection exception");
+        } catch (IOException e) {
+            checkShutdownSignal(530, e);
+        } finally {
+            Host.rabbitmqctl("eval 'application:set_env(rabbit, channel_max, 0).'");
+        }
+    }
+}