Notification for voltage init reactive slacks over a threshold value (#70)

* Notification for voltage init reactive slacks over a threshold value
* Rollback in message payload type from NetworkImpactInfos to String.

Signed-off-by: Franck LECUYER <[email protected]>

Author: FranckLecuyer

src/main/java/org/gridsuite/study/notification/server/NotificationWebSocketHandler.java

@@ -13,7 +13,6 @@
 import org.gridsuite.study.notification.server.dto.Filters;
 import org.gridsuite.study.notification.server.dto.FiltersToAdd;
 import org.gridsuite.study.notification.server.dto.FiltersToRemove;
-import org.gridsuite.study.notification.server.dto.NetworkImpactsInfos;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.beans.factory.annotation.Value;
@@ -95,12 +94,12 @@ private void initMetrics(MeterRegistry meterRegistry) {
         Gauge.builder(CONNECTIONS_METER_NAME, () -> userConnections.values().stream().mapToInt(Integer::intValue).sum()).register(meterRegistry);
     }
 
-    Flux<Message<NetworkImpactsInfos>> flux;
+    Flux<Message<String>> flux;
 
     @Bean
-    public Consumer<Flux<Message<NetworkImpactsInfos>>> consumeNotification() {
+    public Consumer<Flux<Message<String>>> consumeNotification() {
         return f -> {
-            ConnectableFlux<Message<NetworkImpactsInfos>> c = f.log(CATEGORY_BROKER_INPUT, Level.FINE).publish();
+            ConnectableFlux<Message<String>> c = f.log(CATEGORY_BROKER_INPUT, Level.FINE).publish();
             this.flux = c;
             c.connect();
             // Force connect 1 fake subscriber to consumme messages as they come.

src/main/java/org/gridsuite/study/notification/server/dto/EquipmentDeletionInfos.java

@@ -17,7 +17,6 @@
 import com.fasterxml.jackson.core.JsonProcessingException;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.fasterxml.jackson.databind.ObjectWriter;
-import com.google.common.collect.ImmutableSet;
 
 import io.micrometer.core.instrument.MeterRegistry;
 import io.micrometer.core.instrument.simple.SimpleMeterRegistry;
@@ -133,7 +132,7 @@ private void withFilters(String filterStudyUuid, String filterUpdateType, boolea
         }
 
         var notificationWebSocketHandler = new NotificationWebSocketHandler(objectMapper, meterRegistry, Integer.MAX_VALUE);
-        var atomicRef = new AtomicReference<FluxSink<Message<NetworkImpactsInfos>>>();
+        var atomicRef = new AtomicReference<FluxSink<Message<String>>>();
         var flux = Flux.create(atomicRef::set);
         notificationWebSocketHandler.consumeNotification().accept(flux);
         var sink = atomicRef.get();
@@ -148,7 +147,7 @@ private void withFilters(String filterStudyUuid, String filterUpdateType, boolea
             }
         }
 
-        List<GenericMessage<NetworkImpactsInfos>> refMessages = Stream.<Map<String, Object>>of(
+        List<GenericMessage<String>> refMessages = Stream.<Map<String, Object>>of(
                 Map.of(HEADER_STUDY_UUID, "foo", HEADER_UPDATE_TYPE, "oof"),
                 Map.of(HEADER_STUDY_UUID, "bar", HEADER_UPDATE_TYPE, "oof"),
                 Map.of(HEADER_STUDY_UUID, "baz", HEADER_UPDATE_TYPE, "oof"),
@@ -174,7 +173,7 @@ private void withFilters(String filterStudyUuid, String filterUpdateType, boolea
                     HEADER_PARENT_NODE, UUID.randomUUID().toString(), HEADER_REMOVE_CHILDREN, true),
 
                 Map.of(HEADER_STUDY_UUID, "", HEADER_UPDATE_TYPE, "indexation_status_updated", HEADER_INDEXATION_STATUS, "INDEXED"))
-                .map(map -> new GenericMessage<>(new NetworkImpactsInfos(ImmutableSet.of(), ImmutableSet.of(new EquipmentDeletionInfos()), ImmutableSet.of()), map))
+                .map(map -> new GenericMessage<>("", map))
                 .collect(Collectors.toList());
 
         @SuppressWarnings("unchecked")
@@ -300,7 +299,7 @@ public void testWsReceiveFilters() throws JsonProcessingException {
         when(ws2.getAttributes()).thenReturn(map);
 
         var notificationWebSocketHandler = new NotificationWebSocketHandler(new ObjectMapper(), meterRegistry, 60);
-        var flux = Flux.<Message<NetworkImpactsInfos>>empty();
+        var flux = Flux.<Message<String>>empty();
         notificationWebSocketHandler.consumeNotification().accept(flux);
         notificationWebSocketHandler.receive(ws2).subscribe();
 
@@ -327,7 +326,7 @@ public void testWsRemoveFilters() throws JsonProcessingException {
         assertEquals("updateType", ws2.getAttributes().get(FILTER_UPDATE_TYPE));
         assertEquals("studyUuid", ws2.getAttributes().get(FILTER_STUDY_UUID));
         var notificationWebSocketHandler = new NotificationWebSocketHandler(new ObjectMapper(), meterRegistry, Integer.MAX_VALUE);
-        var flux = Flux.<Message<NetworkImpactsInfos>>empty();
+        var flux = Flux.<Message<String>>empty();
         notificationWebSocketHandler.consumeNotification().accept(flux);
         notificationWebSocketHandler.receive(ws2).subscribe();
 
@@ -348,7 +347,7 @@ public void testWsReceiveEmptyFilters() throws JsonProcessingException {
         when(ws2.getAttributes()).thenReturn(map);
 
         var notificationWebSocketHandler = new NotificationWebSocketHandler(new ObjectMapper(), meterRegistry, Integer.MAX_VALUE);
-        var flux = Flux.<Message<NetworkImpactsInfos>>empty();
+        var flux = Flux.<Message<String>>empty();
         notificationWebSocketHandler.consumeNotification().accept(flux);
         notificationWebSocketHandler.receive(ws2).subscribe();
 
@@ -366,7 +365,7 @@ public void testWsReceiveUnprocessableFilter() {
         when(ws2.getAttributes()).thenReturn(map);
 
         var notificationWebSocketHandler = new NotificationWebSocketHandler(new ObjectMapper(), meterRegistry, 60);
-        var flux = Flux.<Message<NetworkImpactsInfos>>empty();
+        var flux = Flux.<Message<String>>empty();
         notificationWebSocketHandler.consumeNotification().accept(flux);
         notificationWebSocketHandler.receive(ws2).subscribe();
 
@@ -379,7 +378,7 @@ public void testHeartbeat() {
         setUpUriComponentBuilder("userId");
 
         var notificationWebSocketHandler = new NotificationWebSocketHandler(null, meterRegistry, 1);
-        var flux = Flux.<Message<NetworkImpactsInfos>>empty();
+        var flux = Flux.<Message<String>>empty();
         notificationWebSocketHandler.consumeNotification().accept(flux);
         notificationWebSocketHandler.handle(ws);
 
@@ -394,13 +393,13 @@ public void testDiscard() {
         setUpUriComponentBuilder("userId");
 
         var notificationWebSocketHandler = new NotificationWebSocketHandler(objectMapper, meterRegistry, Integer.MAX_VALUE);
-        var atomicRef = new AtomicReference<FluxSink<Message<NetworkImpactsInfos>>>();
+        var atomicRef = new AtomicReference<FluxSink<Message<String>>>();
         var flux = Flux.create(atomicRef::set);
         notificationWebSocketHandler.consumeNotification().accept(flux);
         var sink = atomicRef.get();
         Map<String, Object> headers = Map.of(HEADER_STUDY_UUID, "foo", HEADER_UPDATE_TYPE, "oof");
 
-        sink.next(new GenericMessage<>(new NetworkImpactsInfos(ImmutableSet.of(), ImmutableSet.of(new EquipmentDeletionInfos()), ImmutableSet.of()), headers)); // should be discarded, no client connected
+        sink.next(new GenericMessage<>("", headers)); // should be discarded, no client connected
 
         notificationWebSocketHandler.handle(ws);
 
@@ -411,7 +410,7 @@ public void testDiscard() {
         Disposable d1 = out1.map(WebSocketMessage::getPayloadAsText).subscribe(messages1::add);
         d1.dispose();
 
-        sink.next(new GenericMessage<>(new NetworkImpactsInfos(ImmutableSet.of(), ImmutableSet.of(new EquipmentDeletionInfos()), ImmutableSet.of()), headers)); // should be discarded, first client disconnected
+        sink.next(new GenericMessage<>("", headers)); // should be discarded, first client disconnected
 
         notificationWebSocketHandler.handle(ws);