Reverted to check if shouldRequest, occasional double messages are se…

…nt but the check is to see if the request is needed, rather than should it insert into local table.
ed-pilots-network · Mar 18, 2024 · ae8ab50 · ae8ab50
1 parent f36442b
commit ae8ab50
Show file tree

Hide file tree

Showing 7 changed files with 69 additions and 57 deletions.
diff --git a/...kend/trade/application/service/StationArrivalDistanceInterModuleCommunicationService.java b/...kend/trade/application/service/StationArrivalDistanceInterModuleCommunicationService.java
@@ -133,15 +133,17 @@ public boolean isApplicable(Station station) {
     public synchronized void request(Station station) {
         String stationName = station.name();
         String systemName = station.system().name();
-        StationDataRequest stationDataRequest = new StationDataRequest(
-                Module.TRADE, station.name(), station.system().name()
-        );
-
-        JsonNode jsonNode = objectMapper.valueToTree(stationDataRequest);
-        Message message = new Message(Topic.Request.STATION_ARRIVAL_DISTANCE.getTopicName(), jsonNode.toString());
-
-        sendKafkaMessagePort.send(message);
-        createIfNotExistsStationArrivalDistanceRequestPort.createIfNotExists(systemName, stationName);
-
+        boolean shouldRequest = !existsStationArrivalDistanceRequestPort.exists(systemName, stationName);
+        if (shouldRequest) {
+            StationDataRequest stationDataRequest = new StationDataRequest(
+                    Module.TRADE, station.name(), station.system().name()
+            );
+
+            JsonNode jsonNode = objectMapper.valueToTree(stationDataRequest);
+            Message message = new Message(Topic.Request.STATION_ARRIVAL_DISTANCE.getTopicName(), jsonNode.toString());
+
+            sendKafkaMessagePort.send(message);
+            createIfNotExistsStationArrivalDistanceRequestPort.createIfNotExists(systemName, stationName);
+        }
     }
 }
