Emit publish nack when losing connection

Only for messages that are not to be retried.

Fixes #155

(cherry picked from commit 35d3e9d)

Author: acogoluegnes

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

@@ -19,6 +19,7 @@
 import com.rabbitmq.client.MissedHeartbeatException;
 import com.rabbitmq.client.ShutdownSignalException;
 
+import com.rabbitmq.client.impl.recovery.AutorecoveringConnection;
 import java.time.Duration;
 import java.util.Map;
 import java.util.concurrent.Callable;
@@ -38,7 +39,7 @@ public static class ConnectionRecoveryTriggeringPredicate implements Predicate<T
         public boolean test(Throwable throwable) {
             if (throwable instanceof ShutdownSignalException) {
                 ShutdownSignalException sse = (ShutdownSignalException) throwable;
-                return !sse.isInitiatedByApplication() || (sse.getCause() instanceof MissedHeartbeatException);
+                return AutorecoveringConnection.DEFAULT_CONNECTION_RECOVERY_TRIGGERING_CONDITION.test(sse);
             }
             return false;
         }

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

@@ -34,6 +34,8 @@ public class OutboundMessage {
 
     private final byte[] body;
 
+    private volatile boolean published = false;
+
     /**
      * Constructs a new message which is described by the body, the target exchange and the routing key which
      * can be used for smart routing after the message is published to the exchange.
@@ -109,4 +111,12 @@ public String toString() {
                 ", body=" + Arrays.toString(body) +
                 '}';
     }
+
+    void published() {
+        this.published = true;
+    }
+
+    boolean isPublished() {
+        return published;
+    }
 }

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

@@ -18,6 +18,7 @@
 
 import com.rabbitmq.client.*;
 import com.rabbitmq.client.impl.AMQImpl;
+import java.util.Map.Entry;
 import org.reactivestreams.Publisher;
 import org.reactivestreams.Subscriber;
 import org.reactivestreams.Subscription;
@@ -965,8 +966,22 @@ private void handleAckNack(long deliveryTag, boolean multiple, boolean ack) {
                 };
                 channel.addConfirmListener(confirmListener);
 
-
                 this.shutdownListener = sse -> {
+                    // nack outstanding messages to warn downstream
+                    Iterator<Entry<Long, OMSG>> iterator = this.unconfirmed.entrySet()
+                        .iterator();
+                    while (iterator.hasNext()) {
+                        OMSG message = iterator.next().getValue();
+                        if (!message.isPublished()) {
+                            // we deal only with messages that won't be retried here
+                            try {
+                                subscriber.onNext(new OutboundMessageResult<>(message, false, false));
+                                iterator.remove();
+                            } catch (Exception e) {
+                                LOGGER.info("Error while nacking messages after channel failure");
+                            }
+                        }
+                    }
                     // the server is closing the channel because of some error (e.g. exchange does not exist).
                     // sending a signal downstream
                     if (!sse.isHardError() && !sse.isInitiatedByApplication()) {
@@ -986,7 +1001,6 @@ public void onNext(OMSG message) {
             if (checkComplete(message)) {
                 return;
             }
-
             long nextPublishSeqNo = channel.getNextPublishSeqNo();
             try {
                 unconfirmed.putIfAbsent(nextPublishSeqNo, message);
@@ -997,6 +1011,7 @@ public void onNext(OMSG message) {
                         this.propertiesProcessor.apply(message.getProperties(), nextPublishSeqNo),
                         message.getBody()
                 );
+                message.published();
             } catch (Exception e) {
                 unconfirmed.remove(nextPublishSeqNo);
                 try {

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

@@ -823,6 +823,56 @@ void publishConfirmsMaxInFlight() throws InterruptedException {
         assertThat(maxInflight.get()).isLessThanOrEqualTo(maxConcurrency);
     }
 
+    @Test
+    public void publishConfirmsEmitNackForUnconfirmedMessagesOnConnectionFailure() 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);
+
+        CountDownLatch publishLatch = new CountDownLatch(1);
+            doAnswer(answer -> {
+                publishLatch.countDown();
+                try {
+                    Thread.sleep(1000L);
+                } catch (Exception e) {
+                }
+                return null;
+            })
+            .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)
+            .subscribe(
+                outboundMessageResult -> {
+                    if (!outboundMessageResult.isAck()) {
+                        nackLatch.countDown();
+                    }
+                },
+                error -> {
+                });
+
+        assertThat(publishLatch.await(5, TimeUnit.SECONDS)).isTrue();
+
+        ArgumentCaptor<ShutdownListener> shutdownListenerArgumentCaptor = ArgumentCaptor.forClass(ShutdownListener.class);
+        verify(mockChannel).addShutdownListener(shutdownListenerArgumentCaptor.capture());
+
+        ShutdownListener shutdownListener = shutdownListenerArgumentCaptor.getValue();
+        shutdownListener.shutdownCompleted(new ShutdownSignalException(true, false, null, null));
+
+        assertThat(nackLatch.await(5, TimeUnit.SECONDS)).isTrue();
+    }
+
     @Test
     public void publishConfirmsErrorWhilePublishing() throws Exception {
         ConnectionFactory mockConnectionFactory = mock(ConnectionFactory.class);