Merge branch 'main' of https://github.com/gridsuite/study-notification-server into main

# Conflicts:
#	src/main/java/org/gridsuite/study/notification/server/NotificationWebSocketHandler.java

Author: AAJELLAL

.attach_pid162551 renamed to .mvn/lombok-config-copy.marker

@@ -0,0 +1 @@
+import target/configs/powsybl-build-tools.jar!powsybl-build-tools/lombok.config

lombok.config

@@ -1,11 +1,9 @@
 <?xml version="1.0" encoding="UTF-8"?>
 <!--
-
     Copyright (c) 2020, RTE (http://www.rte-france.com)
     This Source Code Form is subject to the terms of the Mozilla Public
     License, v. 2.0. If a copy of the MPL was not distributed with this
     file, You can obtain one at http://mozilla.org/MPL/2.0/.
-
 -->
 <project xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"
          xmlns="http://maven.apache.org/POM/4.0.0"
@@ -15,7 +13,7 @@
     <parent>
         <groupId>com.powsybl</groupId>
         <artifactId>powsybl-parent-ws</artifactId>
-        <version>11</version>
+        <version>15</version>
         <relativePath/>
     </parent>
 
@@ -44,7 +42,7 @@
     </developers>
 
     <properties>
-        <gridsuite-dependencies.version>23</gridsuite-dependencies.version>
+        <gridsuite-dependencies.version>31</gridsuite-dependencies.version>
     </properties>
 
     <build>
@@ -62,8 +60,8 @@
                 <artifactId>spring-boot-maven-plugin</artifactId>
             </plugin>
             <plugin>
-                <groupId>pl.project13.maven</groupId>
-                <artifactId>git-commit-id-plugin</artifactId>
+                <groupId>io.github.git-commit-id</groupId>
+                <artifactId>git-commit-id-maven-plugin</artifactId>
             </plugin>
         </plugins>
     </build>
@@ -102,12 +100,17 @@
         <dependency>
             <groupId>org.projectlombok</groupId>
             <artifactId>lombok</artifactId>
+            <scope>provided</scope>
         </dependency>
-
-        <!-- Runtime dependencies -->
         <dependency>
             <groupId>org.springframework.boot</groupId>
             <artifactId>spring-boot-starter-actuator</artifactId>
+        </dependency>
+
+        <!-- Runtime dependencies -->
+        <dependency>
+            <groupId>io.micrometer</groupId>
+            <artifactId>micrometer-registry-prometheus</artifactId>
             <scope>runtime</scope>
         </dependency>
         <dependency>

pom.xml

@@ -6,7 +6,6 @@
  */
 package org.gridsuite.study.notification.server;
 
-import com.powsybl.ws.commons.Utils;
 import org.springframework.boot.SpringApplication;
 import org.springframework.boot.autoconfigure.SpringBootApplication;
 
@@ -18,7 +17,6 @@
 public class NotificationApplication {
 
     public static void main(String[] args) {
-        Utils.initProperties();
         SpringApplication.run(NotificationApplication.class, args);
     }
 }

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

@@ -6,22 +6,14 @@
  */
 package org.gridsuite.study.notification.server;
 
-import java.io.UnsupportedEncodingException;
-import java.net.URLDecoder;
-import java.nio.charset.StandardCharsets;
-import java.time.Duration;
-import java.util.HashMap;
-import java.util.Map;
-import java.util.function.Consumer;
-import java.util.logging.Level;
-
 import com.fasterxml.jackson.core.JsonProcessingException;
 import com.fasterxml.jackson.databind.ObjectMapper;
-
+import io.micrometer.core.instrument.MeterRegistry;
+import io.micrometer.core.instrument.MultiGauge;
+import io.micrometer.core.instrument.Tags;
 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;
@@ -37,6 +29,18 @@
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 
+import java.io.UnsupportedEncodingException;
+import java.net.URLDecoder;
+import java.nio.charset.StandardCharsets;
+import java.time.Duration;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.function.Consumer;
+import java.util.logging.Level;
+
+import static java.util.stream.Collectors.toList;
+
 /**
  * A WebSocketHandler that sends messages from a broker to websockets opened by clients, interleaving with pings to keep connections open.
  * <p>
@@ -63,36 +67,44 @@ public class NotificationWebSocketHandler implements WebSocketHandler {
     static final String HEADER_TIMESTAMP = "timestamp";
     static final String HEADER_ERROR = "error";
     static final String HEADER_SUBSTATIONS_IDS = "substationsIds";
-    static final String HEADER_DELETED_EQUIPMENTS = "deletedEquipments";
     static final String HEADER_NODE = "node";
     static final String HEADER_NODES = "nodes";
     static final String HEADER_PARENT_NODE = "parentNode";
     static final String HEADER_NEW_NODE = "newNode";
     static final String HEADER_MOVED_NODE = "movedNode";
     static final String HEADER_REMOVE_CHILDREN = "removeChildren";
     static final String HEADER_INSERT_MODE = "insertMode";
+    static final String HEADER_REFERENCE_NODE_UUID = "referenceNodeUuid";
+    static final String HEADER_INDEXATION_STATUS = "indexation_status";
+    static final String USERS_METER_NAME = "app.users";
+    static final String USER_TAG = "user";
+
+    private final ObjectMapper jacksonObjectMapper;
     public static final String HEADER_PARAMS_NAME = "paramsName";
 
-    private ObjectMapper jacksonObjectMapper;
+    private final int heartbeatInterval;
 
-    private int heartbeatInterval;
+    private final Map<String, Integer> userConnections = new ConcurrentHashMap<>();
 
-    public NotificationWebSocketHandler(ObjectMapper jacksonObjectMapper, @Value("${notification.websocket.heartbeat.interval:30}") int heartbeatInterval) {
+    private final MultiGauge multiGauge;
+
+    public NotificationWebSocketHandler(ObjectMapper jacksonObjectMapper, MeterRegistry meterRegistry, @Value("${notification.websocket.heartbeat.interval:30}") int heartbeatInterval) {
         this.jacksonObjectMapper = jacksonObjectMapper;
         this.heartbeatInterval = heartbeatInterval;
+        this.multiGauge = MultiGauge.builder(USERS_METER_NAME).description("The current number of connections per user").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.
             // Otherwise, reactorcore buffers some messages (not until the connectable flux had
-            // at least one subscriber. Is there a better way ?
+            // at least one subscriber). Is there a better way ?
             c.subscribe();
         };
     }
@@ -134,6 +146,8 @@ private static Map<String, Object> toResultHeader(Map<String, Object> messageHea
         passHeader(messageHeader, resHeader, HEADER_NEW_NODE);
         passHeader(messageHeader, resHeader, HEADER_MOVED_NODE);
         passHeader(messageHeader, resHeader, HEADER_USER_ID); // to filter the display of error messages in the front end
+        passHeader(messageHeader, resHeader, HEADER_REFERENCE_NODE_UUID);
+        passHeader(messageHeader, resHeader, HEADER_INDEXATION_STATUS);
         passHeader(messageHeader, resHeader, HEADER_PARAMS_NAME);
 
         return resHeader;
@@ -210,10 +224,30 @@ public Mono<Void> handle(WebSocketSession webSocketSession) {
             webSocketSession.getAttributes().put(FILTER_UPDATE_TYPE, filterUpdateType);
         }
 
-        LOGGER.debug("New websocket connection for studyUuid={}, updateType={}", filterStudyUuid, filterUpdateType);
         return webSocketSession
-                .send(notificationFlux(webSocketSession)
-                        .mergeWith(heartbeatFlux(webSocketSession)))
-                .and(receive(webSocketSession));
+                .send(notificationFlux(webSocketSession).mergeWith(heartbeatFlux(webSocketSession)))
+                .and(receive(webSocketSession))
+                .doFirst(() -> updateConnectionMetrics(webSocketSession))
+                .doFinally(s -> updateDisconnectionMetrics(webSocketSession));
+    }
+
+    private void updateConnectionMetrics(WebSocketSession webSocketSession) {
+        var userId = webSocketSession.getHandshakeInfo().getHeaders().getFirst(HEADER_USER_ID);
+        LOGGER.info("New websocket connection id={} for user={} studyUuid={}, updateType={}", webSocketSession.getId(), userId,
+                webSocketSession.getAttributes().get(FILTER_STUDY_UUID), webSocketSession.getAttributes().get(FILTER_UPDATE_TYPE));
+        userConnections.compute(userId, (k, v) -> (v == null) ? 1 : v + 1);
+        updateConnectionMetricsRegistry();
+    }
+
+    private void updateDisconnectionMetrics(WebSocketSession webSocketSession) {
+        var userId = webSocketSession.getHandshakeInfo().getHeaders().getFirst(HEADER_USER_ID);
+        LOGGER.info("Websocket disconnection id={} for user={}", webSocketSession.getId(), userId);
+        userConnections.computeIfPresent(userId, (k, v) -> v > 1 ? v - 1 : null);
+        updateConnectionMetricsRegistry();
+    }
+
+    private void updateConnectionMetricsRegistry() {
+        multiGauge.register(userConnections.entrySet().stream().map(e -> MultiGauge.Row.of(Tags.of(USER_TAG, e.getKey()), e.getValue()))
+                .collect(toList()), true);
     }
 }