Add Javadoc for channel pooling

References #47

Author: acogoluegnes

src/jmh/java/reactor/rabbitmq/SenderBenchmark.java

@@ -1,3 +1,19 @@
+/*
+ * Copyright (c) 2019 Pivotal Software Inc, All Rights Reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *       http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
 package reactor.rabbitmq;
 
 import com.rabbitmq.client.Connection;

src/jmh/java/reactor/rabbitmq/SenderBenchmarkUtils.java

@@ -1,3 +1,19 @@
+/*
+ * Copyright (c) 2019 Pivotal Software Inc, All Rights Reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *       http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
 package reactor.rabbitmq;
 
 import com.rabbitmq.client.Channel;

src/jmh/java/reactor/rabbitmq/SenderWithLazyChannelPoolBenchmark.java

@@ -1,3 +1,19 @@
+/*
+ * Copyright (c) 2019 Pivotal Software Inc, All Rights Reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *       http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
 package reactor.rabbitmq;
 
 import com.rabbitmq.client.Connection;

src/main/java/reactor/rabbitmq/ChannelPool.java

@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2018-2019 Pivotal Software Inc, All Rights Reserved.
+ * Copyright (c) 2019 Pivotal Software Inc, All Rights Reserved.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -17,17 +17,44 @@
 package reactor.rabbitmq;
 
 import com.rabbitmq.client.Channel;
+import org.reactivestreams.Publisher;
 import reactor.core.publisher.Mono;
 import reactor.core.publisher.SignalType;
 
 import java.util.function.BiConsumer;
 
-public interface ChannelPool {
+/**
+ * Contract to obtain a {@link Channel} and close when sending messages.
+ * <p>
+ * Implementations would typically make it possible to re-use
+ * {@link Channel}s across different calls to <code>Sender#send*</code> methods.
+ *
+ * @see SendOptions
+ * @see Sender#send(Publisher)
+ * @see Sender#send(Publisher, SendOptions)
+ * @see Sender#sendWithPublishConfirms(Publisher)
+ * @see Sender#send(Publisher, SendOptions)
+ * @since 1.1.0
+ */
+public interface ChannelPool extends AutoCloseable {
 
+    /**
+     * The {@link Channel} to use for sending a flux of messages.
+     *
+     * @return
+     */
     Mono<? extends Channel> getChannelMono();
 
+    /**
+     * The closing logic when the {@link Channel} is disposed.
+     *
+     * @return
+     */
     BiConsumer<SignalType, Channel> getChannelCloseHandler();
 
+    /**
+     * Close the pool when it is no longer necessary.
+     */
     void close();
 
 }

src/main/java/reactor/rabbitmq/ChannelPoolFactory.java

@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2018-2019 Pivotal Software Inc, All Rights Reserved.
+ * Copyright (c) 2019 Pivotal Software Inc, All Rights Reserved.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -19,6 +19,14 @@
 import com.rabbitmq.client.Connection;
 import reactor.core.publisher.Mono;
 
+/**
+ * Factory to create default {@link ChannelPool} instances.
+ *
+ * @see ChannelPool
+ * @see LazyChannelPool
+ * @see ChannelPoolOptions
+ * @since 1.1.0
+ */
 public class ChannelPoolFactory {
 
     public static ChannelPool createChannelPool(Mono<? extends Connection> connectionMono) {

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

@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2018-2019 Pivotal Software Inc, All Rights Reserved.
+ * Copyright (c) 2019 Pivotal Software Inc, All Rights Reserved.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -18,12 +18,27 @@
 
 import reactor.core.scheduler.Scheduler;
 
+/**
+ * Options to when creating a {@link ChannelPool}.
+ *
+ * @see ChannelPoolFactory
+ * @see LazyChannelPool
+ * @since 1.1.0
+ */
 public class ChannelPoolOptions {
 
     private Integer maxCacheSize;
 
     private Scheduler subscriptionScheduler;
 
+    /**
+     * Set the maximum size of the pool.
+     * <p>
+     * Default is 5 channels.
+     *
+     * @param maxCacheSize
+     * @return this {@link ChannelPoolOptions} instance
+     */
     public ChannelPoolOptions maxCacheSize(int maxCacheSize) {
         this.maxCacheSize = maxCacheSize;
         return this;
@@ -33,6 +48,12 @@ public Integer getMaxCacheSize() {
         return maxCacheSize;
     }
 
+    /**
+     * Set the scheduler to use when opening {@link com.rabbitmq.client.Channel}s.
+     *
+     * @param subscriptionScheduler
+     * @return this {@link ChannelPoolOptions} instance
+     */
     public ChannelPoolOptions subscriptionScheduler(Scheduler subscriptionScheduler) {
         this.subscriptionScheduler = subscriptionScheduler;
         return this;

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

@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2018-2019 Pivotal Software Inc, All Rights Reserved.
+ * Copyright (c) 2019 Pivotal Software Inc, All Rights Reserved.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -33,17 +33,24 @@
 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-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, 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.
- *
+ * Default implementation of {@link ChannelPool}.
+ * <p>
+ * 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 a channel from the pool. However, in case of high-concurrency environments,
+ * number of channels might exceed channel pool maximum size.
+ * <p>
+ * Channels are added to the pool after their use {@link ChannelPool#getChannelCloseHandler()}
+ * and obtained from the pool when channel is requested {@link ChannelPool#getChannelMono()}.
+ * <p>
+ * If the pool is empty, a new channel is created.
+ * If a channel is no longer needed and the channel pool is full, then the channel is being closed.
+ * If a channel is no longer needed and the channel pool has not reached its
+ * capacity, then the channel is added to the pool.
+ * <p>
  * It uses {@link BlockingQueue} internally in a non-blocking way.
  *
+ * @since 1.1.0
  */
 class LazyChannelPool implements ChannelPool {
 
@@ -70,7 +77,7 @@ public Mono<? extends Channel> getChannelMono() {
             }
             return channel;
         })
-        .subscribeOn(subscriptionScheduler);
+                .subscribeOn(subscriptionScheduler);
     }
 
     @Override

src/main/java/reactor/rabbitmq/SendOptions.java

@@ -101,6 +101,8 @@ public SendOptions channelCloseHandler(BiConsumer<SignalType, Channel> channelCl
 
     /**
      * Set the channel pool to use to send messages.
+     * <p>
+     * It is developer's responsibility to close it if set.
      *
      * @param channelPool
      * @return this {@link SendOptions} instance

src/main/java/reactor/rabbitmq/Sender.java

@@ -156,8 +156,8 @@ public Flux<OutboundMessageResult> sendWithPublishConfirms(Publisher<OutboundMes
                 return channel;
             })
             .flatMapMany(channel -> Flux.from(new PublishConfirmOperator(messages, channel, sendOptions)).doFinally(signalType -> {
-                // channel close is no longer a responsibility of PublishConfirmOperator, not sure whether it is a correct approach
-                // added to avoid creating threads inside PublishConfirmOperator, which make ChannelPool useless
+                // channel closing is done here, to avoid creating threads inside PublishConfirmOperator,
+                // which would make ChannelPool useless
                 if (signalType == SignalType.ON_ERROR) {
                     channelCloseHandler.accept(signalType, channel);
                 } else {

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

@@ -1,3 +1,19 @@
+/*
+ * Copyright (c) 2019 Pivotal Software Inc, All Rights Reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *       http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
 package reactor.rabbitmq;
 
 import com.rabbitmq.client.AMQP;