Response callback support in asyncCounterpartXyz()

dcsaorg · Nov 23, 2023 · deb9d02 · deb9d02
1 parent df5bd39
commit deb9d02
Show file tree

Hide file tree

Showing 12 changed files with 99 additions and 56 deletions.
diff --git a/booking/src/main/java/org/dcsa/conformance/standards/booking/BookingComponentFactory.java b/booking/src/main/java/org/dcsa/conformance/standards/booking/BookingComponentFactory.java
@@ -2,6 +2,7 @@
 
 import com.fasterxml.jackson.databind.JsonNode;
 import java.util.*;
+import java.util.function.BiConsumer;
 import java.util.function.Consumer;
 import java.util.function.Function;
 import java.util.stream.Collectors;
@@ -15,6 +16,7 @@
 import org.dcsa.conformance.core.state.JsonNodeMap;
 import org.dcsa.conformance.core.toolkit.JsonToolkit;
 import org.dcsa.conformance.core.traffic.ConformanceRequest;
+import org.dcsa.conformance.core.traffic.ConformanceResponse;
 import org.dcsa.conformance.standards.booking.party.BookingRole;
 import org.dcsa.conformance.standards.booking.party.Carrier;
 import org.dcsa.conformance.standards.booking.party.Shipper;
@@ -40,7 +42,7 @@ public List<ConformanceParty> createParties(
       PartyConfiguration[] partyConfigurations,
       CounterpartConfiguration[] counterpartConfigurations,
       JsonNodeMap persistentMap,
-      Consumer<ConformanceRequest> asyncWebClient,
+      BiConsumer<ConformanceRequest, Consumer<ConformanceResponse>> asyncWebClient,
       Map<String, ? extends Collection<String>> orchestratorAuthHeader) {
     Map<String, PartyConfiguration> partyConfigurationsByRoleName =
         Arrays.stream(partyConfigurations)

diff --git a/booking/src/main/java/org/dcsa/conformance/standards/booking/party/Carrier.java b/booking/src/main/java/org/dcsa/conformance/standards/booking/party/Carrier.java
@@ -4,6 +4,15 @@
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.fasterxml.jackson.databind.node.ArrayNode;
 import com.fasterxml.jackson.databind.node.ObjectNode;
+import java.time.LocalDate;
+import java.time.OffsetDateTime;
+import java.time.ZoneOffset;
+import java.util.*;
+import java.util.function.BiConsumer;
+import java.util.function.Consumer;
+import java.util.function.Function;
+import java.util.function.Predicate;
+import java.util.stream.StreamSupport;
 import lombok.Builder;
 import lombok.SneakyThrows;
 import lombok.extern.slf4j.Slf4j;
@@ -18,17 +27,6 @@
 import org.dcsa.conformance.core.traffic.ConformanceResponse;
 import org.dcsa.conformance.standards.booking.action.*;
 
-import java.net.MalformedURLException;
-import java.net.URL;
-import java.time.LocalDate;
-import java.time.OffsetDateTime;
-import java.time.ZoneOffset;
-import java.util.*;
-import java.util.function.Consumer;
-import java.util.function.Function;
-import java.util.function.Predicate;
-import java.util.stream.StreamSupport;
-
 @Slf4j
 public class Carrier extends ConformanceParty {
   private static final Random RANDOM = new Random();
@@ -42,7 +40,7 @@ public Carrier(
       PartyConfiguration partyConfiguration,
       CounterpartConfiguration counterpartConfiguration,
       JsonNodeMap persistentMap,
-      Consumer<ConformanceRequest> asyncWebClient,
+      BiConsumer<ConformanceRequest, Consumer<ConformanceResponse>> asyncWebClient,
       Map<String, ? extends Collection<String>> orchestratorAuthHeader) {
     super(
         apiVersion,

diff --git a/booking/src/main/java/org/dcsa/conformance/standards/booking/party/Shipper.java b/booking/src/main/java/org/dcsa/conformance/standards/booking/party/Shipper.java
@@ -4,6 +4,7 @@
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.fasterxml.jackson.databind.node.ObjectNode;
 import java.util.*;
+import java.util.function.BiConsumer;
 import java.util.function.Consumer;
 import lombok.extern.slf4j.Slf4j;
 import org.dcsa.conformance.core.party.ConformanceParty;
@@ -26,7 +27,7 @@ public Shipper(
       PartyConfiguration partyConfiguration,
       CounterpartConfiguration counterpartConfiguration,
       JsonNodeMap persistentMap,
-      Consumer<ConformanceRequest> asyncWebClient,
+      BiConsumer<ConformanceRequest, Consumer<ConformanceResponse>> asyncWebClient,
       Map<String, ? extends Collection<String>> orchestratorAuthHeader) {
     super(
         apiVersion,
@@ -77,15 +78,27 @@ private void sendBookingRequest(JsonNode actionPrompt) {
 
     JsonNode jsonRequestBody =
         JsonToolkit.templateFileToJsonNode(
-          "/standards/booking/messages/booking-api-v20-request.json",
+            "/standards/booking/messages/booking-api-v20-request.json",
             Map.ofEntries(
                 Map.entry(
                     "CARRIER_SERVICE_NAME_PLACEHOLDER",
                     carrierScenarioParameters.carrierServiceName()),
                 Map.entry(
                     "VESSEL_IMO_NUMBER_PLACEHOLDER", carrierScenarioParameters.vesselIMONumber())));
 
-    asyncCounterpartPost("/v2/bookings", jsonRequestBody);
+    asyncCounterpartPost(
+        "/v2/bookings",
+        jsonRequestBody,
+        conformanceResponse -> {
+          JsonNode jsonBody = conformanceResponse.message().body().getJsonBody();
+          String cbrr = jsonBody.get("carrierBookingRequestReference").asText();
+          String bookingStatus = jsonBody.get("bookingStatus").asText();
+          ObjectNode updatedBooking =
+                ((ObjectNode) jsonRequestBody)
+                  .put("bookingStatus", bookingStatus)
+                  .put("carrierBookingRequestReference", cbrr);
+          persistentMap.save(cbrr, updatedBooking);
+        });
 
     addOperatorLogEntry(
         "Sent a booking request with the parameters: %s"
@@ -97,14 +110,12 @@ private void sendCancelEntireBooking(JsonNode actionPrompt) {
     String cbrr = actionPrompt.get("cbrr").asText();
 
     asyncCounterpartPatch(
-      "/v2/bookings/%s?operation=cancelBooking".formatted(cbrr),
-      new ObjectMapper().createObjectNode()
-        .put("bookingStatus", BookingState.CANCELLED.wireName())
-    );
+        "/v2/bookings/%s?operation=cancelBooking".formatted(cbrr),
+        new ObjectMapper()
+            .createObjectNode()
+            .put("bookingStatus", BookingState.CANCELLED.wireName()));
 
-    addOperatorLogEntry(
-      "Sent a cancel booking request of '%s'"
-        .formatted(cbrr));
+    addOperatorLogEntry("Sent a cancel booking request of '%s'".formatted(cbrr));
   }
 
   @Override

diff --git a/core/src/main/java/org/dcsa/conformance/core/AbstractComponentFactory.java b/core/src/main/java/org/dcsa/conformance/core/AbstractComponentFactory.java
@@ -1,25 +1,25 @@
 package org.dcsa.conformance.core;
 
+import com.fasterxml.jackson.databind.JsonNode;
 import java.util.*;
+import java.util.function.BiConsumer;
 import java.util.function.Consumer;
 import java.util.function.Predicate;
 import java.util.stream.Stream;
-
-import com.fasterxml.jackson.databind.JsonNode;
-import org.dcsa.conformance.core.check.JsonSchemaValidator;
 import org.dcsa.conformance.core.party.ConformanceParty;
 import org.dcsa.conformance.core.party.CounterpartConfiguration;
 import org.dcsa.conformance.core.party.PartyConfiguration;
 import org.dcsa.conformance.core.scenario.ScenarioListBuilder;
 import org.dcsa.conformance.core.state.JsonNodeMap;
 import org.dcsa.conformance.core.traffic.ConformanceRequest;
+import org.dcsa.conformance.core.traffic.ConformanceResponse;
 
 public abstract class AbstractComponentFactory {
   public abstract List<ConformanceParty> createParties(
       PartyConfiguration[] partyConfigurations,
       CounterpartConfiguration[] counterpartConfigurations,
       JsonNodeMap persistentMap,
-      Consumer<ConformanceRequest> asyncWebClient,
+      BiConsumer<ConformanceRequest, Consumer<ConformanceResponse>> asyncWebClient,
       Map<String, ? extends Collection<String>> orchestratorAuthHeader);
 
   public abstract ScenarioListBuilder<?> createScenarioListBuilder(

diff --git a/core/src/main/java/org/dcsa/conformance/core/party/ConformanceParty.java b/core/src/main/java/org/dcsa/conformance/core/party/ConformanceParty.java
@@ -10,6 +10,7 @@
 import java.net.http.HttpResponse;
 import java.time.Duration;
 import java.util.*;
+import java.util.function.BiConsumer;
 import java.util.function.Consumer;
 import java.util.stream.StreamSupport;
 import lombok.SneakyThrows;
@@ -29,15 +30,14 @@ public abstract class ConformanceParty implements StatefulEntity {
   protected final CounterpartConfiguration counterpartConfiguration;
 
   /**
-   * Used to store full documents between steps.
-   * Unlike the state saved and loaded via exportJsonState and importJsonState,
-   * which is entirely (the whole map) stored within a single DynamoDB item,
-   * the items in this map are separately stored each in its own DynamoDB item.
-   * This is to avoid growing the state map past the size limit of DynamoDB items.
+   * Used to store full documents between steps. Unlike the state saved and loaded via
+   * exportJsonState and importJsonState, which is entirely (the whole map) stored within a single
+   * DynamoDB item, the items in this map are separately stored each in its own DynamoDB item. This
+   * is to avoid growing the state map past the size limit of DynamoDB items.
    */
   protected final JsonNodeMap persistentMap;
 
-  private final Consumer<ConformanceRequest> asyncWebClient;
+  private final BiConsumer<ConformanceRequest, Consumer<ConformanceResponse>> asyncWebClient;
   private final Map<String, ? extends Collection<String>> orchestratorAuthHeader;
   private final ActionPromptsQueue actionPromptsQueue = new ActionPromptsQueue();
 
@@ -49,7 +49,7 @@ public ConformanceParty(
       PartyConfiguration partyConfiguration,
       CounterpartConfiguration counterpartConfiguration,
       JsonNodeMap persistentMap,
-      Consumer<ConformanceRequest> asyncWebClient,
+      BiConsumer<ConformanceRequest, Consumer<ConformanceResponse>> asyncWebClient,
       Map<String, ? extends Collection<String>> orchestratorAuthHeader) {
     this.apiVersion = apiVersion;
     this.partyConfiguration = partyConfiguration;
@@ -140,10 +140,15 @@ protected void asyncOrchestratorPostPartyInput(JsonNode jsonPartyInput) {
                 "orchestrator",
                 orchestratorAuthHeader,
                 new ConformanceMessageBody(jsonPartyInput),
-                System.currentTimeMillis())));
+                System.currentTimeMillis())),
+        conformanceResponse -> {});
   }
 
   protected void asyncCounterpartGet(String path) {
+    asyncCounterpartGet(path, conformanceResponse -> {});
+  }
+
+  protected void asyncCounterpartGet(String path, Consumer<ConformanceResponse> responseCallback) {
     asyncWebClient.accept(
         new ConformanceRequest(
             "GET",
@@ -162,27 +167,40 @@ protected void asyncCounterpartGet(String path) {
                         counterpartConfiguration.getAuthHeaderName(),
                         List.of(counterpartConfiguration.getAuthHeaderValue())),
                 new ConformanceMessageBody(""),
-                System.currentTimeMillis())));
+                System.currentTimeMillis())),
+        responseCallback);
   }
 
   protected void asyncCounterpartPatch(String path, JsonNode jsonBody) {
-    _asyncCounterpartPatchPostOrPut("PATCH", path, Collections.emptyMap(), jsonBody);
+    asyncCounterpartPatch(path, jsonBody, conformanceResponse -> {});
   }
 
   protected void asyncCounterpartPost(String path, JsonNode jsonBody) {
-    _asyncCounterpartPatchPostOrPut("POST", path, Collections.emptyMap(), jsonBody);
+    asyncCounterpartPost(path, jsonBody, conformanceResponse -> {});
   }
 
   protected void asyncCounterpartPut(String path, JsonNode jsonBody) {
-    _asyncCounterpartPatchPostOrPut("PUT", path, Collections.emptyMap(), jsonBody);
+    asyncCounterpartPut(path, jsonBody, conformanceResponse -> {});
+  }
+
+  protected void asyncCounterpartPatch(String path, JsonNode jsonBody, Consumer<ConformanceResponse> responseCallback) {
+    _asyncCounterpartPatchPostOrPut("PATCH", path, Collections.emptyMap(), jsonBody, responseCallback);
+  }
+
+  protected void asyncCounterpartPost(String path, JsonNode jsonBody, Consumer<ConformanceResponse> responseCallback) {
+    _asyncCounterpartPatchPostOrPut("POST", path, Collections.emptyMap(), jsonBody, responseCallback);
+  }
+
+  protected void asyncCounterpartPut(String path, JsonNode jsonBody, Consumer<ConformanceResponse> responseCallback) {
+    _asyncCounterpartPatchPostOrPut("PUT", path, Collections.emptyMap(), jsonBody, responseCallback);
   }
 
   private void _asyncCounterpartPatchPostOrPut(
-    String method,
-    String path,
-    Map<String, ? extends Collection<String>> queryParams,
-    JsonNode jsonBody
-  ) {
+      String method,
+      String path,
+      Map<String, ? extends Collection<String>> queryParams,
+      JsonNode jsonBody,
+      Consumer<ConformanceResponse> responseCallback) {
     asyncWebClient.accept(
         new ConformanceRequest(
             method,
@@ -201,7 +219,8 @@ private void _asyncCounterpartPatchPostOrPut(
                         counterpartConfiguration.getAuthHeaderName(),
                         List.of(counterpartConfiguration.getAuthHeaderValue())),
                 new ConformanceMessageBody(jsonBody),
-                System.currentTimeMillis())));
+                System.currentTimeMillis())),
+        responseCallback);
   }
 
   public abstract ConformanceResponse handleRequest(ConformanceRequest request);

