RabbitMQ-OG cleanup message type, make ack/nack idempotent

Author: ozangunalp

smallrye-reactive-messaging-rabbitmq-og/src/main/java/io/smallrye/reactive/messaging/rabbitmq/og/IncomingRabbitMQMessage.java

@@ -1,12 +1,12 @@
 package io.smallrye.reactive.messaging.rabbitmq.og;
 
 import java.nio.charset.StandardCharsets;
+import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.CompletionStage;
+import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.function.BiFunction;
 import java.util.function.Function;
-import java.util.function.Supplier;
 
-import org.eclipse.microprofile.reactive.messaging.Message;
 import org.eclipse.microprofile.reactive.messaging.Metadata;
 
 import com.rabbitmq.client.AMQP;
@@ -27,6 +27,8 @@
  */
 public class IncomingRabbitMQMessage<T> implements ContextAwareMessage<T>, MetadataInjectableMessage<T> {
 
+    private final AtomicBoolean acknowledged = new AtomicBoolean(false);
+
     private final T payload;
     private Metadata metadata;
     private final RabbitMQAckHandler ackHandler;
@@ -146,73 +148,45 @@ public java.util.Map<String, Object> getHeaders() {
     }
 
     @Override
-    public Supplier<CompletionStage<Void>> getAck() {
-        return () -> ackHandler.handle(this);
+    public Function<Metadata, CompletionStage<Void>> getAckWithMetadata() {
+        return this::ack;
     }
 
     @Override
-    public Function<Throwable, CompletionStage<Void>> getNack() {
-        return (failure) -> nackHandler.handle(this, null, failure);
+    public BiFunction<Throwable, Metadata, CompletionStage<Void>> getNackWithMetadata() {
+        return this::nack;
     }
 
     @Override
-    public BiFunction<Throwable, Metadata, CompletionStage<Void>> getNackWithMetadata() {
-        return this::nack;
+    public CompletionStage<Void> ack(Metadata metadata) {
+        if (acknowledged.compareAndSet(false, true)) {
+            return ackHandler.handle(this)
+                    .whenComplete((v, t) -> {
+                        if (t != null) {
+                            acknowledged.set(false);
+                        }
+                    });
+        }
+        return CompletableFuture.completedFuture(null);
     }
 
-    /**
-     * Nack with metadata support.
-     *
-     * @param reason the reason for the nack
-     * @param metadata additional nack metadata
-     * @return a completion stage
-     */
     public CompletionStage<Void> nack(Throwable reason, Metadata metadata) {
-        return nackHandler.handle(this, metadata, reason);
+        if (acknowledged.compareAndSet(false, true)) {
+            return nackHandler.handle(this, metadata, reason)
+                    .whenComplete((v, t) -> {
+                        if (t != null) {
+                            acknowledged.set(false);
+                        }
+                    });
+        }
+        return CompletableFuture.completedFuture(null);
     }
 
     @Override
     public synchronized void injectMetadata(Object metadataObject) {
         this.metadata = this.metadata.with(metadataObject);
     }
 
-    /**
-     * Create a new message with a different payload but same metadata and ack/nack handlers.
-     * This is useful for transforming messages in a processing chain.
-     *
-     * @param <P> the new payload type
-     * @param newPayload the new payload
-     * @return a new message with the new payload
-     */
-    public <P> Message<P> withPayload(P newPayload) {
-        return new Message<P>() {
-            @Override
-            public P getPayload() {
-                return newPayload;
-            }
-
-            @Override
-            public Metadata getMetadata() {
-                return IncomingRabbitMQMessage.this.metadata;
-            }
-
-            @Override
-            public Supplier<CompletionStage<Void>> getAck() {
-                return IncomingRabbitMQMessage.this.getAck();
-            }
-
-            @Override
-            public Function<Throwable, CompletionStage<Void>> getNack() {
-                return IncomingRabbitMQMessage.this.getNack();
-            }
-
-            @Override
-            public BiFunction<Throwable, Metadata, CompletionStage<Void>> getNackWithMetadata() {
-                return IncomingRabbitMQMessage.this.getNackWithMetadata();
-            }
-        };
-    }
-
     /**
      * Default converter from byte array to String using UTF-8.
      */