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.

2 files changed, 39 insertions, 11 deletions

diff --git a/src/main/java/at/gv/egiz/moazs/service/MsgService.java b/src/main/java/at/gv/egiz/moazs/service/MsgService.java
index 6b260e7..6fce6ae 100644
--- a/src/main/java/at/gv/egiz/moazs/service/MsgService.java
+++ b/src/main/java/at/gv/egiz/moazs/service/MsgService.java
@@ -20,6 +20,7 @@
  * that you distribute must include a readable copy of the "NOTICE" text file.
  *******************************************************************************/
 package at.gv.egiz.moazs.service;
+import at.gv.egiz.moazs.backend.MsgResponseBackend;
 import at.gv.egiz.moazs.repository.DeliveryRepository;
 import at.gv.egiz.moazs.scheme.MsgResponse;
 import at.gv.egiz.moazs.scheme.NameSpace;
@@ -27,15 +28,21 @@ import at.gv.egiz.moazs.scheme.NotificationResponse;
 import at.gv.egiz.moazs.scheme.RequestStatusResponse;
 import at.gv.zustellung.msg.xsd.*;
 import org.apache.cxf.annotations.SchemaValidation;
+import org.apache.cxf.binding.soap.SoapFault;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.core.task.TaskExecutor;
 import org.springframework.stereotype.Service;
 
-import java.util.function.Consumer;
+import javax.xml.namespace.QName;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
 
 import static at.gv.zustellung.msg.xsd.DeliveryNotificationACKType.deliveryNotificationACKTypeBuilder;
 import static at.gv.zustellung.msg.xsd.DeliveryRequestStatusACKType.deliveryRequestStatusACKTypeBuilder;
 import static at.gv.zustellung.msg.xsd.GetVersionResponse.getVersionResponseBuilder;
+import static java.lang.String.format;
 import static java.util.concurrent.CompletableFuture.runAsync;
 
 /**
@@ -47,12 +54,15 @@ import static java.util.concurrent.CompletableFuture.runAsync;
     @org.apache.cxf.feature.Features (features = "org.apache.cxf.ext.logging.LoggingFeature")
 public class MsgService implements Zuse2AppPort {
 
+    private static final Logger log = LoggerFactory.getLogger(MsgService.class);
+    private static final String BACKEND_ERROR_MSG = "Could not process %s.";
+
     private final DeliveryRepository repository;
-    private final Consumer<String> backend;
+    private final MsgResponseBackend backend;
     private final TaskExecutor taskExecutor;
 
     @Autowired
-    public MsgService(DeliveryRepository repository, Consumer<String> msgResponseBackend, TaskExecutor taskExecutor) {
+    public MsgService(DeliveryRepository repository, MsgResponseBackend msgResponseBackend, TaskExecutor taskExecutor) {
         this.repository = repository;
         this.backend = msgResponseBackend;
         this.taskExecutor = taskExecutor;
@@ -68,8 +78,14 @@ public class MsgService implements Zuse2AppPort {
     @Override
     public DeliveryRequestStatusACKType status(DeliveryRequestStatusType status) {
         var response = new RequestStatusResponse(status);
-        sendToWork(response);
-        return statusAck(response.getAppDeliveryID(), response.getZSDeliveryID());
+        try {
+            sendToBackend(response).get();
+            return statusAck(response.getAppDeliveryID(), response.getZSDeliveryID());
+        } catch (InterruptedException | ExecutionException e) {
+            var message = format(BACKEND_ERROR_MSG, "request status");
+            log.error(message, e);
+            throw toFault(e, message);
+        }
     }
 
     private DeliveryRequestStatusACKType statusAck(String appDeliveryID, String zsDeliveryID) {
@@ -83,13 +99,24 @@ public class MsgService implements Zuse2AppPort {
     @Override
     public DeliveryNotificationACKType notification(DeliveryNotificationType notification) {
         var response = new NotificationResponse(notification);
-        sendToWork(response);
-        return notificationAck(response.getAppDeliveryID(), response.getZSDeliveryID());
+        try {
+            sendToBackend(response).get();
+            return notificationAck(response.getAppDeliveryID(), response.getZSDeliveryID());
+        } catch (InterruptedException | ExecutionException e) {
+            var message = format(BACKEND_ERROR_MSG, "delivery notification");
+            log.error(message, e);
+            throw toFault(e, message);
+        }
+    }
+
+    private RuntimeException toFault(Exception e, String fallbackMessage) {
+        Throwable cause = (e.getCause() != null) ? e.getCause() : e;
+        return new SoapFault(fallbackMessage, cause, new QName("faultcode"));
     }
 
-    private void sendToWork(MsgResponse response) {
-        runAsync(() -> repository.store(response), taskExecutor)
-            .thenRunAsync(() -> backend.accept(response.getResponseID()), taskExecutor);
+    private CompletableFuture<Void> sendToBackend(MsgResponse response) {
+        return runAsync(() -> repository.store(response), taskExecutor)
+            .thenCompose(ignore -> backend.accept(response.getResponseID()));
     }
 
     private DeliveryNotificationACKType notificationAck(String appDeliveryID, String zsDeliveryID) {
diff --git a/src/main/java/at/gv/egiz/moazs/service/MzsService.java b/src/main/java/at/gv/egiz/moazs/service/MzsService.java
index ee14af8..caaa251 100644
--- a/src/main/java/at/gv/egiz/moazs/service/MzsService.java
+++ b/src/main/java/at/gv/egiz/moazs/service/MzsService.java
@@ -142,7 +142,8 @@ public class MzsService implements App2MzsPortType {
 
         log.info(SERVICE_TIME_OUT_REACHED_MSG, appDeliveryID);
         var sinkParams = request.getConfig().getMsgResponseSinks();
-        requestProcessed.thenAcceptAsync(response -> hub.applySinks(response, sinkParams), taskExecutor);
+
+        requestProcessed.thenCompose(response -> hub.applySinks(response, sinkParams));
 
         return generatePartialSuccessResponse(appDeliveryID);
     }