Merge pull request #45 from pmackowski/channel-pool-send-options

Added channelMono and channelCloseHandler properties to SendOptions

Author: acogoluegnes

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

@@ -16,6 +16,10 @@
 
 package reactor.rabbitmq;
 
+import com.rabbitmq.client.Channel;
+import reactor.core.publisher.Mono;
+import reactor.core.publisher.SignalType;
+
 import java.time.Duration;
 import java.util.function.BiConsumer;
 
@@ -28,6 +32,10 @@ public class SendOptions {
         Duration.ofSeconds(10), Duration.ofMillis(200), ExceptionHandlers.CONNECTION_RECOVERY_PREDICATE
     );
 
+    private Mono<? extends Channel> channelMono;
+
+    private BiConsumer<SignalType, Channel> channelCloseHandler;
+
     public BiConsumer<Sender.SendContext, Exception> getExceptionHandler() {
         return exceptionHandler;
     }
@@ -37,4 +45,21 @@ public SendOptions exceptionHandler(BiConsumer<Sender.SendContext, Exception> ex
         return this;
     }
 
+    public Mono<? extends Channel> getChannelMono() {
+        return channelMono;
+    }
+
+    public SendOptions channelMono(Mono<? extends Channel> channelMono) {
+        this.channelMono = channelMono;
+        return this;
+    }
+
+    public BiConsumer<SignalType, Channel> getChannelCloseHandler() {
+        return channelCloseHandler;
+    }
+
+    public SendOptions channelCloseHandler(BiConsumer<SignalType, Channel> channelCloseHandler) {
+        this.channelCloseHandler = channelCloseHandler;
+        return this;
+    }
 }

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

@@ -113,8 +113,9 @@ public Mono<Void> send(Publisher<OutboundMessage> messages, SendOptions options)
         // TODO using a pool of channels?
         // would be much more efficient if send is called very often
         // less useful if seldom called, only for long or infinite message flux
-        final Mono<? extends Channel> currentChannelMono = getChannelMono();
+        final Mono<? extends Channel> currentChannelMono = getChannelMono(options);
         final BiConsumer<SendContext, Exception> exceptionHandler = options.getExceptionHandler();
+        final BiConsumer<SignalType, Channel> channelCloseHandler = getChannelCloseHandler(options);
 
         return currentChannelMono.flatMapMany(channel ->
             Flux.from(messages)
@@ -143,7 +144,8 @@ public Flux<OutboundMessageResult> sendWithPublishConfirms(Publisher<OutboundMes
         // TODO using a pool of channels?
         // would be much more efficient if send is called very often
         // less useful if seldom called, only for long or infinite message flux
-        final Mono<? extends Channel> currentChannelMono = getChannelMono();
+        final Mono<? extends Channel> currentChannelMono = getChannelMono(options);
+        final BiConsumer<SignalType, Channel> channelCloseHandler = getChannelCloseHandler(options);
 
         return currentChannelMono.map(channel -> {
                 try {
@@ -156,8 +158,15 @@ public Flux<OutboundMessageResult> sendWithPublishConfirms(Publisher<OutboundMes
             .flatMapMany(channel -> new PublishConfirmOperator(messages, channel, channelCloseHandler, options));
     }
 
-    private Mono<? extends Channel> getChannelMono() {
-        return channelMono != null ? channelMono : connectionMono.map(CHANNEL_CREATION_FUNCTION);
+    private Mono<? extends Channel> getChannelMono(SendOptions options) {
+        return Stream.of(options.getChannelMono(), channelMono)
+                .filter(Objects::nonNull)
+                .findFirst().orElse(connectionMono.map(CHANNEL_CREATION_FUNCTION));
+    }
+
+    private BiConsumer<SignalType, Channel> getChannelCloseHandler(SendOptions options) {
+        return options.getChannelCloseHandler() != null ?
+                options.getChannelCloseHandler() : this.channelCloseHandler;
     }
 
     public RpcClient rpcClient(String exchange, String routingKey) {

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

@@ -549,6 +549,25 @@ public void senderWithCustomChannelCloseHandler() throws Exception {
                 .verifyComplete();
     }
 
+    @Test
+    public void senderWithCustomChannelCloseHandlerPriority() throws Exception {
+        int nbMessages = 10;
+        Flux<OutboundMessage> msgFlux = Flux.range(0, nbMessages).map(i -> new OutboundMessage("", queue, "".getBytes()));
+
+        SenderChannelCloseHandler channelCloseHandlerInSenderOptions = mock(SenderChannelCloseHandler.class);
+        SenderChannelCloseHandler channelCloseHandlerInSendOptions = mock(SenderChannelCloseHandler.class);
+
+        SenderOptions senderOptions = new SenderOptions().channelCloseHandler(channelCloseHandlerInSenderOptions);
+        sender = createSender(senderOptions);
+        SendOptions sendOptions = new SendOptions().channelCloseHandler(channelCloseHandlerInSendOptions);
+
+        StepVerifier.create(sender.send(msgFlux, sendOptions))
+                .verifyComplete();
+
+        verify(channelCloseHandlerInSenderOptions, never()).accept(any(SignalType.class), any(Channel.class));
+        verify(channelCloseHandlerInSendOptions, times(1)).accept(any(SignalType.class), any(Channel.class));
+    }
+
     @Test
     public void publishConfirms() throws Exception {
         int nbMessages = 10;