LazyChannelPool (added subscribeOn, signalType) and added channelPool property to SenderOptions

Author: flooq

src/main/java/reactor/rabbitmq/LazyChannelPool.java

@@ -33,13 +33,13 @@
 import static reactor.rabbitmq.ChannelCloseHandlers.SENDER_CHANNEL_CLOSE_HANDLER_INSTANCE;
 
 /**
- * This channel pool is lazy initialized. It might even not reach its maximum size {@link ChannelPoolOptions#getMaxCacheSize()} in low-traffic environments.
- * It always tries to obtain channel from the pool. However, in case of high-concurrency environment, number of channels might exceeds channel pool maximum size.
+ * This channel pool is lazy initialized. It might even not reach its maximum size {@link ChannelPoolOptions#getMaxCacheSize()} in low-concurrency environments.
+ * It always tries to obtain channel from the pool. However, in case of high-concurrency environments, number of channels might exceeds channel pool maximum size.
  *
  * Channels are added to the pool after their use {@link ChannelPool#getChannelCloseHandler()} and obtained from the pool when channel is requested {@link ChannelPool#getChannelMono()}.
  *
  * If pool is empty, new channel is created.
- * If channel is no longer needed and the channel pool is full (high-traffic), then channel is being closed.
+ * If channel is no longer needed and the channel pool is full, then channel is being closed.
  * If channel is no longer needed and the channel pool has not reached its capacity, then channel is added to the pool.
  *
  * It uses {@link BlockingQueue} internally in a non-blocking way.
@@ -69,7 +69,8 @@ public Mono<? extends Channel> getChannelMono() {
                 channel = createChannel(connection);
             }
             return channel;
-        }).subscribeOn(subscriptionScheduler);
+        })
+        .subscribeOn(subscriptionScheduler);
     }
 
     @Override
@@ -78,8 +79,7 @@ public BiConsumer<SignalType, Channel> getChannelCloseHandler() {
             if (!channel.isOpen()) {
                 return;
             }
-            // maybe also close channel if signalType == SignalType.ON_ERROR ?
-            boolean offer = channelsQueue.offer(channel);
+            boolean offer = signalType == SignalType.ON_COMPLETE && channelsQueue.offer(channel);
             if (!offer) {
                 SENDER_CHANNEL_CLOSE_HANDLER_INSTANCE.accept(signalType, channel);
             }

src/main/java/reactor/rabbitmq/SenderOptions.java

@@ -161,6 +161,20 @@ public SenderOptions channelCloseHandler(BiConsumer<SignalType, Channel> channel
         return this;
     }
 
+    /**
+     * Set the channel pool to use to send messages.
+     *
+     * @param channelPool
+     * @return this {@link SenderOptions} instance
+     * @since 1.1.0
+     */
+    public SenderOptions channelPool(ChannelPool channelPool) {
+        this.channelMono = channelPool.getChannelMono();
+        this.channelCloseHandler = channelPool.getChannelCloseHandler();
+        return this;
+    }
+
+
     public SenderOptions resourceManagementChannelMono(Mono<? extends Channel> resourceManagementChannelMono) {
         this.resourceManagementChannelMono = resourceManagementChannelMono;
         return this;

src/test/java/reactor/rabbitmq/RabbitFluxTests.java

@@ -19,7 +19,6 @@
 import com.rabbitmq.client.*;
 import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.BeforeEach;
-import org.junit.jupiter.api.Disabled;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.params.ParameterizedTest;
 import org.junit.jupiter.params.provider.Arguments;
@@ -590,40 +589,6 @@ public void senderWithCustomChannelCloseHandlerPriority(int nbMessages,
         verify(channelCloseHandlerInSendOptions, times(1)).accept(any(SignalType.class), any(Channel.class));
     }
 
-    @Test // temporary test, jvm warm-up ignored
-    @Disabled
-    public void senderWithChannelPoolPerformance() throws Exception {
-        int nbMessages = 1;
-        int count = 500;
-        int expectedSpeedupRatio = 3;
-
-        Flux<OutboundMessage> msgFlux = Flux.range(0, nbMessages).map(i -> new OutboundMessage("", queue, "".getBytes()));
-
-        ChannelPool channelPool = ChannelPoolFactory.createChannelPool(Mono.just(connection));
-        SendOptions sendOptions = new SendOptions().channelPool(channelPool);
-
-        sender = createSender();
-
-        Mono<Void> sendChannelPoolMono = Flux.range(0, count)
-                .flatMap(i -> sender.send(msgFlux, sendOptions))
-                .then();
-
-        Mono<Void> sendMono = Flux.range(0, count)
-                .flatMap(i -> sender.send(msgFlux))
-                .then();
-
-        Duration durationSendChannelPool = StepVerifier.create(sendChannelPoolMono).verifyComplete();
-        Duration durationSendChannelAlwaysCreated = StepVerifier.create(sendMono).verifyComplete();
-
-        int totalMessages = nbMessages * count * 2;
-        StepVerifier.create(consume(queue, totalMessages))
-                .expectNextCount(totalMessages)
-                .verifyComplete();
-
-        assertTrue(durationSendChannelPool.toMillis() * expectedSpeedupRatio < durationSendChannelAlwaysCreated.toMillis(),
-                String.format("Sender with channel pool is not %s times faster. Duration with channel pool is %s and without is %s", expectedSpeedupRatio, durationSendChannelPool, durationSendChannelAlwaysCreated));
-    }
-
     @Test
     public void publishConfirms() throws Exception {
         int nbMessages = 10;