diff --git a/...ckend/trade/application/service/StationLandingPadSizeInterModuleCommunicationService.java b/...ckend/trade/application/service/StationLandingPadSizeInterModuleCommunicationService.java
@@ -66,16 +66,18 @@ public boolean isApplicable(Station station) {
     public synchronized void request(Station station) {
         String stationName = station.name();
         String systemName = station.system().name();
-        StationDataRequest stationDataRequest = new StationDataRequest(
-                Module.TRADE, stationName, systemName
-        );
-        JsonNode jsonNode = objectMapper.valueToTree(stationDataRequest);
-
-        Message message = new Message(Topic.Request.STATION_MAX_LANDING_PAD_SIZE.getTopicName(), jsonNode.toString());
-
-        sendKafkaMessagePort.send(message);
-        createIfNotExistsStationLandingPadSizeRequestPort.createIfNotExists(systemName, stationName);
-
+        boolean shouldRequest = !existsStationLandingPadSizeRequestPort.exists(systemName, stationName);
+        if (shouldRequest) {
+            StationDataRequest stationDataRequest = new StationDataRequest(
+                    Module.TRADE, stationName, systemName
+            );
+            JsonNode jsonNode = objectMapper.valueToTree(stationDataRequest);
+
+            Message message = new Message(Topic.Request.STATION_MAX_LANDING_PAD_SIZE.getTopicName(), jsonNode.toString());
+
+            sendKafkaMessagePort.send(message);
+            createIfNotExistsStationLandingPadSizeRequestPort.createIfNotExists(systemName, stationName);
+        }
     }
 
     @Override

diff --git a/...pn/backend/trade/application/service/StationPlanetaryInterModuleCommunicationService.java b/...pn/backend/trade/application/service/StationPlanetaryInterModuleCommunicationService.java
@@ -65,16 +65,18 @@ public boolean isApplicable(Station station) {
     public synchronized void request(Station station) {
         String stationName = station.name();
         String systemName = station.system().name();
-        StationDataRequest stationDataRequest = new StationDataRequest(
-                Module.TRADE, stationName, systemName
-        );
-
-        JsonNode jsonNode = objectMapper.valueToTree(stationDataRequest);
-        Message message = new Message(Topic.Request.STATION_IS_PLANETARY.getTopicName(), jsonNode.toString());
-
-        sendKafkaMessagePort.send(message);
-        createIfNotExistsStationPlanetaryRequestPort.createIfNotExists(systemName, stationName);
-
+        boolean shouldRequest = !existsStationPlanetaryRequestPort.exists(systemName, stationName);
+        if (shouldRequest) {
+            StationDataRequest stationDataRequest = new StationDataRequest(
+                    Module.TRADE, stationName, systemName
+            );
+
+            JsonNode jsonNode = objectMapper.valueToTree(stationDataRequest);
+            Message message = new Message(Topic.Request.STATION_IS_PLANETARY.getTopicName(), jsonNode.toString());
+
+            sendKafkaMessagePort.send(message);
+            createIfNotExistsStationPlanetaryRequestPort.createIfNotExists(systemName, stationName);
+        }
     }
 
     @Override

diff --git a/...ckend/trade/application/service/StationRequireOdysseyInterModuleCommunicationService.java b/...ckend/trade/application/service/StationRequireOdysseyInterModuleCommunicationService.java
@@ -65,15 +65,17 @@ public boolean isApplicable(Station station) {
     public synchronized void request(Station station) {
         String stationName = station.name();
         String systemName = station.system().name();
-        StationDataRequest stationDataRequest = new StationDataRequest(
-                Module.TRADE, stationName, systemName
-        );
-        JsonNode jsonNode = objectMapper.valueToTree(stationDataRequest);
-        Message message = new Message(Topic.Request.STATION_REQUIRE_ODYSSEY.getTopicName(), jsonNode.toString());
-
-        sendKafkaMessagePort.send(message);
-        createIfNotExistsStationRequireOdysseyRequestPort.createIfNotExists(systemName, stationName);
-
+        boolean shouldRequest = !existsStationRequireOdysseyRequestPort.exists(systemName, stationName);
+        if (shouldRequest) {
+            StationDataRequest stationDataRequest = new StationDataRequest(
+                    Module.TRADE, stationName, systemName
+            );
+            JsonNode jsonNode = objectMapper.valueToTree(stationDataRequest);
+            Message message = new Message(Topic.Request.STATION_REQUIRE_ODYSSEY.getTopicName(), jsonNode.toString());
+
+            sendKafkaMessagePort.send(message);
+            createIfNotExistsStationRequireOdysseyRequestPort.createIfNotExists(systemName, stationName);
+        }
     }
 
     @Override

diff --git a/...pn/backend/trade/application/service/SystemCoordinateInterModuleCommunicationService.java b/...pn/backend/trade/application/service/SystemCoordinateInterModuleCommunicationService.java
@@ -61,14 +61,16 @@ public boolean isApplicable(System system) {
     @Override
     public synchronized void request(System system) {
         final String systemName = system.name();
-        SystemDataRequest systemDataRequest = new SystemDataRequest(Module.TRADE, systemName);
-
-        JsonNode jsonNode = objectMapper.valueToTree(systemDataRequest);
-        Message message = new Message(Topic.Request.SYSTEM_COORDINATES.getTopicName(), jsonNode.toString());
-
-        sendKafkaMessagePort.send(message);
-        createIfNotExistsSystemCoordinateRequestPort.createIfNotExists(systemName);
-
+        boolean shouldRequest = !existsSystemCoordinateRequestPort.exists(systemName);
+        if (shouldRequest) {
+            SystemDataRequest systemDataRequest = new SystemDataRequest(Module.TRADE, systemName);
+
+            JsonNode jsonNode = objectMapper.valueToTree(systemDataRequest);
+            Message message = new Message(Topic.Request.SYSTEM_COORDINATES.getTopicName(), jsonNode.toString());
+
+            sendKafkaMessagePort.send(message);
+            createIfNotExistsSystemCoordinateRequestPort.createIfNotExists(systemName);
+        }
     }
 
     @Override

diff --git a/.../edpn/backend/trade/application/service/SystemEliteIdInterModuleCommunicationService.java b/.../edpn/backend/trade/application/service/SystemEliteIdInterModuleCommunicationService.java
@@ -60,14 +60,17 @@ public boolean isApplicable(System system) {
     @Override
     public synchronized void request(System system) {
         String systemName = system.name();
-        SystemDataRequest systemDataRequest = new SystemDataRequest(
-                Module.TRADE, systemName
-        );
-        JsonNode jsonNode = objectMapper.valueToTree(systemDataRequest);
-        Message message = new Message(Topic.Request.SYSTEM_ELITE_ID.getTopicName(), jsonNode.toString());
-
-        sendKafkaMessagePort.send(message);
-        createIfNotExistsSystemEliteIdRequestPort.createIfNotExists(systemName);
+        boolean shouldRequest = !existsSystemEliteIdRequestPort.exists(systemName);
+        if (shouldRequest) {
+            SystemDataRequest systemDataRequest = new SystemDataRequest(
+                    Module.TRADE, systemName
+            );
+            JsonNode jsonNode = objectMapper.valueToTree(systemDataRequest);
+            Message message = new Message(Topic.Request.SYSTEM_ELITE_ID.getTopicName(), jsonNode.toString());
+
+            sendKafkaMessagePort.send(message);
+            createIfNotExistsSystemEliteIdRequestPort.createIfNotExists(systemName);
+        }
     }
 
     @Override

diff --git a/...a/io/edpn/backend/trade/application/service/RequestStationArrivalDistanceServiceTest.java b/...a/io/edpn/backend/trade/application/service/RequestStationArrivalDistanceServiceTest.java
@@ -170,8 +170,7 @@ public void testRequestWhenIdDoesNotExist() {
         JsonNode mockJsonNode = mock(JsonNode.class);
         String mockJsonString = "jsonString";
         Message message = new Message(Topic.Request.STATION_ARRIVAL_DISTANCE.getTopicName(), "jsonString");
-
-        when(existsStationArrivalDistanceRequestPort.exists(systemName, stationName)).thenReturn(false);
+
         when(objectMapper.valueToTree(argThat(arg -> {
             if (arg instanceof StationDataRequest stationDataRequest) {
                 return systemName.equals(stationDataRequest.systemName()) && stationName.equals(stationDataRequest.stationName()) && Module.TRADE.equals(stationDataRequest.requestingModule());