Merge pull request #37 from digitalservicebund/send_static_test_message

Send static test message

src/main/java/de/bund/digitalservice/a2j/Application.java

@@ -2,8 +2,10 @@
 
 import org.springframework.boot.SpringApplication;
 import org.springframework.boot.autoconfigure.SpringBootApplication;
+import org.springframework.scheduling.annotation.EnableScheduling;
 
 @SpringBootApplication
+@EnableScheduling
 public class Application {
 
   @Generated

src/main/java/de/bund/digitalservice/a2j/config/EgvpClientConfig.java

@@ -1,6 +1,6 @@
 package de.bund.digitalservice.a2j.config;
 
-import de.bund.digitalservice.a2j.service.egvp.EgvpClient;
+import de.bund.digitalservice.a2j.service.egvp.client.EgvpClient;
 import org.springframework.beans.factory.annotation.Value;
 import org.springframework.boot.web.client.RestTemplateBuilder;
 import org.springframework.context.annotation.Bean;
@@ -12,7 +12,7 @@
 public class EgvpClientConfig {
 
   @Bean
-  public EgvpClient egvpClient(@Value("egvp.client.baseUri") String baseUri) {
+  public EgvpClient egvpClient(@Value("${egvp.client.baseUri}") String baseUri) {
     RestTemplate restClient =
         new RestTemplateBuilder()
             .errorHandler(new DefaultResponseErrorHandler())

src/main/java/de/bund/digitalservice/a2j/config/SecurityConfig.java

@@ -22,7 +22,9 @@ public SecurityConfig(CallbackVerificationFilter callbackVerificationFilter) {
   @Bean
   public SecurityFilterChain springSecurityWebFilterChain(HttpSecurity http) throws Exception {
     return http.csrf(
-            csrf -> csrf.ignoringRequestMatchers("/api/sender/submit", "/callbacks/fit-connect"))
+            csrf ->
+                csrf.ignoringRequestMatchers(
+                    "/api/test/**", "/api/sender/submit", "/callbacks/fit-connect"))
         .authorizeHttpRequests(
             requests ->
                 requests

src/main/java/de/bund/digitalservice/a2j/controller/SubscriberController.java

@@ -1,7 +1,9 @@
 package de.bund.digitalservice.a2j.controller;
 
+import de.bund.digitalservice.a2j.service.egvp.client.EgvpClientException;
 import de.bund.digitalservice.a2j.service.subscriber.SubscriberService;
 import dev.fitko.fitconnect.api.domain.model.callback.NewSubmissionsCallback;
+import dev.fitko.fitconnect.api.domain.model.submission.SubmissionForPickup;
 import org.springframework.web.bind.annotation.PostMapping;
 import org.springframework.web.bind.annotation.RequestBody;
 import org.springframework.web.bind.annotation.RestController;
@@ -17,6 +19,12 @@ public SubscriberController(SubscriberService service) {
 
   @PostMapping("callbacks/fit-connect")
   public void newSubmission(@RequestBody NewSubmissionsCallback callback) {
-    callback.getSubmissions().forEach(service::pickUpSubmission);
+    for (SubmissionForPickup submissionForPickup : callback.getSubmissions()) {
+      try {
+        service.pickUpSubmission(submissionForPickup);
+      } catch (EgvpClientException e) {
+        throw new RuntimeException(e.getMessage());
+      }
+    }
   }
 }

src/main/java/de/bund/digitalservice/a2j/repository/egvp/MessageInTransit.java

@@ -0,0 +1,22 @@
+package de.bund.digitalservice.a2j.repository.egvp;
+
+public record MessageInTransit(String userId, String customId, String bundIdMailbox) {
+
+  public int hashCode() {
+
+    return customId.hashCode();
+  }
+
+  public boolean equals(Object obj) {
+    if (obj == null) {
+      return false;
+    }
+
+    if (obj.getClass() != this.getClass()) {
+      return false;
+    }
+
+    return customId.equals(((MessageInTransit) obj).customId)
+        && userId.equals(((MessageInTransit) obj).userId);
+  }
+}

src/main/java/de/bund/digitalservice/a2j/repository/egvp/MessagesInTransitInMemoryRepository.java

@@ -0,0 +1,24 @@
+package de.bund.digitalservice.a2j.repository.egvp;
+
+import java.util.HashSet;
+import java.util.Set;
+import org.springframework.stereotype.Component;
+
+@Component
+public class MessagesInTransitInMemoryRepository implements MessagesInTransitRepository {
+  public Set<MessageInTransit> messages = new HashSet<>();
+
+  public Set<MessageInTransit> getAll() {
+    return messages;
+  }
+
+  @Override
+  public void add(MessageInTransit message) {
+    this.messages.add(message);
+  }
+
+  @Override
+  public void removeAll(Set<MessageInTransit> messages) {
+    this.messages.removeAll(messages);
+  }
+}

src/main/java/de/bund/digitalservice/a2j/repository/egvp/MessagesInTransitRepository.java

@@ -0,0 +1,12 @@
+package de.bund.digitalservice.a2j.repository.egvp;
+
+import java.util.Set;
+
+public interface MessagesInTransitRepository {
+
+  public void add(MessageInTransit message);
+
+  public void removeAll(Set<MessageInTransit> messages);
+
+  public Set<MessageInTransit> getAll();
+}

src/main/java/de/bund/digitalservice/a2j/service/egvp/EgvpOutboxService.java

@@ -0,0 +1,75 @@
+package de.bund.digitalservice.a2j.service.egvp;
+
+import de.bund.digitalservice.a2j.repository.egvp.MessageInTransit;
+import de.bund.digitalservice.a2j.repository.egvp.MessagesInTransitRepository;
+import de.bund.digitalservice.a2j.service.egvp.client.EgvpClient;
+import de.bund.digitalservice.a2j.service.egvp.client.EgvpClientException;
+import de.bund.digitalservice.a2j.service.egvp.client.MessageDeliveryStatusResponse;
+import de.bund.digitalservice.a2j.service.egvp.client.SendMessageRequest;
+import de.bund.digitalservice.a2j.service.egvp.client.SendMessageResponse;
+import java.util.HashSet;
+import java.util.Set;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.scheduling.annotation.Scheduled;
+import org.springframework.stereotype.Service;
+
+@Service
+public class EgvpOutboxService {
+
+  private final EgvpClient client;
+
+  private final MessagesInTransitRepository repository;
+
+  private static final Logger logger = LoggerFactory.getLogger(EgvpOutboxService.class);
+
+  EgvpOutboxService(EgvpClient client, MessagesInTransitRepository repository) {
+    this.client = client;
+    this.repository = repository;
+  }
+
+  public String sendMessage(SendMessageRequest request) throws EgvpClientException {
+
+    SendMessageResponse response = this.client.sendMessage(request);
+    repository.add(
+        new MessageInTransit(request.userId(), response.customId(), request.bundIdMailbox()));
+    return response.customId();
+  }
+
+  @Scheduled(fixedRate = 5000)
+  public void retrieveDeliveryStatus() {
+    logger.info("POLLING MESSAGE DELIVERY");
+    Set<MessageInTransit> deliveredMessages = new HashSet<>();
+    Set<MessageInTransit> failedMessages = new HashSet<>();
+
+    Set<MessageInTransit> inTransit = repository.getAll();
+
+    inTransit.forEach(
+        m -> {
+          logger.info("checking status for message with customId {}", m.customId());
+          try {
+            MessageDeliveryStatusResponse response =
+                this.client.checkMessageStatus(m.userId(), m.customId());
+            if (response.delivered()) {
+              deliveredMessages.add(m);
+              // TODO send confirmation file to user
+            } else {
+              logger.info(
+                  "message {} with customId {} still in transit : {}",
+                  response.messageId(),
+                  m.customId(),
+                  response.pendingReason());
+            }
+          } catch (EgvpClientException e) {
+            failedMessages.add(m);
+            // TODO send failure to user
+            logger.error("Unable to deliver message :{}", e.getMessage());
+          }
+        });
+    repository.removeAll(
+        Stream.concat(deliveredMessages.stream(), failedMessages.stream())
+            .collect(Collectors.toSet()));
+  }
+}

src/main/java/de/bund/digitalservice/a2j/service/egvp/EgvpClient.java → src/main/java/de/bund/digitalservice/a2j/service/egvp/client/EgvpClient.java

@@ -1,11 +1,7 @@
-package de.bund.digitalservice.a2j.service.egvp;
+package de.bund.digitalservice.a2j.service.egvp.client;
 
-import de.bund.digitalservice.a2j.service.egvp.DTO.*;
 import java.util.Objects;
-import org.springframework.web.client.HttpClientErrorException;
-import org.springframework.web.client.HttpServerErrorException;
-import org.springframework.web.client.RestClientException;
-import org.springframework.web.client.RestTemplate;
+import org.springframework.web.client.*;
 
 public class EgvpClient {
   private final RestTemplate client;
@@ -18,17 +14,19 @@ public SendMessageResponse sendMessage(SendMessageRequest request) throws EgvpCl
     return invoke(
         () ->
             this.client
-                .postForEntity("/sendMessage", request, SendMessageResponse.class)
+                .postForEntity(
+                    "/sendMessage/{userId}", request, SendMessageResponse.class, request.userId())
                 .getBody());
   }
 
-  public MessageDeliveryStatusResponse checkMessageStatus(String customId)
+  public MessageDeliveryStatusResponse checkMessageStatus(String userId, String customId)
       throws EgvpClientException {
     return invoke(
         () ->
             this.client.getForObject(
-                "/getMessageDeliveryStatus/{customId}",
+                "/getMessageDeliveryStatus/{userId}/{customId}",
                 MessageDeliveryStatusResponse.class,
+                userId,
                 customId));
   }
 
@@ -48,6 +46,9 @@ private <T> T invoke(Operation<T> operation) throws EgvpClientException {
     } catch (HttpServerErrorException serverErrorException) {
       throw new EgvpClientException(
           serverErrorException.getStatusCode().toString(), serverErrorException);
+    } catch (RestClientException ex) {
+
+      throw new EgvpClientException(ex.getMessage().toString(), ex);
     }
   }
 }

src/main/java/de/bund/digitalservice/a2j/service/egvp/EgvpClientException.java → src/main/java/de/bund/digitalservice/a2j/service/egvp/client/EgvpClientException.java

@@ -1,6 +1,10 @@
-package de.bund.digitalservice.a2j.service.egvp;
+package de.bund.digitalservice.a2j.service.egvp.client;
 
 public class EgvpClientException extends Exception {
+  public EgvpClientException(String message) {
+    super(message);
+  }
+
   public EgvpClientException(String message, Throwable cause) {
     super(message, cause);
   }

src/main/java/de/bund/digitalservice/a2j/service/egvp/client/GetVersionResponse.java

@@ -0,0 +1,3 @@
+package de.bund.digitalservice.a2j.service.egvp.client;
+
+public record GetVersionResponse(String version) {}

src/main/java/de/bund/digitalservice/a2j/service/egvp/DTO/MessageDeliveryStatusResponse.java → src/main/java/de/bund/digitalservice/a2j/service/egvp/client/MessageDeliveryStatusResponse.java

@@ -1,4 +1,4 @@
-package de.bund.digitalservice.a2j.service.egvp.DTO;
+package de.bund.digitalservice.a2j.service.egvp.client;
 
 public record MessageDeliveryStatusResponse(
     String messageId, Boolean delivered, String pendingReason, String path) {}

src/main/java/de/bund/digitalservice/a2j/service/egvp/DTO/ResponseError.java → src/main/java/de/bund/digitalservice/a2j/service/egvp/client/ResponseError.java

@@ -1,3 +1,3 @@
-package de.bund.digitalservice.a2j.service.egvp.DTO;
+package de.bund.digitalservice.a2j.service.egvp.client;
 
 public record ResponseError(String responseCode, String errorDetail, String errorDescription) {}

src/main/java/de/bund/digitalservice/a2j/service/egvp/DTO/SendMessageRequest.java → src/main/java/de/bund/digitalservice/a2j/service/egvp/client/SendMessageRequest.java

@@ -1,6 +1,7 @@
-package de.bund.digitalservice.a2j.service.egvp.DTO;
+package de.bund.digitalservice.a2j.service.egvp.client;
 
 public record SendMessageRequest(
+    String userId,
     String receiverId,
     String bundIdMailbox,
     String subject,

src/main/java/de/bund/digitalservice/a2j/service/egvp/client/SendMessageResponse.java

@@ -0,0 +1,3 @@
+package de.bund.digitalservice.a2j.service.egvp.client;
+
+public record SendMessageResponse(String customId) {}

src/main/java/de/bund/digitalservice/a2j/service/subscriber/FitConnectSubscriberService.java

@@ -1,26 +1,46 @@
 package de.bund.digitalservice.a2j.service.subscriber;
 
+import de.bund.digitalservice.a2j.service.egvp.EgvpOutboxService;
+import de.bund.digitalservice.a2j.service.egvp.client.EgvpClientException;
+import de.bund.digitalservice.a2j.service.egvp.client.SendMessageRequest;
 import dev.fitko.fitconnect.api.domain.model.submission.SubmissionForPickup;
 import dev.fitko.fitconnect.api.domain.subscriber.ReceivedSubmission;
 import dev.fitko.fitconnect.client.SubscriberClient;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.annotation.Value;
 import org.springframework.stereotype.Service;
 
+import java.util.Objects;
+
 @Service
 public class FitConnectSubscriberService implements SubscriberService {
 
   private final SubscriberClient client;
+  private final EgvpOutboxService egvpService;
+  private final String testUserId;
   private static final Logger logger = LoggerFactory.getLogger(FitConnectSubscriberService.class);
 
-  public FitConnectSubscriberService(SubscriberClient client) {
+  public FitConnectSubscriberService(SubscriberClient client, EgvpOutboxService egvpService,  @Value("${egvp.client.testUserId}") String userId) {
     this.client = client;
+    this.egvpService = egvpService;
+    this.testUserId = userId;
   }
 
-  public void pickUpSubmission(SubmissionForPickup submissionForPickup) {
+
+  public void pickUpSubmission(SubmissionForPickup submissionForPickup) throws EgvpClientException {
     ReceivedSubmission receivedSubmission = client.requestSubmission(submissionForPickup);
     logger.info("Submission requested. SubmissionId: {}", submissionForPickup.getSubmissionId());
 
+    this.egvpService.sendMessage(
+        new SendMessageRequest(
+            testUserId,
+            testUserId,
+            "mailbox?",
+            "testmessage_" + receivedSubmission.getCaseId(),
+            Objects.requireNonNull(this.getClass().getClassLoader().getResource("test/hello_world.pdf")).getPath(),
+            Objects.requireNonNull(this.getClass().getClassLoader().getResource("test/xjustiz_nachricht.xml")).getPath()));
+
     receivedSubmission.acceptSubmission();
     logger.info("Submission accepted. CaseId: {}", receivedSubmission.getCaseId());
   }

src/main/java/de/bund/digitalservice/a2j/service/subscriber/SubscriberService.java

@@ -1,8 +1,9 @@
 package de.bund.digitalservice.a2j.service.subscriber;
 
+import de.bund.digitalservice.a2j.service.egvp.client.EgvpClientException;
 import dev.fitko.fitconnect.api.domain.model.submission.SubmissionForPickup;
 
 public interface SubscriberService {
 
-  void pickUpSubmission(SubmissionForPickup submissionForPickup);
+  void pickUpSubmission(SubmissionForPickup submissionForPickup) throws EgvpClientException;
 }

src/main/resources/application-local.yaml.example

@@ -16,5 +16,9 @@ fitConnect:
     jsonUri: https://schema.fitko.de/fim/s17000717_1.0.schema.json
   callbackSecret: s3cr3t
 egvp:
+  outbox:
+    path: 'path/to/outbox/'
+  inbox:
+    path: 'path/to/inbox/'
   client:
-    baseUri: "localhost:8088"
+    baseUri: 'localhost:8088'

src/main/resources/application.yaml

@@ -18,3 +18,7 @@ management:
       enabled: true
   server:
     port: 8081
+egvp:
+  client:
+    testUserId:
+    baseUri: