Change msg service: Acknowledge Response iff Backend Succeeds

- Before: Upon receipt of a message via the msg/ endpoint, MOA ZS
  would immediately acknowledge the receipt without verifying that the
  message was successfully processed by the backend.
- Now: MOA ZS receives a message via the msg/ endpoint, forwards it to
  the sinks, and acknowledges the receipt if and only if the
  processing succeeded.

1 files changed, 17 insertions, 8 deletions

diff --git a/src/main/java/at/gv/egiz/moazs/backend/MsgResponseSinkHub.java b/src/main/java/at/gv/egiz/moazs/backend/MsgResponseSinkHub.java
index 3927055..64b559f 100644
--- a/src/main/java/at/gv/egiz/moazs/backend/MsgResponseSinkHub.java
+++ b/src/main/java/at/gv/egiz/moazs/backend/MsgResponseSinkHub.java
@@ -22,6 +22,7 @@
 package at.gv.egiz.moazs.backend;
 
 import at.gv.egiz.moazs.scheme.MsgResponse;
+import at.gv.egiz.moazs.scheme.RequestStatusResponse;
 import at.gv.zustellung.app2mzs.xsd.MsgResponseSinksType;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -29,6 +30,12 @@ import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.core.task.TaskExecutor;
 import org.springframework.stereotype.Component;
 
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionStage;
+
+import static java.util.concurrent.CompletableFuture.allOf;
 import static java.util.concurrent.CompletableFuture.supplyAsync;
 
 /**
@@ -43,37 +50,39 @@ public class MsgResponseSinkHub {
     private final SaveResponseToFileSink saveResponseSink;
     private final LogResponseSink logResponseSink;
     private final ForwardResponseToServiceSink forwardResponseSink;
-    private final TaskExecutor taskExecutor;
 
     @Autowired
     public MsgResponseSinkHub(SaveResponseToFileSink saveResponseSink, LogResponseSink logResponseSink,
-                              ForwardResponseToServiceSink forwardResponseSink, TaskExecutor taskExecutor) {
+                              ForwardResponseToServiceSink forwardResponseSink) {
         this.saveResponseSink = saveResponseSink;
         this.logResponseSink = logResponseSink;
         this.forwardResponseSink = forwardResponseSink;
-        this.taskExecutor = taskExecutor;
     }
 
     /**
      * Apply all sinks that are configured in {@code sinkParams} to {@code msgResponse}.
      */
-    public void applySinks(MsgResponse msgResponse, MsgResponseSinksType sinkParams) {
+    public CompletableFuture<Void> applySinks(MsgResponse msgResponse, MsgResponseSinksType sinkParams) {
 
         log.info("Apply Sinks to msg:{} with appDeliveryID={}.",
                 msgResponse.getRootElementLocalPart(), msgResponse.getAppDeliveryID());
 
+        List<CompletableFuture<Void>> futures = new ArrayList();
+
         if (sinkParams.getSaveResponseToFile().isActive()) {
-            supplyAsync(() -> saveResponseSink.save(msgResponse, sinkParams.getSaveResponseToFile().getPath()), taskExecutor);
+            futures.add(saveResponseSink.save(msgResponse, sinkParams.getSaveResponseToFile().getPath()));
         }
 
         if (sinkParams.isLogResponse()) {
-            supplyAsync(() -> logResponseSink.log(msgResponse), taskExecutor);
+            futures.add(logResponseSink.log(msgResponse));
         }
 
         if (sinkParams.getForwardResponseToService().isActive()) {
-            supplyAsync(() -> forwardResponseSink.send(
-                    msgResponse, sinkParams.getForwardResponseToService().getAppClient()), taskExecutor);
+            futures.add(forwardResponseSink.send(
+                    msgResponse, sinkParams.getForwardResponseToService().getAppClient()));
         }
 
+        return allOf(futures.toArray(new CompletableFuture[futures.size()]));
+
     }
 }