Finalize "Removal of JAXB Class Not Found Fix"

- Problem: I fixed CompletableFuture.*async within the DeliveryRequest
  flow, but forgot to do the same for the DeliveryRequestStatus /
  DeliveryNotification flow.
- Solution: Ensure that all CompletableFuture.*async() invocations
  have Spring's task executor attached.

6 files changed, 31 insertions, 16 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 0da9c92..81398da 100644
--- a/src/main/java/at/gv/egiz/moazs/backend/MsgResponseBackend.java
+++ b/src/main/java/at/gv/egiz/moazs/backend/MsgResponseBackend.java
@@ -31,6 +31,7 @@ import at.gv.zustellung.app2mzs.xsd.DeliveryRequestType;
 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.Component;
 
 import java.util.Map;
@@ -64,18 +65,21 @@ public class MsgResponseBackend implements Consumer<String> {
     private final MsgResponseSinkHub hub;
     private final Map<String, ConfigType> deliveryRequestConfigs;
     private final MzsDeliveryRequestValidator validator;
+    private final TaskExecutor taskExecutor;
 
     @Autowired
     public MsgResponseBackend(DeliveryRepository repository,
                               Consumer<byte[]> signatureVerifier,
                               MsgResponseSinkHub hub,
                               Map<String, ConfigType> deliveryRequestConfigs,
-                              MzsDeliveryRequestValidator validator) {
+                              MzsDeliveryRequestValidator validator,
+                              TaskExecutor taskExecutor) {
         this.repository = repository;
         this.signatureVerifier = signatureVerifier;
         this.hub = hub;
         this.deliveryRequestConfigs = deliveryRequestConfigs;
         this.validator = validator;
+        this.taskExecutor = taskExecutor;
     }
 
     /**
@@ -91,8 +95,8 @@ public class MsgResponseBackend implements Consumer<String> {
     @Override
     public void accept(String responseID) {
         log.info("Backend accepts MsgResponse with responseID={}.", responseID);
-        supplyAsync(() -> verifySignature(responseID))
-                .thenAcceptAsync(msgResponse -> applySinks(msgResponse))
+        supplyAsync(() -> verifySignature(responseID), taskExecutor)
+                .thenAcceptAsync(msgResponse -> applySinks(msgResponse), taskExecutor)
                 .exceptionally(ex -> {
                     log.error(ex.getMessage(), ex);
                     return null;
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 61df4b2..3927055 100644
--- a/src/main/java/at/gv/egiz/moazs/backend/MsgResponseSinkHub.java
+++ b/src/main/java/at/gv/egiz/moazs/backend/MsgResponseSinkHub.java
@@ -26,6 +26,7 @@ import at.gv.zustellung.app2mzs.xsd.MsgResponseSinksType;
 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.Component;
 
 import static java.util.concurrent.CompletableFuture.supplyAsync;
@@ -42,13 +43,15 @@ 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) {
+                              ForwardResponseToServiceSink forwardResponseSink, TaskExecutor taskExecutor) {
         this.saveResponseSink = saveResponseSink;
         this.logResponseSink = logResponseSink;
         this.forwardResponseSink = forwardResponseSink;
+        this.taskExecutor = taskExecutor;
     }
 
     /**
@@ -60,16 +63,16 @@ public class MsgResponseSinkHub {
                 msgResponse.getRootElementLocalPart(), msgResponse.getAppDeliveryID());
 
         if (sinkParams.getSaveResponseToFile().isActive()) {
-            supplyAsync(() -> saveResponseSink.save(msgResponse, sinkParams.getSaveResponseToFile().getPath()));
+            supplyAsync(() -> saveResponseSink.save(msgResponse, sinkParams.getSaveResponseToFile().getPath()), taskExecutor);
         }
 
         if (sinkParams.isLogResponse()) {
-            supplyAsync(() -> logResponseSink.log(msgResponse));
+            supplyAsync(() -> logResponseSink.log(msgResponse), taskExecutor);
         }
 
         if (sinkParams.getForwardResponseToService().isActive()) {
             supplyAsync(() -> forwardResponseSink.send(
-                    msgResponse, sinkParams.getForwardResponseToService().getAppClient()));
+                    msgResponse, sinkParams.getForwardResponseToService().getAppClient()), taskExecutor);
         }
 
     }
diff --git a/src/main/java/at/gv/egiz/moazs/backend/SaveResponseToFileSink.java b/src/main/java/at/gv/egiz/moazs/backend/SaveResponseToFileSink.java
index c2652d9..c844511 100644
--- a/src/main/java/at/gv/egiz/moazs/backend/SaveResponseToFileSink.java
+++ b/src/main/java/at/gv/egiz/moazs/backend/SaveResponseToFileSink.java
@@ -36,6 +36,7 @@ import java.nio.charset.StandardCharsets;
 import java.text.SimpleDateFormat;
 import java.util.Date;
 import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.Executor;
 
 import static at.gv.egiz.moazs.MoaZSException.moaZSException;
 import static java.lang.String.format;
@@ -56,11 +57,13 @@ public class SaveResponseToFileSink {
     private final SimpleDateFormat isoFormatter;
     private final Marshaller msgMarshaller;
     private final DeliveryRepository repository;
+    private final Executor taskExecutor;
 
     @Autowired
-    public SaveResponseToFileSink(Marshaller msgMarshaller, DeliveryRepository repository) {
+    public SaveResponseToFileSink(Marshaller msgMarshaller, DeliveryRepository repository, Executor taskExecutor) {
         this.msgMarshaller = msgMarshaller;
         this.repository = repository;
+        this.taskExecutor = taskExecutor;
         this.isoFormatter = new SimpleDateFormat("yyyy-MM-dd'T'HH:mm:ss");
     }
 
@@ -77,13 +80,13 @@ public class SaveResponseToFileSink {
                 response.getRootElementLocalPart(), response.getAppDeliveryID());
 
         var responsePath = generatePath(rootPath, responseID, "xml");
-        var storeResponseToFileSystemFuture = supplyAsync(() -> msgMarshaller.marshallXml(response.getResponseAsJAXBElement()))
+        var storeResponseToFileSystemFuture = supplyAsync(() -> msgMarshaller.marshallXml(response.getResponseAsJAXBElement()), taskExecutor)
                 .thenApply(responseString -> responseString.getBytes(StandardCharsets.UTF_8))
                 .thenAccept(responseByteArray -> storeToFile(responsePath, responseByteArray))
                 .exceptionally(ex -> logException(ex, responseID));
 
         var binaryResponsePath = generatePath(rootPath, responseID, "binary.xml");
-        var storeBinaryResponseToFileSystemFuture = supplyAsync(() -> repository.retrieveBinaryResponse(responseID).get())
+        var storeBinaryResponseToFileSystemFuture = supplyAsync(() -> repository.retrieveBinaryResponse(responseID).get(), taskExecutor)
                 .thenAccept(binaryResponseByteArray -> storeToFile(binaryResponsePath, binaryResponseByteArray))
                 .exceptionally(ex -> logException(ex, responseID));
 
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 8c71166..6b260e7 100644
--- a/src/main/java/at/gv/egiz/moazs/service/MsgService.java
+++ b/src/main/java/at/gv/egiz/moazs/service/MsgService.java
@@ -28,6 +28,7 @@ import at.gv.egiz.moazs.scheme.RequestStatusResponse;
 import at.gv.zustellung.msg.xsd.*;
 import org.apache.cxf.annotations.SchemaValidation;
 import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.core.task.TaskExecutor;
 import org.springframework.stereotype.Service;
 
 import java.util.function.Consumer;
@@ -43,16 +44,18 @@ import static java.util.concurrent.CompletableFuture.runAsync;
  */
 @Service
 @SchemaValidation(type = SchemaValidation.SchemaValidationType.BOTH)
-@org.apache.cxf.feature.Features (features = "org.apache.cxf.ext.logging.LoggingFeature")
+    @org.apache.cxf.feature.Features (features = "org.apache.cxf.ext.logging.LoggingFeature")
 public class MsgService implements Zuse2AppPort {
 
     private final DeliveryRepository repository;
     private final Consumer<String> backend;
+    private final TaskExecutor taskExecutor;
 
     @Autowired
-    public MsgService(DeliveryRepository repository, Consumer<String> msgResponseBackend) {
+    public MsgService(DeliveryRepository repository, Consumer<String> msgResponseBackend, TaskExecutor taskExecutor) {
         this.repository = repository;
         this.backend = msgResponseBackend;
+        this.taskExecutor = taskExecutor;
     }
 
     @Override
@@ -85,8 +88,8 @@ public class MsgService implements Zuse2AppPort {
     }
 
     private void sendToWork(MsgResponse response) {
-        runAsync(() -> repository.store(response))
-            .thenRunAsync(() -> backend.accept(response.getResponseID()));
+        runAsync(() -> repository.store(response), taskExecutor)
+            .thenRunAsync(() -> backend.accept(response.getResponseID()), taskExecutor);
     }
 
     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 55f0554..ee14af8 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,7 @@ 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));
+        requestProcessed.thenAcceptAsync(response -> hub.applySinks(response, sinkParams), taskExecutor);
 
         return generatePartialSuccessResponse(appDeliveryID);
     }
