Merge pull request #203 from quarkiverse/feature/failure-fix

Fixed acknowledge of published messages

.github/project.yml

@@ -1,4 +1,4 @@
 release:
-  current-version: "3.15.3"
+  current-version: "3.15.4"
   next-version: "3.16.0-SNAPSHOT"
 

docs/modules/ROOT/pages/includes/attributes.adoc

@@ -1,3 +1,3 @@
-:project-version: 3.15.3
+:project-version: 3.15.4
 
 :examples-dir: ./../examples/

runtime/src/main/java/io/quarkiverse/reactive/messaging/nats/jetstream/client/DefaultConnection.java

@@ -239,6 +239,8 @@ public <T> Uni<Message<T>> publish(Message<T> message, PublishConfiguration conf
             }
         })
                 .emitOn(context::runOnContext)
+                .onItem().transformToUni(this::acknowledge)
+                .onFailure().recoverWithUni(failure -> notAcknowledge(message, failure))
                 .onFailure().transform(failure -> new PublishException(failure.getMessage(), failure));
     }
 
@@ -397,6 +399,17 @@ private FetchConsumer fetchConsumer(final ConsumerContext consumerContext, final
         }
     }
 
+    private <T> Uni<Message<T>> acknowledge(final Message<T> message) {
+        return Uni.createFrom().completionStage(message.ack())
+                .onItem().transform(v -> message);
+    }
+
+    private <T> Uni<Message<T>> notAcknowledge(final Message<T> message, final Throwable throwable) {
+        return Uni.createFrom().completionStage(message.nack(throwable))
+                .onItem().invoke(() -> logger.warnf(throwable, "Message not acknowledged: %s", throwable.getMessage()))
+                .onItem().transformToUni(v -> Uni.createFrom().item(message));
+    }
+
     private <T> Uni<ConsumerContext> getConsumerContext(final FetchConsumerConfiguration<T> configuration) {
         return Uni.createFrom().item(Unchecked.supplier(() -> {
             try {