Emit publish nack when retry times out

Fixes #156

(cherry picked from commit a923eed)

Conflicts:
	src/main/java/reactor/rabbitmq/ExceptionHandlers.java
	src/main/java/reactor/rabbitmq/OutboundMessage.java
	src/main/java/reactor/rabbitmq/Sender.java
	src/test/java/reactor/rabbitmq/ExceptionHandlersTests.java
	src/test/java/reactor/rabbitmq/RabbitFluxTests.java

Author: acogoluegnes

src/main/java/reactor/rabbitmq/ExceptionHandlers.java

@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2018-2019 Pivotal Software Inc, All Rights Reserved.
+ * Copyright (c) 2018-2021 VMware, Inc. or its affiliates.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -16,7 +16,6 @@
 
 package reactor.rabbitmq;
 
-import com.rabbitmq.client.MissedHeartbeatException;
 import com.rabbitmq.client.ShutdownSignalException;
 
 import com.rabbitmq.client.impl.recovery.AutorecoveringConnection;
@@ -72,7 +71,14 @@ public static class SimpleRetryTemplate {
 
         private final Predicate<Throwable> predicate;
 
+        private boolean failOnTimeout;
+
         public SimpleRetryTemplate(Duration timeout, Duration waitingTime, Predicate<Throwable> predicate) {
+            this(timeout, waitingTime, predicate, true);
+        }
+
+        public SimpleRetryTemplate(Duration timeout, Duration waitingTime, Predicate<Throwable> predicate,
+                boolean failOnTimeout) {
             if (timeout == null || timeout.isNegative() || timeout.isZero()) {
                 throw new IllegalArgumentException("Timeout must be greater than 0");
             }
@@ -88,11 +94,13 @@ public SimpleRetryTemplate(Duration timeout, Duration waitingTime, Predicate<Thr
             this.timeout = timeout.toMillis();
             this.waitingTime = waitingTime.toMillis();
             this.predicate = predicate;
+            this.failOnTimeout = failOnTimeout;
         }
 
         public void retry(Callable<Void> operation, Exception e) {
             if (predicate.test(e)) {
                 int elapsedTime = 0;
+                boolean callSucceeded = false;
                 while (elapsedTime < timeout) {
                     try {
                         Thread.sleep(waitingTime);
@@ -102,13 +110,17 @@ public void retry(Callable<Void> operation, Exception e) {
                     elapsedTime += waitingTime;
                     try {
                         operation.call();
+                        callSucceeded = true;
                         break;
                     } catch (Throwable sendingException) {
                         if (!predicate.test(sendingException)) {
                             throw new RabbitFluxException("Not retryable exception thrown during retry", sendingException);
                         }
                     }
                 }
+                if (!callSucceeded && failOnTimeout) {
+                    throw new RabbitFluxRetryTimeoutException("Retry timed out after " + this.timeout + " ms", e);
+                }
             } else {
                 throw new RabbitFluxException("Not retryable exception, cannot retry", e);
             }
@@ -122,7 +134,7 @@ public static class RetryAcknowledgmentExceptionHandler implements BiConsumer<Re
         public RetryAcknowledgmentExceptionHandler(Duration timeout, Duration waitingTime,
             Predicate<Throwable> predicate) {
             this.retryTemplate = new SimpleRetryTemplate(
-                timeout, waitingTime, predicate
+                timeout, waitingTime, predicate, false
             );
         }
 
@@ -140,8 +152,13 @@ public static class RetrySendingExceptionHandler implements BiConsumer<Sender.Se
         private final SimpleRetryTemplate retryTemplate;
 
         public RetrySendingExceptionHandler(Duration timeout, Duration waitingTime, Predicate<Throwable> predicate) {
+            this(timeout, waitingTime, predicate, true);
+        }
+
+        public RetrySendingExceptionHandler(Duration timeout, Duration waitingTime, Predicate<Throwable> predicate,
+                boolean failOnTimeout) {
             this.retryTemplate = new SimpleRetryTemplate(
-                timeout, waitingTime, predicate
+                timeout, waitingTime, predicate, failOnTimeout
             );
         }
 

src/main/java/reactor/rabbitmq/OutboundMessage.java

@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2017 Pivotal Software Inc, All Rights Reserved.
+ * Copyright (c) 2017-2021 VMware, Inc. or its affiliates.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.

src/main/java/reactor/rabbitmq/RabbitFluxRetryTimeoutException.java

@@ -0,0 +1,28 @@
+/*
+ * Copyright (c) 2021 VMware, Inc. or its affiliates.
+ *
+ * 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
+ *
+ *       https://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;
+
+/**
+ *
+ */
+public class RabbitFluxRetryTimeoutException extends RabbitFluxException {
+
+  public RabbitFluxRetryTimeoutException(String message, Throwable cause) {
+    super(message, cause);
+  }
+
+}

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

@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2017-2019 Pivotal Software Inc, All Rights Reserved.
+ * Copyright (c) 2017-2021 VMware, Inc. or its affiliates.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -1016,6 +1016,8 @@ public void onNext(OMSG message) {
                 unconfirmed.remove(nextPublishSeqNo);
                 try {
                     this.exceptionHandler.accept(new ConfirmSendContext<>(channel, message, this), e);
+                } catch (RabbitFluxRetryTimeoutException timeoutException) {
+                    subscriber.onNext(new OutboundMessageResult<>(message, false, false));
                 } catch (Exception innerException) {
                     handleError(innerException, new OutboundMessageResult<>(message, false, false));
                 }

src/test/java/reactor/rabbitmq/ExceptionHandlersTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2018 Pivotal Software Inc, All Rights Reserved.
+ * Copyright (c) 2018-2021 VMware, Inc. or its affiliates.
  *
  * 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,7 @@
 import com.rabbitmq.client.AlreadyClosedException;
 import com.rabbitmq.client.AuthenticationFailureException;
 import com.rabbitmq.client.ShutdownSignalException;
+import org.assertj.core.api.Assertions;
 import org.junit.jupiter.api.Test;
 
 import java.io.IOException;
@@ -30,6 +31,7 @@
 
 import static java.time.Duration.ofMillis;
 import static java.util.Collections.singletonMap;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertFalse;
 import static org.junit.jupiter.api.Assertions.assertThrows;
@@ -83,13 +85,27 @@ public void connectionRecoveryTriggering() {
     @Test
     void retryTimeoutIsReached() {
         exceptionHandler = new ExceptionHandlers.RetrySendingExceptionHandler(
-            ofMillis(100), ofMillis(10), new ExceptionHandlers.ExceptionPredicate(singletonMap(Exception.class, true))
+            ofMillis(100), ofMillis(10),
+            new ExceptionHandlers.ExceptionPredicate(singletonMap(Exception.class, true)), false
         );
         exceptionHandler.accept(sendContext(() -> {
             throw new Exception();
         }), new Exception());
     }
 
+    @Test
+    void shouldThrowTimeoutExceptionWhenConfigured() {
+        exceptionHandler = new ExceptionHandlers.RetrySendingExceptionHandler(
+            ofMillis(100), ofMillis(10),
+            new ExceptionHandlers.ExceptionPredicate(singletonMap(Exception.class, true)), true
+        );
+        assertThatThrownBy(() -> {
+            exceptionHandler.accept(sendContext(() -> {
+                throw new Exception();
+            }), new Exception());
+        }).isInstanceOf(RabbitFluxRetryTimeoutException.class);
+    }
+
     @Test
     void retrySucceeds() {
         exceptionHandler = new ExceptionHandlers.RetrySendingExceptionHandler(

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

@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2017-2020 Pivotal Software Inc, All Rights Reserved.
+ * Copyright (c) 2017-2021 VMware, Inc. or its affiliates.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -873,6 +873,46 @@ public void publishConfirmsEmitNackForUnconfirmedMessagesOnConnectionFailure() t
         assertThat(nackLatch.await(5, TimeUnit.SECONDS)).isTrue();
     }
 
+    @Test
+    public void publishConfirmsEmitNackOnRetryTimeout() throws Exception {
+        ConnectionFactory mockConnectionFactory = mock(ConnectionFactory.class);
+        Connection mockConnection = mock(Connection.class);
+        Channel mockChannel = mock(Channel.class);
+        when(mockConnectionFactory.newConnection()).thenReturn(mockConnection);
+        when(mockConnection.createChannel()).thenReturn(mockChannel);
+        when(mockConnection.isOpen()).thenReturn(true);
+        when(mockChannel.getConnection()).thenReturn(mockConnection);
+
+        AtomicLong publishSequence = new AtomicLong();
+        when(mockChannel.getNextPublishSeqNo()).thenAnswer(invocation -> publishSequence.incrementAndGet());
+        when(mockChannel.isOpen()).thenReturn(true);
+
+        doAnswer(answer -> {
+            throw new RuntimeException();
+        })
+            .when(mockChannel).basicPublish(anyString(), anyString(), eq(false), nullable(AMQP.BasicProperties.class), any(byte[].class));
+
+        Flux<OutboundMessage> msgFlux = Flux.just(new OutboundMessage("", queue, "".getBytes()));
+        CountDownLatch nackLatch = new CountDownLatch(1);
+        sender = createSender(new SenderOptions()
+            .connectionFactory(mockConnectionFactory));
+        sender
+            .sendWithPublishConfirms(msgFlux,
+                new SendOptions().exceptionHandler((ctx, ex) -> {
+                    throw new RabbitFluxRetryTimeoutException(null, null);
+                }))
+            .subscribe(
+                outboundMessageResult -> {
+                    if (!outboundMessageResult.isAck()) {
+                        nackLatch.countDown();
+                    }
+                },
+                error -> {
+                });
+
+        assertThat(nackLatch.await(5, TimeUnit.SECONDS)).isTrue();
+    }
+
     @Test
     public void publishConfirmsErrorWhilePublishing() throws Exception {
         ConnectionFactory mockConnectionFactory = mock(ConnectionFactory.class);