diff --git a/src/test/java/at/gv/egiz/moazs/SaveResponseToFileSinkTest.java b/src/test/java/at/gv/egiz/moazs/SaveResponseToFileSinkTest.java
index c1d742a..1a9b16b 100644
--- a/src/test/java/at/gv/egiz/moazs/SaveResponseToFileSinkTest.java
+++ b/src/test/java/at/gv/egiz/moazs/SaveResponseToFileSinkTest.java
@@ -33,11 +33,13 @@ import org.mockito.Mock;
 import org.mockito.junit.MockitoJUnitRunner;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
+import org.springframework.core.task.TaskExecutor;
 
 import java.io.File;
 import java.io.IOException;
 import java.nio.charset.StandardCharsets;
 import java.util.Collection;
+import java.util.concurrent.ForkJoinPool;
 
 import static at.gv.zustellung.msg.xsd.DeliveryRequestStatusType.Success.successBuilder;
 import static at.gv.zustellung.msg.xsd.DeliveryRequestStatusType.deliveryRequestStatusTypeBuilder;
@@ -69,7 +71,7 @@ public class SaveResponseToFileSinkTest {
 
     @Before
     public void setup() {
-        sink = new SaveResponseToFileSink(marshaller, repository);
+        sink = new SaveResponseToFileSink(marshaller, repository, ForkJoinPool.commonPool());
         deleteRoot();
 
     }