Merge pull request #199 from AvitalOfstein/feature/nackFunctionalityI…

…mplementation Support Nack functionality for AMQP implementation - avoid consumers …
Netflix · Apr 13, 2023 · 59e1845 · 59e1845
2 parents 5bcaac2 + e0d224b
commit 59e1845
Showing 1 changed file with 37 additions and 0 deletions.
diff --git a/...eue/amqp/src/main/java/com/netflix/conductor/contribs/queue/amqp/AMQPObservableQueue.java b/...eue/amqp/src/main/java/com/netflix/conductor/contribs/queue/amqp/AMQPObservableQueue.java
@@ -231,6 +231,43 @@ public void ackMsg(Message message) throws Exception {
         }
     }
 
+    @Override
+    public void nack(List<Message> messages) {
+        for (final Message message : messages) {
+            int retryIndex = 1;
+            while (true) {
+                try {
+                    LOGGER.info("NACK message with delivery tag {}", message.getReceipt());
+                    Channel chn =
+                            amqpConnection.getOrCreateChannel(
+                                    ConnectionType.SUBSCRIBER,
+                                    getSettings().getQueueOrExchangeName());
+                    chn.basicNack(Long.parseLong(message.getReceipt()), false, false);
+                    LOGGER.info("Nack'ed the message with delivery tag {}", message.getReceipt());
+                    break;
+                } catch (final Exception e) {
+                    AMQPRetryPattern retry = retrySettings;
+                    if (retry == null) {
+                        LOGGER.error(
+                                "Cannot NACK message with delivery tag {}",
+                                message.getReceipt(),
+                                e);
+                    }
+                    try {
+                        retry.continueOrPropogate(e, retryIndex);
+                    } catch (Exception ex) {
+                        LOGGER.error(
+                                "Retries completed. Cannot NACK message with delivery tag {}",
+                                message.getReceipt(),
+                                e);
+                        break;
+                    }
+                    retryIndex++;
+                }
+            }
+        }
+    }
+
     private static AMQP.BasicProperties buildBasicProperties(
             final Message message, final AMQPSettings settings) {
         return new AMQP.BasicProperties.Builder()