Test recovery with Netty

Author: acogoluegnes

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

@@ -32,7 +32,6 @@
 import com.rabbitmq.client.impl.CredentialsRefreshService;
 import com.rabbitmq.client.impl.DefaultCredentialsRefreshService.DefaultCredentialsRefreshServiceBuilder;
 import com.rabbitmq.client.impl.DefaultExceptionHandler;
-import com.rabbitmq.client.impl.nio.NioParams;
 import com.rabbitmq.perf.Metrics.ConfigurationContext;
 import com.rabbitmq.perf.Utils.GsonOAuth2ClientCredentialsGrantCredentialsProvider;
 import com.rabbitmq.perf.metrics.CompositeMetricsFormatter;
@@ -797,7 +796,8 @@ private static ConnectionFactory configureNioIfRequested(
     int nbThreads = Utils.intArg(cmd, "niot", -1);
     int executorSize = Utils.intArg(cmd, "niotp", -1);
     if (nbThreads > 0 || executorSize > 0) {
-      NioParams nioParams = new NioParams();
+      com.rabbitmq.client.impl.nio.NioParams nioParams =
+          new com.rabbitmq.client.impl.nio.NioParams();
       int[] nbThreadsAndExecutorSize = getNioNbThreadsAndExecutorSize(nbThreads, executorSize);
       nioParams.setNbIoThreads(nbThreadsAndExecutorSize[0]);
       // FIXME we cannot limit the max size of the thread pool because of
@@ -847,7 +847,7 @@ protected static int[] getNioNbThreadsAndExecutorSize(
   }
 
   @SuppressWarnings("deprecation")
-  private static NioParams nioParams(ConnectionFactory cf) {
+  private static com.rabbitmq.client.impl.nio.NioParams nioParams(ConnectionFactory cf) {
     return cf.getNioParams();
   }
 

src/test/java/com/rabbitmq/perf/it/ConnectionRecoveryIT.java

@@ -29,10 +29,8 @@
 import com.rabbitmq.client.Channel;
 import com.rabbitmq.client.Connection;
 import com.rabbitmq.client.ConnectionFactory;
-import com.rabbitmq.client.impl.nio.NioParams;
 import com.rabbitmq.perf.MulticastParams;
 import com.rabbitmq.perf.MulticastSet;
-import com.rabbitmq.perf.NamedThreadFactory;
 import com.rabbitmq.perf.PerformanceMetricsAdapter;
 import com.rabbitmq.perf.metrics.PerformanceMetrics;
 import java.io.IOException;
@@ -99,18 +97,17 @@ public Duration interval() {
       };
 
   static Stream<Arguments> configurationArguments() {
-    return Stream.of(blockingIoAndNio(multicastParamsConfigurers()));
+    return Stream.of(blockingIoAndNetty(multicastParamsConfigurers()));
   }
 
-  static Arguments[] blockingIoAndNio(List<Consumer<MulticastParams>> multicastParamsConfigurers) {
+  static Arguments[] blockingIoAndNetty(
+      List<Consumer<MulticastParams>> multicastParamsConfigurers) {
     List<Arguments> arguments = new ArrayList<>();
     for (Consumer<MulticastParams> configurer : multicastParamsConfigurers) {
       arguments.add(
           Arguments.of(
               configurer, namedConsumer("blocking IO", (Consumer<ConnectionFactory>) cf -> {})));
-      arguments.add(
-          Arguments.of(
-              configurer, namedConsumer("NIO", (Consumer<ConnectionFactory>) cf -> cf.useNio())));
+      arguments.add(Arguments.of(configurer, namedConsumer("Netty", ConnectionFactory::useNetty)));
     }
 
     return arguments.toArray(new Arguments[0]);
@@ -146,7 +143,7 @@ static Stream<Arguments> configurationArgumentsForSeveralUris() {
             namedConsumer("one server-named queue", empty()),
             namedConsumer("several queues", severalQueues()),
             namedConsumer("queue sequence", queueSequence()))
-        .map(configurer -> Arguments.of(configurer));
+        .map(Arguments::of);
   }
 
   static Consumer<MulticastParams> empty() {
@@ -299,23 +296,11 @@ public void shouldRecoverWhenConnectionsAreKilledAndUsingPublishingInterval(
   }
 
   @Test
-  public void shouldRecoverWithNio(TestInfo info) throws Exception {
+  public void shouldRecoverWithNetty(TestInfo info) throws Exception {
     params.setQueueNames(Arrays.asList("one", "two", "three"));
     params.setProducerCount(10);
     params.setConsumerCount(10);
-    cf.useNio();
-    cf.setNioParams(
-        new NioParams()
-            .setNbIoThreads(10)
-            // see PerfTest#configureNioIfRequested
-            .setNioExecutor(
-                new ThreadPoolExecutor(
-                    10,
-                    params.getProducerCount() + params.getConsumerCount() + 5,
-                    30L,
-                    TimeUnit.SECONDS,
-                    new SynchronousQueue<>(),
-                    new NamedThreadFactory("perf-test-nio-"))));
+    cf.useNetty();
     int producerConsumerCount = params.getProducerCount();
     MulticastSet set =
         new MulticastSet(performanceMetrics, cf, params, "", URIS, latchCompletionHandler(1, info));