Add test for channel mono choice when sending messages

References #41

Author: acogoluegnes

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

@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2018 Pivotal Software Inc, All Rights Reserved.
+ * Copyright (c) 2018-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.
@@ -29,11 +29,21 @@
 public class SendOptions {
 
     private BiConsumer<Sender.SendContext, Exception> exceptionHandler = new ExceptionHandlers.RetrySendingExceptionHandler(
-        Duration.ofSeconds(10), Duration.ofMillis(200), ExceptionHandlers.CONNECTION_RECOVERY_PREDICATE
+            Duration.ofSeconds(10), Duration.ofMillis(200), ExceptionHandlers.CONNECTION_RECOVERY_PREDICATE
     );
 
+    /**
+     * Channel mono to use to send messages.
+     *
+     * @since 1.1.0
+     */
     private Mono<? extends Channel> channelMono;
 
+    /**
+     * Channel closing logic.
+     *
+     * @since 1.1.0
+     */
     private BiConsumer<SignalType, Channel> channelCloseHandler;
 
     public BiConsumer<Sender.SendContext, Exception> getExceptionHandler() {
@@ -45,19 +55,45 @@ public SendOptions exceptionHandler(BiConsumer<Sender.SendContext, Exception> ex
         return this;
     }
 
+    /**
+     * Return the channel mono to use to send messages.
+     *
+     * @return
+     * @since 1.1.0
+     */
     public Mono<? extends Channel> getChannelMono() {
         return channelMono;
     }
 
+    /**
+     * Set the channel mono to use to send messages.
+     *
+     * @param channelMono
+     * @return this {@link SendOptions} instance
+     * @since 1.1.0
+     */
     public SendOptions channelMono(Mono<? extends Channel> channelMono) {
         this.channelMono = channelMono;
         return this;
     }
 
+    /**
+     * Return the channel closing logic.
+     *
+     * @return
+     * @since 1.1.0
+     */
     public BiConsumer<SignalType, Channel> getChannelCloseHandler() {
         return channelCloseHandler;
     }
 
+    /**
+     * Set the channel closing logic.
+     *
+     * @param channelCloseHandler
+     * @return this {@link SendOptions} instance
+     * @since 1.1.0
+     */
     public SendOptions channelCloseHandler(BiConsumer<SignalType, Channel> channelCloseHandler) {
         this.channelCloseHandler = channelCloseHandler;
         return this;

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

@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2017-2018 Pivotal Software Inc, All Rights Reserved.
+ * Copyright (c) 2017-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.
@@ -110,6 +110,7 @@ public Mono<Void> send(Publisher<OutboundMessage> messages) {
     }
 
     public Mono<Void> send(Publisher<OutboundMessage> messages, SendOptions options) {
+        options = options == null ? new 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
@@ -141,12 +142,14 @@ public Flux<OutboundMessageResult> sendWithPublishConfirms(Publisher<OutboundMes
     }
 
     public Flux<OutboundMessageResult> sendWithPublishConfirms(Publisher<OutboundMessage> messages, SendOptions options) {
+        options = options == null ? new 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(options);
         final BiConsumer<SignalType, Channel> channelCloseHandler = getChannelCloseHandler(options);
 
+        SendOptions sendOptions = options;
         return currentChannelMono.map(channel -> {
                 try {
                     channel.confirmSelect();
@@ -155,10 +158,11 @@ public Flux<OutboundMessageResult> sendWithPublishConfirms(Publisher<OutboundMes
                 }
                 return channel;
             })
-            .flatMapMany(channel -> new PublishConfirmOperator(messages, channel, channelCloseHandler, options));
+            .flatMapMany(channel -> new PublishConfirmOperator(messages, channel, channelCloseHandler, sendOptions));
     }
 
-    private Mono<? extends Channel> getChannelMono(SendOptions options) {
+    // package-protected for testing
+    Mono<? extends Channel> getChannelMono(SendOptions options) {
         return Stream.of(options.getChannelMono(), channelMono)
                 .filter(Objects::nonNull)
                 .findFirst().orElse(connectionMono.map(CHANNEL_CREATION_FUNCTION));
@@ -511,12 +515,12 @@ public void onSubscribe(Subscription subscription) {
             channel.addConfirmListener(new ConfirmListener() {
 
                 @Override
-                public void handleAck(long deliveryTag, boolean multiple) throws IOException {
+                public void handleAck(long deliveryTag, boolean multiple) {
                     handleAckNack(deliveryTag, multiple, true);
                 }
 
                 @Override
-                public void handleNack(long deliveryTag, boolean multiple) throws IOException {
+                public void handleNack(long deliveryTag, boolean multiple) {
                     handleAckNack(deliveryTag, multiple, false);
                 }
 

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

@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2017-2018 Pivotal Software Inc, All Rights Reserved.
+ * Copyright (c) 2017-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.
@@ -550,7 +550,7 @@ public void senderWithCustomChannelCloseHandler() throws Exception {
     }
 
     @Test
-    public void senderWithCustomChannelCloseHandlerPriority() throws Exception {
+    public void senderWithCustomChannelCloseHandlerPriority() {
         int nbMessages = 10;
         Flux<OutboundMessage> msgFlux = Flux.range(0, nbMessages).map(i -> new OutboundMessage("", queue, "".getBytes()));
 

src/test/java/reactor/rabbitmq/SenderTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2018 Pivotal Software Inc, All Rights Reserved.
+ * Copyright (c) 2018-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.
@@ -23,10 +23,12 @@
 import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
+import org.mockito.Mockito;
+import reactor.core.publisher.Mono;
 
 import java.util.UUID;
 
-import static org.junit.jupiter.api.Assertions.fail;
+import static org.junit.jupiter.api.Assertions.*;
 import static reactor.rabbitmq.RabbitFlux.createSender;
 
 /**
@@ -64,7 +66,7 @@ public void tearDown() throws Exception {
     }
 
     @Test
-    void cannotReuseChannelOnError() {
+    void canReuseChannelOnError() {
         sender = createSender();
         try {
             sender.declare(QueueSpecification.queue(queue).autoDelete(true)).block();
@@ -74,4 +76,17 @@ void cannotReuseChannelOnError() {
         }
         sender.declare(QueueSpecification.queue()).block();
     }
+
+    @Test
+    void channelMonoPriority() {
+        Mono<Channel> senderChannelMono = Mono.just(Mockito.mock(Channel.class));
+        Mono<Channel> sendChannelMono = Mono.just(Mockito.mock(Channel.class));
+        sender = createSender();
+        assertNotNull(sender.getChannelMono(new SendOptions()));
+        assertSame(sendChannelMono, sender.getChannelMono(new SendOptions().channelMono(sendChannelMono)));
+
+        sender = createSender(new SenderOptions().channelMono(senderChannelMono));
+        assertSame(senderChannelMono, sender.getChannelMono(new SendOptions()));
+        assertSame(sendChannelMono, sender.getChannelMono(new SendOptions().channelMono(sendChannelMono)));
+    }
 }