Merge bug22525

Author: Simon MacMullen

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

@@ -382,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 =
@@ -421,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
      */

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;

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).'");
+        }
+    }
+}

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

@@ -41,5 +41,6 @@ public static void add(TestSuite suite) {
         suite.addTestSuite(PersistenceGuarantees.class);
         suite.addTestSuite(Shutdown.class);
         suite.addTestSuite(BlockedConnection.class);
+        suite.addTestSuite(ChannelLimitNegotiation.class);
     }
 }