ChannelPool - some simple fixes

Author: flooq

src/main/java/reactor/rabbitmq/ChannelPoolOptions.java

@@ -17,21 +17,20 @@
 package reactor.rabbitmq;
 
 import reactor.core.scheduler.Scheduler;
-import reactor.core.scheduler.Schedulers;
 
 public class ChannelPoolOptions {
 
-    private int maxSize = 5;
+    private Integer maxCacheSize;
 
-    private Scheduler subscriptionScheduler = Schedulers.newElastic("sender-channel-pool");
+    private Scheduler subscriptionScheduler;
 
-    public ChannelPoolOptions maxSize(int maxSize) {
-        this.maxSize = maxSize;
+    public ChannelPoolOptions maxCacheSize(int maxCacheSize) {
+        this.maxCacheSize = maxCacheSize;
         return this;
     }
 
-    public int getMaxSize() {
-        return maxSize;
+    public Integer getMaxCacheSize() {
+        return maxCacheSize;
     }
 
     public ChannelPoolOptions subscriptionScheduler(Scheduler subscriptionScheduler) {

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

@@ -21,6 +21,7 @@
 import reactor.core.publisher.Mono;
 import reactor.core.publisher.SignalType;
 import reactor.core.scheduler.Scheduler;
+import reactor.core.scheduler.Schedulers;
 
 import java.io.IOException;
 import java.util.ArrayList;
@@ -32,8 +33,8 @@
 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#getMaxSize()} in low-traffic environments.
- * It always tries to obtain channel from the pool. However, in case of high-traffic 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-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.
  *
  * Channels are added to the pool after their use {@link ChannelPool#getChannelCloseHandler()} and obtained from the pool when channel is requested {@link ChannelPool#getChannelMono()}.
  *
@@ -46,14 +47,19 @@
  */
 class LazyChannelPool implements ChannelPool {
 
+    private static final int DEFAULT_CHANNEL_POOL_SIZE = 5;
+
     private final Mono<? extends Connection> connectionMono;
     private final BlockingQueue<Channel> channelsQueue;
     private final Scheduler subscriptionScheduler;
 
     LazyChannelPool(Mono<? extends Connection> connectionMono, ChannelPoolOptions channelPoolOptions) {
-        this.channelsQueue = new LinkedBlockingQueue<>(channelPoolOptions.getMaxSize());
+        int channelsQueueCapacity = channelPoolOptions.getMaxCacheSize() == null ?
+                DEFAULT_CHANNEL_POOL_SIZE : channelPoolOptions.getMaxCacheSize();
+        this.channelsQueue = new LinkedBlockingQueue<>(channelsQueueCapacity);
         this.connectionMono = connectionMono;
-        this.subscriptionScheduler = channelPoolOptions.getSubscriptionScheduler();
+        this.subscriptionScheduler = channelPoolOptions.getSubscriptionScheduler() == null ?
+                Schedulers.newElastic("sender-channel-pool") : channelPoolOptions.getSubscriptionScheduler();
     }
 
     public Mono<? extends Channel> getChannelMono() {

src/test/java/reactor/rabbitmq/LazyChannelPoolTests.java

@@ -7,6 +7,7 @@
 import org.junit.jupiter.api.Test;
 import reactor.core.publisher.Mono;
 import reactor.test.StepVerifier;
+import reactor.test.scheduler.VirtualTimeScheduler;
 
 import java.io.IOException;
 import java.time.Duration;
@@ -34,7 +35,9 @@ void setUp() throws IOException {
     @Test
     void testChannelPoolLazyInitialization() throws Exception {
         int maxChannelPoolSize = 2;
-        ChannelPoolOptions channelPoolOptions = new ChannelPoolOptions().maxSize(maxChannelPoolSize);
+        ChannelPoolOptions channelPoolOptions = new ChannelPoolOptions()
+                .maxCacheSize(maxChannelPoolSize)
+                .subscriptionScheduler(VirtualTimeScheduler.create());
         lazyChannelPool = new LazyChannelPool(Mono.just(connection), channelPoolOptions);
 
         StepVerifier.withVirtualTime(() ->
@@ -67,7 +70,9 @@ void testChannelPoolLazyInitialization() throws Exception {
     @Test
     void testChannelPoolExceedsMaxPoolSize() throws Exception {
         int maxChannelPoolSize = 2;
-        ChannelPoolOptions channelPoolOptions = new ChannelPoolOptions().maxSize(maxChannelPoolSize);
+        ChannelPoolOptions channelPoolOptions = new ChannelPoolOptions()
+                .maxCacheSize(maxChannelPoolSize)
+                .subscriptionScheduler(VirtualTimeScheduler.create());
         lazyChannelPool = new LazyChannelPool(Mono.just(connection), channelPoolOptions);
 
         StepVerifier.withVirtualTime(() ->
@@ -107,7 +112,9 @@ void testChannelPoolExceedsMaxPoolSize() throws Exception {
     @Test
     void testChannelPool() throws Exception {
         int maxChannelPoolSize = 1;
-        ChannelPoolOptions channelPoolOptions = new ChannelPoolOptions().maxSize(maxChannelPoolSize);
+        ChannelPoolOptions channelPoolOptions = new ChannelPoolOptions()
+                .maxCacheSize(maxChannelPoolSize)
+                .subscriptionScheduler(VirtualTimeScheduler.create());
         lazyChannelPool = new LazyChannelPool(Mono.just(connection), channelPoolOptions);
 
         StepVerifier.withVirtualTime(() ->
@@ -170,7 +177,7 @@ private void verifyBasicPublishOnce(Channel channel) throws Exception {
     }
 
     private void verifyBasicPublish(Channel channel, int times) throws Exception {
-        verify(channel, times(times)).basicPublish(anyString(), anyString(), any(AMQP.BasicProperties.class), any(byte[].class));
+        verify(channel, times(times)).basicPublish(anyString(), anyString(), nullable(AMQP.BasicProperties.class), any(byte[].class));
     }
 
     private Channel channel(int channelNumber) {