diff --git a/...src/main/java/org/dcsa/conformance/standards/eblissuance/EblIssuanceComponentFactory.java b/...src/main/java/org/dcsa/conformance/standards/eblissuance/EblIssuanceComponentFactory.java
@@ -2,6 +2,7 @@
 
 import com.fasterxml.jackson.databind.JsonNode;
 import java.util.*;
+import java.util.function.BiConsumer;
 import java.util.function.Consumer;
 import java.util.function.Function;
 import java.util.stream.Collectors;
@@ -15,6 +16,7 @@
 import org.dcsa.conformance.core.state.JsonNodeMap;
 import org.dcsa.conformance.core.toolkit.JsonToolkit;
 import org.dcsa.conformance.core.traffic.ConformanceRequest;
+import org.dcsa.conformance.core.traffic.ConformanceResponse;
 import org.dcsa.conformance.standards.eblissuance.party.EblIssuanceCarrier;
 import org.dcsa.conformance.standards.eblissuance.party.EblIssuancePlatform;
 import org.dcsa.conformance.standards.eblissuance.party.EblIssuanceRole;
@@ -40,7 +42,7 @@ public List<ConformanceParty> createParties(
       PartyConfiguration[] partyConfigurations,
       CounterpartConfiguration[] counterpartConfigurations,
       JsonNodeMap persistentMap,
-      Consumer<ConformanceRequest> asyncWebClient,
+      BiConsumer<ConformanceRequest, Consumer<ConformanceResponse>> asyncWebClient,
       Map<String, ? extends Collection<String>> orchestratorAuthHeader) {
     Map<String, PartyConfiguration> partyConfigurationsByRoleName =
         Arrays.stream(partyConfigurations)

diff --git a/...ce/src/main/java/org/dcsa/conformance/standards/eblissuance/party/EblIssuanceCarrier.java b/...ce/src/main/java/org/dcsa/conformance/standards/eblissuance/party/EblIssuanceCarrier.java
@@ -4,6 +4,7 @@
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.fasterxml.jackson.databind.node.ObjectNode;
 import java.util.*;
+import java.util.function.BiConsumer;
 import java.util.function.Consumer;
 import lombok.extern.slf4j.Slf4j;
 import org.dcsa.conformance.core.party.ConformanceParty;
@@ -31,7 +32,7 @@ public EblIssuanceCarrier(
       PartyConfiguration partyConfiguration,
       CounterpartConfiguration counterpartConfiguration,
       JsonNodeMap persistentMap,
-      Consumer<ConformanceRequest> asyncWebClient,
+      BiConsumer<ConformanceRequest, Consumer<ConformanceResponse>> asyncWebClient,
       Map<String, ? extends Collection<String>> orchestratorAuthHeader) {
     super(
         apiVersion,

diff --git a/...e/src/main/java/org/dcsa/conformance/standards/eblissuance/party/EblIssuancePlatform.java b/...e/src/main/java/org/dcsa/conformance/standards/eblissuance/party/EblIssuancePlatform.java
@@ -5,6 +5,7 @@
 import com.fasterxml.jackson.databind.node.ArrayNode;
 import com.fasterxml.jackson.databind.node.ObjectNode;
 import java.util.*;
+import java.util.function.BiConsumer;
 import java.util.function.Consumer;
 import java.util.stream.StreamSupport;
 import lombok.extern.slf4j.Slf4j;
@@ -13,7 +14,6 @@
 import org.dcsa.conformance.core.party.PartyConfiguration;
 import org.dcsa.conformance.core.scenario.ConformanceAction;
 import org.dcsa.conformance.core.state.JsonNodeMap;
-import org.dcsa.conformance.core.state.StateManagementUtil;
 import org.dcsa.conformance.core.traffic.ConformanceMessageBody;
 import org.dcsa.conformance.core.traffic.ConformanceRequest;
 import org.dcsa.conformance.core.traffic.ConformanceResponse;
@@ -30,7 +30,7 @@ public EblIssuancePlatform(
       PartyConfiguration partyConfiguration,
       CounterpartConfiguration counterpartConfiguration,
       JsonNodeMap persistentMap,
-      Consumer<ConformanceRequest> asyncWebClient,
+      BiConsumer<ConformanceRequest, Consumer<ConformanceResponse>> asyncWebClient,
       Map<String, ? extends Collection<String>> orchestratorAuthHeader) {
     super(
         apiVersion,

diff --git a/...c/main/java/org/dcsa/conformance/standards/eblsurrender/EblSurrenderComponentFactory.java b/...c/main/java/org/dcsa/conformance/standards/eblsurrender/EblSurrenderComponentFactory.java
@@ -2,6 +2,7 @@
 
 import com.fasterxml.jackson.databind.JsonNode;
 import java.util.*;
+import java.util.function.BiConsumer;
 import java.util.function.Consumer;
 import java.util.function.Function;
 import java.util.stream.Collectors;
@@ -15,6 +16,7 @@
 import org.dcsa.conformance.core.state.JsonNodeMap;
 import org.dcsa.conformance.core.toolkit.JsonToolkit;
 import org.dcsa.conformance.core.traffic.ConformanceRequest;
+import org.dcsa.conformance.core.traffic.ConformanceResponse;
 import org.dcsa.conformance.standards.eblsurrender.party.EblSurrenderCarrier;
 import org.dcsa.conformance.standards.eblsurrender.party.EblSurrenderPlatform;
 import org.dcsa.conformance.standards.eblsurrender.party.EblSurrenderRole;
@@ -40,7 +42,7 @@ public List<ConformanceParty> createParties(
       PartyConfiguration[] partyConfigurations,
       CounterpartConfiguration[] counterpartConfigurations,
       JsonNodeMap persistentMap,
-      Consumer<ConformanceRequest> asyncWebClient,
+      BiConsumer<ConformanceRequest, Consumer<ConformanceResponse>> asyncWebClient,
       Map<String, ? extends Collection<String>> orchestratorAuthHeader) {
     Map<String, PartyConfiguration> partyConfigurationsByRoleName =
         Arrays.stream(partyConfigurations)

diff --git a/.../src/main/java/org/dcsa/conformance/standards/eblsurrender/party/EblSurrenderCarrier.java b/.../src/main/java/org/dcsa/conformance/standards/eblsurrender/party/EblSurrenderCarrier.java
@@ -5,6 +5,7 @@
 import com.fasterxml.jackson.databind.node.ArrayNode;
 import com.fasterxml.jackson.databind.node.ObjectNode;
 import java.util.*;
+import java.util.function.BiConsumer;
 import java.util.function.Consumer;
 import java.util.stream.StreamSupport;
 import lombok.extern.slf4j.Slf4j;
@@ -13,7 +14,6 @@
 import org.dcsa.conformance.core.party.PartyConfiguration;
 import org.dcsa.conformance.core.scenario.ConformanceAction;
 import org.dcsa.conformance.core.state.JsonNodeMap;
-import org.dcsa.conformance.core.state.StateManagementUtil;
 import org.dcsa.conformance.core.traffic.ConformanceMessageBody;
 import org.dcsa.conformance.core.traffic.ConformanceRequest;
 import org.dcsa.conformance.core.traffic.ConformanceResponse;
@@ -31,7 +31,7 @@ public EblSurrenderCarrier(
       PartyConfiguration partyConfiguration,
       CounterpartConfiguration counterpartConfiguration,
       JsonNodeMap persistentMap,
-      Consumer<ConformanceRequest> asyncWebClient,
+      BiConsumer<ConformanceRequest, Consumer<ConformanceResponse>> asyncWebClient,
       Map<String, ? extends Collection<String>> orchestratorAuthHeader) {
     super(
         apiVersion,