Use dedicated channel to create resources

This avoids re-using the exact same channel between producers/consumers
when using queue pattern option and exclusive resources.

References #106

Author: acogoluegnes

src/main/java/com/rabbitmq/perf/MulticastParams.java

@@ -354,7 +354,7 @@ public Producer createProducer(Connection connection, Stats stats, MulticastSet.
     public Consumer createConsumer(Connection connection, Stats stats, MulticastSet.CompletionHandler completionHandler) throws IOException {
         TopologyHandlerResult topologyHandlerResult = this.topologyHandler.configureQueuesForClient(connection);
         connection = topologyHandlerResult.connection;
-        Channel channel = topologyHandlerResult.channel;
+        Channel channel = connection.createChannel();
         if (consumerTxSize > 0) channel.txSelect();
         if (consumerPrefetch > 0) channel.basicQos(consumerPrefetch);
         if (channelPrefetch > 0) channel.basicQos(channelPrefetch, true);
@@ -553,20 +553,13 @@ static class TopologyHandlerResult {
          */
         final Connection connection;
 
-        /**
-         * The channel used to create the configured resources.
-         * Must be used by the agent that asked to declare resources.
-         */
-        final Channel channel;
-
         /**
          * The configured queues.
          */
         final List<String> configuredQueues;
 
-        TopologyHandlerResult(Connection connection, Channel channel, List<String> configuredQueues) {
+        TopologyHandlerResult(Connection connection, List<String> configuredQueues) {
             this.connection = connection;
-            this.channel = channel;
             this.configuredQueues = configuredQueues;
         }
     }
@@ -601,7 +594,8 @@ protected Connection maybeUseCachedConnection(List<String> queues, Connection co
             return connectionToUse;
         }
 
-        protected List<String> configureQueues(Connection connection, Channel channel, List<String> queues, Runnable afterQueueConfigurationCallback) throws IOException {
+        protected List<String> configureQueues(Connection connection, List<String> queues, Runnable afterQueueConfigurationCallback) throws IOException {
+            Channel channel = connection.createChannel();
             if (!params.predeclared || !exchangeExists(connection, params.exchangeName)) {
                 channel.exchangeDeclare(params.exchangeName, params.exchangeType);
             }
@@ -632,6 +626,7 @@ protected List<String> configureQueues(Connection connection, Channel channel, L
                 }
                 afterQueueConfigurationCallback.run();
             }
+            channel.abort();
             return generatedQueueNames;
         }
 
@@ -665,22 +660,20 @@ public String getRoutingKey() {
         public TopologyHandlerResult configureQueuesForClient(Connection connection) throws IOException {
             if (this.params.isExclusive()) {
                 Connection connectionToUse = maybeUseCachedConnection(this.queueNames, connection);
-                Channel channel = connectionToUse.createChannel();
                 return new TopologyHandlerResult(
-                    connectionToUse, channel, configureQueues(connectionToUse, channel, this.queueNames, () -> {})
+                    connectionToUse, configureQueues(connectionToUse, this.queueNames, () -> {})
                 );
             } else {
-                Channel channel = connection.createChannel();
                 return new TopologyHandlerResult(
-                    connection, channel, configureQueues(connection, channel, this.queueNames, () -> {})
+                    connection, configureQueues(connection, this.queueNames, () -> {})
                 ) ;
             }
         }
 
         @Override
         public List<String> configureAllQueues(Connection connection) throws IOException {
             if (shouldConfigureQueues() && !this.params.isExclusive()) {
-                return configureQueues(connection, connection.createChannel(), this.queueNames, () -> {});
+                return configureQueues(connection, this.queueNames, () -> {});
             }
             return null;
         }
@@ -729,17 +722,13 @@ public String getRoutingKey() {
         public TopologyHandlerResult configureQueuesForClient(Connection connection) throws IOException {
             if (this.params.isExclusive()) {
                 Connection connectionToUse = maybeUseCachedConnection(getQueueNamesForClient(), connection);
-                Channel channel = connectionToUse.createChannel();
                 return new TopologyHandlerResult(
                     connectionToUse,
-                    channel,
-                    configureQueues(connectionToUse, channel, getQueueNamesForClient(), () -> {})
+                    configureQueues(connectionToUse, getQueueNamesForClient(), () -> {})
                 );
             } else {
-                Channel channel = connection.createChannel();
                 return new TopologyHandlerResult(
                     connection,
-                    channel,
                     getQueueNamesForClient()
                 );
             }
@@ -752,8 +741,7 @@ public List<String> configureAllQueues(Connection connection) throws IOException
             if (this.params.isExclusive()) {
                 return null;
             } else {
-                Channel channel = connection.createChannel();
-                return configureQueues(connection, channel, getQueueNames(), () -> this.next());
+                return configureQueues(connection, getQueueNames(), () -> this.next());
             }
         }
 

src/test/java/com/rabbitmq/perf/TopologyTest.java

@@ -143,7 +143,7 @@ public void defaultParameters()
         set.run();
 
         verify(cf, times(1 + 1 + 1)).newConnection(anyString()); // consumers, producers, configuration (not used)
-        verify(c, times(0 + 1 + 1)).createChannel(); // queue configuration (none), consumer, producer
+        verify(c, times(1 + 1 + 1)).createChannel(); // queue configuration, consumer, producer
         verify(ch, times(1))
             .queueDeclare(eq(""), anyBoolean(), anyBoolean(), anyBoolean(), isNull());
         verify(ch, times(1))
@@ -164,7 +164,7 @@ public void nProducersAndConsumer()
         set.run();
 
         verify(cf, times(10 + 10 + 1)).newConnection(anyString()); // consumers, producers, configuration (not used)
-        verify(c, times(0 + 10 + 10)).createChannel(); // queue configuration (none), consumer, producer
+        verify(c, times(10 + 10 + 10)).createChannel(); // queue configuration, consumer, producer
         verify(ch, times(10))
             .queueDeclare(eq(""), anyBoolean(), anyBoolean(), anyBoolean(), isNull());
         verify(ch, times(10))
@@ -187,7 +187,7 @@ public void producers1Consumers2QueueSpecified() throws Exception {
         set.run();
 
         verify(cf, times(2 + 1 + 1)).newConnection(anyString()); // consumers, producers, configuration (not used)
-        verify(c, times(0 + 2 + 1)).createChannel(); // queue configuration (none), consumer, producer
+        verify(c, times(2 + 2 + 1)).createChannel(); // queue configuration, consumer, producer
         verify(ch, times(2))
             .queueDeclare(eq(queue), anyBoolean(), anyBoolean(), anyBoolean(), isNull());
         verify(ch, times(2))
@@ -210,7 +210,7 @@ public void producers2Consumers4QueueSpecified() throws Exception {
         set.run();
 
         verify(cf, times(4 + 2 + 1)).newConnection(anyString()); // consumers, producers, configuration (not used)
-        verify(c, times(0 + 4 + 2)).createChannel(); // queue configuration (none), consumer, producer
+        verify(c, times(4 + 4 + 2)).createChannel(); // queue configuration, consumer, producer
         verify(ch, times(4))
             .queueDeclare(eq(queue), anyBoolean(), anyBoolean(), anyBoolean(), isNull());
         verify(ch, times(4))
@@ -261,7 +261,7 @@ public void qosIsSetOnTheChannel() throws Exception {
         set.run();
 
         verify(cf, times(1 + 1 + 1)).newConnection(anyString()); // consumers, producers, configuration (not used)
-        verify(c, times(0 + 1 + 1)).createChannel(); // queue configuration (none), consumer, producer
+        verify(c, times(1 + 1 + 1)).createChannel(); // queue configuration, consumer, producer
         verify(ch, times(1))
             .queueDeclare(eq(""), anyBoolean(), anyBoolean(), anyBoolean(), isNull());
         verify(ch, times(1))
@@ -382,7 +382,7 @@ public void exclusiveQueue(String exclusive) throws Exception {
         set.run();
 
         verify(cf, times(1 + 1 + 1)).newConnection(anyString()); // consumers, producers, configuration (not used)
-        verify(c, times(0 + 1 + 1)).createChannel(); // queue configuration (none), consumer, producer
+        verify(c, times(1 + 1 + 1)).createChannel(); // queue configuration, consumer, producer
         verify(ch, times(1))
             .queueDeclare(eq(""), anyBoolean(), eq(valueOf(exclusive)), anyBoolean(), isNull());
         verify(ch, times(1))