2 files changed, 26 insertions, 19 deletions

diff --git a/src/main/java/at/gv/egiz/moazs/backend/MsgResponseBackend.java b/src/main/java/at/gv/egiz/moazs/backend/MsgResponseBackend.java
index 81398da..45a4000 100644
--- a/src/main/java/at/gv/egiz/moazs/backend/MsgResponseBackend.java
+++ b/src/main/java/at/gv/egiz/moazs/backend/MsgResponseBackend.java
@@ -35,6 +35,7 @@ import org.springframework.core.task.TaskExecutor;
 import org.springframework.stereotype.Component;
 
 import java.util.Map;
+import java.util.concurrent.CompletableFuture;
 import java.util.function.Consumer;
 import java.util.function.Supplier;
 
@@ -49,7 +50,7 @@ import static java.util.concurrent.CompletableFuture.supplyAsync;
  *
  */
 @Component
-public class MsgResponseBackend implements Consumer<String> {
+public class MsgResponseBackend {
 
     private static final Logger log = LoggerFactory.getLogger(MsgService.class);
 
@@ -92,15 +93,10 @@ public class MsgResponseBackend implements Consumer<String> {
      *
      * @param responseID refers to MsgResponse Object.
      */
-    @Override
-    public void accept(String responseID) {
+    public CompletableFuture<Void> accept(String responseID) {
         log.info("Backend accepts MsgResponse with responseID={}.", responseID);
-        supplyAsync(() -> verifySignature(responseID), taskExecutor)
-                .thenAcceptAsync(msgResponse -> applySinks(msgResponse), taskExecutor)
-                .exceptionally(ex -> {
-                    log.error(ex.getMessage(), ex);
-                    return null;
-                });
+        return supplyAsync(() -> verifySignature(responseID), taskExecutor)
+                .thenCompose(msgResponse -> applySinks(msgResponse));
     }
 
     private MsgResponse verifySignature(String responseID) {
@@ -117,16 +113,18 @@ public class MsgResponseBackend implements Consumer<String> {
 
             signatureVerifier.accept(binaryResponse);
             return response;
+
         } catch (MoaZSException ex) {
             log.error(ex.getMessage(), ex);
             var text = format(MOASP_SIGNATURE_INVALID_ERROR_MSG, response.getAppDeliveryID());
             var code = ERROR_MOASP_SIGNATURE_INVALID;
             return response.generateError(text, code);
+
         }
 
     }
 
-    private void applySinks(MsgResponse msgResponse) {
+    private CompletableFuture<Void> applySinks(MsgResponse msgResponse) {
         var appDeliveryID = msgResponse.getAppDeliveryID();
 
         var config = repository
@@ -135,7 +133,7 @@ public class MsgResponseBackend implements Consumer<String> {
                 .getConfig();
 
         var sinkParams = config.getMsgResponseSinks();
-        hub.applySinks(msgResponse, sinkParams);
+        return hub.applySinks(msgResponse, sinkParams);
     }
 
     private Supplier<DeliveryRequestType> supplyRequestWithDefaultConfig(String appDeliveryID) {
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()]));
+
     }
 }