Skip to content

Commit 61f743c

Browse files
AbdelHedhiliAbdelHedhili
authored
refactor filter mechanism (#54)
Signed-off-by: Abdelsalem <[email protected]>
1 parent 17d2d3b commit 61f743c

File tree

5 files changed

+309
-38
lines changed

5 files changed

+309
-38
lines changed

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

Lines changed: 58 additions & 14 deletions
Original file line numberDiff line numberDiff line change
@@ -18,6 +18,9 @@
1818
import com.fasterxml.jackson.core.JsonProcessingException;
1919
import com.fasterxml.jackson.databind.ObjectMapper;
2020

21+
import org.gridsuite.study.notification.server.dto.Filters;
22+
import org.gridsuite.study.notification.server.dto.FiltersToAdd;
23+
import org.gridsuite.study.notification.server.dto.FiltersToRemove;
2124
import org.gridsuite.study.notification.server.dto.NetworkImpactsInfos;
2225
import org.slf4j.Logger;
2326
import org.slf4j.LoggerFactory;
@@ -51,7 +54,9 @@ public class NotificationWebSocketHandler implements WebSocketHandler {
5154
private static final String CATEGORY_BROKER_INPUT = NotificationWebSocketHandler.class.getName() + ".messages.input-broker";
5255
private static final String CATEGORY_WS_OUTPUT = NotificationWebSocketHandler.class.getName() + ".messages.output-websocket";
5356
static final String QUERY_STUDY_UUID = "studyUuid";
57+
static final String FILTER_STUDY_UUID = QUERY_STUDY_UUID;
5458
static final String QUERY_UPDATE_TYPE = "updateType";
59+
static final String FILTER_UPDATE_TYPE = QUERY_UPDATE_TYPE;
5560
static final String HEADER_USER_ID = "userId";
5661
static final String HEADER_STUDY_UUID = "studyUuid";
5762
static final String HEADER_UPDATE_TYPE = "updateType";
@@ -94,18 +99,13 @@ public Consumer<Flux<Message<NetworkImpactsInfos>>> consumeNotification() {
9499
/**
95100
* map from the broker flux to the filtered flux for one websocket client, extracting only relevant fields.
96101
*/
97-
private Flux<WebSocketMessage> notificationFlux(WebSocketSession webSocketSession,
98-
String filterStudyUuid,
99-
String filterUpdateType) {
100-
return flux.transform(f -> {
101-
Flux<Message<NetworkImpactsInfos>> res = f;
102-
if (filterStudyUuid != null) {
103-
res = res.filter(m -> filterStudyUuid.equals(m.getHeaders().get(HEADER_STUDY_UUID)));
104-
}
105-
if (filterUpdateType != null) {
106-
res = res.filter(m -> filterUpdateType.equals(m.getHeaders().get(HEADER_UPDATE_TYPE)));
107-
}
108-
return res;
102+
private Flux<WebSocketMessage> notificationFlux(WebSocketSession webSocketSession) {
103+
return flux.filter(message -> {
104+
String filterStudyUuid = (String) webSocketSession.getAttributes().get(FILTER_STUDY_UUID);
105+
return filterStudyUuid == null || filterStudyUuid.equals(message.getHeaders().get(HEADER_STUDY_UUID));
106+
}).filter(message -> {
107+
String filterUpdateType = (String) webSocketSession.getAttributes().get(FILTER_UPDATE_TYPE);
108+
return filterUpdateType == null || filterUpdateType.equals(message.getHeaders().get(HEADER_UPDATE_TYPE));
109109
}).map(m -> {
110110
try {
111111
return jacksonObjectMapper.writeValueAsString(Map.of(
@@ -151,6 +151,45 @@ private Flux<WebSocketMessage> heartbeatFlux(WebSocketSession webSocketSession)
151151
.pingMessage(dbf -> dbf.wrap((webSocketSession.getId() + "-" + n).getBytes(StandardCharsets.UTF_8))));
152152
}
153153

154+
public Flux<WebSocketMessage> receive(WebSocketSession webSocketSession) {
155+
return webSocketSession.receive()
156+
.doOnNext(webSocketMessage -> {
157+
try {
158+
//if it's not the heartbeat
159+
if (webSocketMessage.getType().equals(WebSocketMessage.Type.TEXT)) {
160+
String wsPayload = webSocketMessage.getPayloadAsText();
161+
LOGGER.debug("Message received : {} by session {}", wsPayload, webSocketSession.getId());
162+
Filters receivedFilters = jacksonObjectMapper.readValue(webSocketMessage.getPayloadAsText(), Filters.class);
163+
handleReceivedFilters(webSocketSession, receivedFilters);
164+
}
165+
} catch (JsonProcessingException e) {
166+
LOGGER.error(e.toString(), e);
167+
}
168+
});
169+
}
170+
171+
private void handleReceivedFilters(WebSocketSession webSocketSession, Filters filters) {
172+
if (filters.getFiltersToRemove() != null) {
173+
FiltersToRemove filtersToRemove = filters.getFiltersToRemove();
174+
if (Boolean.TRUE.equals(filtersToRemove.getRemoveUpdateType())) {
175+
webSocketSession.getAttributes().remove(FILTER_UPDATE_TYPE);
176+
}
177+
if (Boolean.TRUE.equals(filtersToRemove.getRemoveStudyUuid())) {
178+
webSocketSession.getAttributes().remove(FILTER_STUDY_UUID);
179+
}
180+
}
181+
if (filters.getFiltersToAdd() != null) {
182+
FiltersToAdd filtersToAdd = filters.getFiltersToAdd();
183+
//because null is not allowed in ConcurrentHashMap and will cause the websocket to close
184+
if (filtersToAdd.getUpdateType() != null) {
185+
webSocketSession.getAttributes().put(FILTER_UPDATE_TYPE, filtersToAdd.getUpdateType());
186+
}
187+
if (filtersToAdd.getStudyUuid() != null) {
188+
webSocketSession.getAttributes().put(FILTER_STUDY_UUID, filtersToAdd.getStudyUuid());
189+
}
190+
}
191+
}
192+
154193
@Override
155194
public Mono<Void> handle(WebSocketSession webSocketSession) {
156195
var uri = webSocketSession.getHandshakeInfo().getUri();
@@ -159,15 +198,20 @@ public Mono<Void> handle(WebSocketSession webSocketSession) {
159198
if (filterStudyUuid != null) {
160199
try {
161200
filterStudyUuid = URLDecoder.decode(filterStudyUuid, StandardCharsets.UTF_8.toString());
201+
webSocketSession.getAttributes().put(FILTER_STUDY_UUID, filterStudyUuid);
162202
} catch (UnsupportedEncodingException e) {
163203
throw new NotificationServerRuntimeException(e.getMessage());
164204
}
165205
}
166206
String filterUpdateType = parameters.getFirst(QUERY_UPDATE_TYPE);
207+
if (filterUpdateType != null) {
208+
webSocketSession.getAttributes().put(FILTER_UPDATE_TYPE, filterUpdateType);
209+
}
210+
167211
LOGGER.debug("New websocket connection for studyUuid={}, updateType={}", filterStudyUuid, filterUpdateType);
168212
return webSocketSession
169-
.send(notificationFlux(webSocketSession, filterStudyUuid, filterUpdateType)
213+
.send(notificationFlux(webSocketSession)
170214
.mergeWith(heartbeatFlux(webSocketSession)))
171-
.and(webSocketSession.receive());
215+
.and(receive(webSocketSession));
172216
}
173217
}

‎src/main/java/org/gridsuite/study/notification/server/dto/Filters.java‎

Lines changed: 28 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,28 @@
1+
/**
2+
* Copyright (c) 2023, RTE (http://www.rte-france.com)
3+
* This Source Code Form is subject to the terms of the Mozilla Public
4+
* License, v. 2.0. If a copy of the MPL was not distributed with this
5+
* file, You can obtain one at http://mozilla.org/MPL/2.0/.
6+
*/
7+
package org.gridsuite.study.notification.server.dto;
8+
9+
import com.fasterxml.jackson.annotation.JsonInclude;
10+
import lombok.AllArgsConstructor;
11+
import lombok.Getter;
12+
import lombok.NoArgsConstructor;
13+
import lombok.Setter;
14+
15+
/**
16+
* @author Abdelsalem Hedhili <abdelsalem.hedhili at rte-france.com>
17+
*/
18+
@Getter
19+
@Setter
20+
@AllArgsConstructor
21+
@NoArgsConstructor
22+
@JsonInclude(JsonInclude.Include.NON_NULL)
23+
public class Filters {
24+
25+
private FiltersToAdd filtersToAdd;
26+
27+
private FiltersToRemove filtersToRemove;
28+
}

‎src/main/java/org/gridsuite/study/notification/server/dto/FiltersToAdd.java‎

Lines changed: 28 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,28 @@
1+
/**
2+
* Copyright (c) 2023, RTE (http://www.rte-france.com)
3+
* This Source Code Form is subject to the terms of the Mozilla Public
4+
* License, v. 2.0. If a copy of the MPL was not distributed with this
5+
* file, You can obtain one at http://mozilla.org/MPL/2.0/.
6+
*/
7+
package org.gridsuite.study.notification.server.dto;
8+
9+
import com.fasterxml.jackson.annotation.JsonInclude;
10+
import lombok.AllArgsConstructor;
11+
import lombok.Getter;
12+
import lombok.NoArgsConstructor;
13+
import lombok.Setter;
14+
15+
/**
16+
* @author Abdelsalem Hedhili <abdelsalem.hedhili at rte-france.com>
17+
*/
18+
@Getter
19+
@Setter
20+
@AllArgsConstructor
21+
@NoArgsConstructor
22+
@JsonInclude(JsonInclude.Include.NON_NULL)
23+
public class FiltersToAdd {
24+
25+
private String updateType;
26+
27+
private String studyUuid;
28+
}

‎src/main/java/org/gridsuite/study/notification/server/dto/FiltersToRemove.java‎

Lines changed: 28 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,28 @@
1+
/**
2+
* Copyright (c) 2023, RTE (http://www.rte-france.com)
3+
* This Source Code Form is subject to the terms of the Mozilla Public
4+
* License, v. 2.0. If a copy of the MPL was not distributed with this
5+
* file, You can obtain one at http://mozilla.org/MPL/2.0/.
6+
*/
7+
package org.gridsuite.study.notification.server.dto;
8+
9+
import com.fasterxml.jackson.annotation.JsonInclude;
10+
import lombok.AllArgsConstructor;
11+
import lombok.Getter;
12+
import lombok.NoArgsConstructor;
13+
import lombok.Setter;
14+
15+
/**
16+
* @author Abdelsalem Hedhili <abdelsalem.hedhili at rte-france.com>
17+
*/
18+
@Getter
19+
@Setter
20+
@AllArgsConstructor
21+
@NoArgsConstructor
22+
@JsonInclude(JsonInclude.Include.NON_NULL)
23+
public class FiltersToRemove {
24+
25+
private Boolean removeUpdateType;
26+
27+
private Boolean removeStudyUuid;
28+
}

0 commit comments

Comments
 (0)