dendi-wang/thingsboard
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Use single WebSocket session

Browse Source
This commit is contained in:
ViacheslavKlimov 2023-11-28 18:25:37 +02:00
parent db22c014bd
commit c992ebac9d
10 changed files with 142 additions and 134 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

143
application/src/main/java/org/thingsboard/server/service/ws/DefaultWebSocketService.java
View File

@ -66,7 +66,7 @@ import org.thingsboard.server.service.subscription.TbTimeSeriesSubscription;
import org.thingsboard.server.service.ws.notification.NotificationCommandsHandler; import org.thingsboard.server.service.ws.notification.NotificationCommandsHandler;
import org.thingsboard.server.service.ws.notification.cmd.NotificationCmdsWrapper; import org.thingsboard.server.service.ws.notification.cmd.NotificationCmdsWrapper;
import org.thingsboard.server.service.ws.notification.cmd.WsCmd; import org.thingsboard.server.service.ws.notification.cmd.WsCmd;
import org.thingsboard.server.service.ws.telemetry.cmd.TelemetryPluginCmdsWrapper; import org.thingsboard.server.service.ws.telemetry.cmd.WsCmdsWrapper;
import org.thingsboard.server.service.ws.telemetry.cmd.v1.AttributesSubscriptionCmd; import org.thingsboard.server.service.ws.telemetry.cmd.v1.AttributesSubscriptionCmd;
import org.thingsboard.server.service.ws.telemetry.cmd.v1.GetHistoryCmd; import org.thingsboard.server.service.ws.telemetry.cmd.v1.GetHistoryCmd;
import org.thingsboard.server.service.ws.telemetry.cmd.v1.SubscriptionCmd; import org.thingsboard.server.service.ws.telemetry.cmd.v1.SubscriptionCmd;
@ -149,8 +149,7 @@ public class DefaultWebSocketService implements WebSocketService {
private ScheduledExecutorService pingExecutor; private ScheduledExecutorService pingExecutor;
private String serviceId; private String serviceId;
private List<WsCmdListHandler<TelemetryPluginCmdsWrapper, ?>> telemetryCmdsHandlers; private List<WsCmdsHandler<? extends WsCmd>> cmdsHandlers;
private List<WsCmdHandler<NotificationCmdsWrapper, ? extends WsCmd>> notificationCmdsHandlers;
@PostConstruct @PostConstruct
public void init() { public void init() {
@ -160,25 +159,23 @@ public class DefaultWebSocketService implements WebSocketService {
pingExecutor = Executors.newSingleThreadScheduledExecutor(ThingsBoardThreadFactory.forName("telemetry-web-socket-ping")); pingExecutor = Executors.newSingleThreadScheduledExecutor(ThingsBoardThreadFactory.forName("telemetry-web-socket-ping"));
pingExecutor.scheduleWithFixedDelay(this::sendPing, pingTimeout / NUMBER_OF_PING_ATTEMPTS, pingTimeout / NUMBER_OF_PING_ATTEMPTS, TimeUnit.MILLISECONDS); pingExecutor.scheduleWithFixedDelay(this::sendPing, pingTimeout / NUMBER_OF_PING_ATTEMPTS, pingTimeout / NUMBER_OF_PING_ATTEMPTS, TimeUnit.MILLISECONDS);
telemetryCmdsHandlers = List.of( cmdsHandlers = List.of(
newCmdsHandler(TelemetryPluginCmdsWrapper::getAttrSubCmds, this::handleWsAttributesSubscriptionCmd), newCmdsHandler(WsCmdsWrapper::getAttrSubCmds, this::handleWsAttributesSubscriptionCmd),
newCmdsHandler(TelemetryPluginCmdsWrapper::getTsSubCmds, this::handleWsTimeseriesSubscriptionCmd), newCmdsHandler(WsCmdsWrapper::getTsSubCmds, this::handleWsTimeseriesSubscriptionCmd),
newCmdsHandler(TelemetryPluginCmdsWrapper::getHistoryCmds, this::handleWsHistoryCmd), newCmdsHandler(WsCmdsWrapper::getHistoryCmds, this::handleWsHistoryCmd),
newCmdsHandler(TelemetryPluginCmdsWrapper::getEntityDataCmds, this::handleWsEntityDataCmd), newCmdsHandler(WsCmdsWrapper::getEntityDataCmds, this::handleWsEntityDataCmd),
newCmdsHandler(TelemetryPluginCmdsWrapper::getAlarmDataCmds, this::handleWsAlarmDataCmd), newCmdsHandler(WsCmdsWrapper::getAlarmDataCmds, this::handleWsAlarmDataCmd),
newCmdsHandler(TelemetryPluginCmdsWrapper::getEntityCountCmds, this::handleWsEntityCountCmd), newCmdsHandler(WsCmdsWrapper::getEntityCountCmds, this::handleWsEntityCountCmd),
newCmdsHandler(TelemetryPluginCmdsWrapper::getAlarmCountCmds, this::handleWsAlarmCountCmd), newCmdsHandler(WsCmdsWrapper::getAlarmCountCmds, this::handleWsAlarmCountCmd),
newCmdsHandler(TelemetryPluginCmdsWrapper::getEntityDataUnsubscribeCmds, this::handleWsDataUnsubscribeCmd), newCmdsHandler(WsCmdsWrapper::getEntityDataUnsubscribeCmds, this::handleWsDataUnsubscribeCmd),
newCmdsHandler(TelemetryPluginCmdsWrapper::getAlarmDataUnsubscribeCmds, this::handleWsDataUnsubscribeCmd), newCmdsHandler(WsCmdsWrapper::getAlarmDataUnsubscribeCmds, this::handleWsDataUnsubscribeCmd),
newCmdsHandler(TelemetryPluginCmdsWrapper::getEntityCountUnsubscribeCmds, this::handleWsDataUnsubscribeCmd), newCmdsHandler(WsCmdsWrapper::getEntityCountUnsubscribeCmds, this::handleWsDataUnsubscribeCmd),
newCmdsHandler(TelemetryPluginCmdsWrapper::getAlarmCountUnsubscribeCmds, this::handleWsDataUnsubscribeCmd) newCmdsHandler(WsCmdsWrapper::getAlarmCountUnsubscribeCmds, this::handleWsDataUnsubscribeCmd),
); newCmdHandler(WsCmdsWrapper::getUnreadNotificationsSubCmd, notificationCmdsHandler::handleUnreadNotificationsSubCmd),
notificationCmdsHandlers = List.of( newCmdHandler(WsCmdsWrapper::getUnreadNotificationsCountSubCmd, notificationCmdsHandler::handleUnreadNotificationsCountSubCmd),
newCmdHandler(NotificationCmdsWrapper::getUnreadSubCmd, notificationCmdsHandler::handleUnreadNotificationsSubCmd), newCmdHandler(WsCmdsWrapper::getMarkNotificationAsReadCmd, notificationCmdsHandler::handleMarkAsReadCmd),
newCmdHandler(NotificationCmdsWrapper::getUnreadCountSubCmd, notificationCmdsHandler::handleUnreadNotificationsCountSubCmd), newCmdHandler(WsCmdsWrapper::getMarkAllNotificationsAsReadCmd, notificationCmdsHandler::handleMarkAllAsReadCmd),
newCmdHandler(NotificationCmdsWrapper::getMarkAsReadCmd, notificationCmdsHandler::handleMarkAsReadCmd), newCmdHandler(WsCmdsWrapper::getNotificationsUnsubCmd, notificationCmdsHandler::handleUnsubCmd)
newCmdHandler(NotificationCmdsWrapper::getMarkAllAsReadCmd, notificationCmdsHandler::handleMarkAllAsReadCmd),
newCmdHandler(NotificationCmdsWrapper::getUnsubCmd, notificationCmdsHandler::handleUnsubCmd)
); );
} }
@ -221,8 +218,8 @@ public class DefaultWebSocketService implements WebSocketService {
try { try {
switch (sessionRef.getSessionType()) { switch (sessionRef.getSessionType()) {
case TELEMETRY: case GENERAL:
processTelemetryCmds(sessionRef, msg); processCmds(sessionRef, msg);
break; break;
case NOTIFICATIONS: case NOTIFICATIONS:
processNotificationCmds(sessionRef, msg); processNotificationCmds(sessionRef, msg);
@ -234,25 +231,27 @@ public class DefaultWebSocketService implements WebSocketService {
} }
} }
private void processTelemetryCmds(WebSocketSessionRef sessionRef, String msg) throws JsonProcessingException { private void processCmds(WebSocketSessionRef sessionRef, String msg) throws JsonProcessingException {
TelemetryPluginCmdsWrapper cmdsWrapper = JacksonUtil.fromString(msg, TelemetryPluginCmdsWrapper.class); WsCmdsWrapper cmdsWrapper = JacksonUtil.fromString(msg, WsCmdsWrapper.class);
if (cmdsWrapper == null) { processCmds(sessionRef, cmdsWrapper);
return;
}
for (WsCmdListHandler<TelemetryPluginCmdsWrapper, ?> cmdHandler : telemetryCmdsHandlers) {
List<?> cmds = cmdHandler.extractCmds(cmdsWrapper);
if (cmds != null) {
cmdHandler.handle(sessionRef, cmds);
}
}
} }
private void processNotificationCmds(WebSocketSessionRef sessionRef, String msg) throws IOException { private void processNotificationCmds(WebSocketSessionRef sessionRef, String msg) throws IOException {
NotificationCmdsWrapper cmdsWrapper = JacksonUtil.fromString(msg, NotificationCmdsWrapper.class); NotificationCmdsWrapper cmdsWrapper = JacksonUtil.fromString(msg, NotificationCmdsWrapper.class);
for (WsCmdHandler<NotificationCmdsWrapper, ? extends WsCmd> cmdHandler : notificationCmdsHandlers) { processCmds(sessionRef, cmdsWrapper.toCommonCmdsWrapper());
WsCmd cmd = cmdHandler.extractCmd(cmdsWrapper); }
if (cmd != null) {
private void processCmds(WebSocketSessionRef sessionRef, WsCmdsWrapper cmdsWrapper) {
if (cmdsWrapper == null) {
return;
}
String sessionId = sessionRef.getSessionId(); String sessionId = sessionRef.getSessionId();
for (WsCmdsHandler<? extends WsCmd> cmdHandler : cmdsHandlers) {
List<? extends WsCmd> cmds = cmdHandler.extract(cmdsWrapper);
if (cmds == null) {
continue;
}
for (WsCmd cmd : cmds) {
if (validateSessionMetadata(sessionRef, cmd.getCmdId(), sessionId)) { if (validateSessionMetadata(sessionRef, cmd.getCmdId(), sessionId)) {
try { try {
cmdHandler.handle(sessionRef, cmd); cmdHandler.handle(sessionRef, cmd);
@ -269,8 +268,7 @@ public class DefaultWebSocketService implements WebSocketService {
String sessionId = sessionRef.getSessionId(); String sessionId = sessionRef.getSessionId();
log.debug("[{}] Processing: {}", sessionId, cmd); log.debug("[{}] Processing: {}", sessionId, cmd);
if (validateSessionMetadata(sessionRef, cmd.getCmdId(), sessionId) if (validateSubscriptionCmd(sessionRef, cmd)) {
&& validateSubscriptionCmd(sessionRef, cmd)) {
entityDataSubService.handleCmd(sessionRef, cmd); entityDataSubService.handleCmd(sessionRef, cmd);
} }
} }
@ -279,8 +277,7 @@ public class DefaultWebSocketService implements WebSocketService {
String sessionId = sessionRef.getSessionId(); String sessionId = sessionRef.getSessionId();
log.debug("[{}] Processing: {}", sessionId, cmd); log.debug("[{}] Processing: {}", sessionId, cmd);
if (validateSessionMetadata(sessionRef, cmd.getCmdId(), sessionId) if (validateSubscriptionCmd(sessionRef, cmd)) {
&& validateSubscriptionCmd(sessionRef, cmd)) {
entityDataSubService.handleCmd(sessionRef, cmd); entityDataSubService.handleCmd(sessionRef, cmd);
} }
} }
@ -289,8 +286,7 @@ public class DefaultWebSocketService implements WebSocketService {
String sessionId = sessionRef.getSessionId(); String sessionId = sessionRef.getSessionId();
log.debug("[{}] Processing: {}", sessionId, cmd); log.debug("[{}] Processing: {}", sessionId, cmd);
if (validateSessionMetadata(sessionRef, cmd.getCmdId(), sessionId) if (validateSubscriptionCmd(sessionRef, cmd)) {
&& validateSubscriptionCmd(sessionRef, cmd)) {
entityDataSubService.handleCmd(sessionRef, cmd); entityDataSubService.handleCmd(sessionRef, cmd);
} }
} }
@ -298,18 +294,14 @@ public class DefaultWebSocketService implements WebSocketService {
private void handleWsDataUnsubscribeCmd(WebSocketSessionRef sessionRef, UnsubscribeCmd cmd) { private void handleWsDataUnsubscribeCmd(WebSocketSessionRef sessionRef, UnsubscribeCmd cmd) {
String sessionId = sessionRef.getSessionId(); String sessionId = sessionRef.getSessionId();
log.debug("[{}] Processing: {}", sessionId, cmd); log.debug("[{}] Processing: {}", sessionId, cmd);
if (validateSessionMetadata(sessionRef, cmd.getCmdId(), sessionId)) {
entityDataSubService.cancelSubscription(sessionRef.getSessionId(), cmd); entityDataSubService.cancelSubscription(sessionRef.getSessionId(), cmd);
} }
}
private void handleWsAlarmCountCmd(WebSocketSessionRef sessionRef, AlarmCountCmd cmd) { private void handleWsAlarmCountCmd(WebSocketSessionRef sessionRef, AlarmCountCmd cmd) {
String sessionId = sessionRef.getSessionId(); String sessionId = sessionRef.getSessionId();
log.debug("[{}] Processing: {}", sessionId, cmd); log.debug("[{}] Processing: {}", sessionId, cmd);
if (validateSessionMetadata(sessionRef, cmd.getCmdId(), sessionId) if (validateSubscriptionCmd(sessionRef, cmd)) {
&& validateSubscriptionCmd(sessionRef, cmd)) {
entityDataSubService.handleCmd(sessionRef, cmd); entityDataSubService.handleCmd(sessionRef, cmd);
} }
} }
@ -457,7 +449,6 @@ public class DefaultWebSocketService implements WebSocketService {
String sessionId = sessionRef.getSessionId(); String sessionId = sessionRef.getSessionId();
log.debug("[{}] Processing: {}", sessionId, cmd); log.debug("[{}] Processing: {}", sessionId, cmd);
if (validateSessionMetadata(sessionRef, cmd, sessionId)) {
if (cmd.isUnsubscribe()) { if (cmd.isUnsubscribe()) {
unsubscribe(sessionRef, cmd, sessionId); unsubscribe(sessionRef, cmd, sessionId);
} else if (validateSubscriptionCmd(sessionRef, cmd)) { } else if (validateSubscriptionCmd(sessionRef, cmd)) {
@ -472,7 +463,6 @@ public class DefaultWebSocketService implements WebSocketService {
} }
} }
} }
}
private void handleWsAttributesSubscriptionByKeys(WebSocketSessionRef sessionRef, private void handleWsAttributesSubscriptionByKeys(WebSocketSessionRef sessionRef,
AttributesSubscriptionCmd cmd, String sessionId, EntityId entityId, AttributesSubscriptionCmd cmd, String sessionId, EntityId entityId,
@ -511,7 +501,7 @@ public class DefaultWebSocketService implements WebSocketService {
.build(); .build();
subLock.lock(); subLock.lock();
try{ try {
oldSubService.addSubscription(sub); oldSubService.addSubscription(sub);
sendWsMsg(sessionRef, new TelemetrySubscriptionUpdate(cmd.getCmdId(), attributesData)); sendWsMsg(sessionRef, new TelemetrySubscriptionUpdate(cmd.getCmdId(), attributesData));
} finally { } finally {
@ -543,15 +533,6 @@ public class DefaultWebSocketService implements WebSocketService {
} }
private void handleWsHistoryCmd(WebSocketSessionRef sessionRef, GetHistoryCmd cmd) { private void handleWsHistoryCmd(WebSocketSessionRef sessionRef, GetHistoryCmd cmd) {
String sessionId = sessionRef.getSessionId();
WsSessionMetaData sessionMD = wsSessionsMap.get(sessionId);
if (sessionMD == null) {
log.warn("[{}] Session meta data not found. ", sessionId);
TelemetrySubscriptionUpdate update = new TelemetrySubscriptionUpdate(cmd.getCmdId(), SubscriptionErrorCode.INTERNAL_ERROR,
SESSION_META_DATA_NOT_FOUND);
sendWsMsg(sessionRef, update);
return;
}
if (cmd.getEntityId() == null || cmd.getEntityId().isEmpty() || cmd.getEntityType() == null || cmd.getEntityType().isEmpty()) { if (cmd.getEntityId() == null || cmd.getEntityId().isEmpty() || cmd.getEntityType() == null || cmd.getEntityType().isEmpty()) {
TelemetrySubscriptionUpdate update = new TelemetrySubscriptionUpdate(cmd.getCmdId(), SubscriptionErrorCode.BAD_REQUEST, TelemetrySubscriptionUpdate update = new TelemetrySubscriptionUpdate(cmd.getCmdId(), SubscriptionErrorCode.BAD_REQUEST,
"Device id is empty!"); "Device id is empty!");
@ -662,7 +643,6 @@ public class DefaultWebSocketService implements WebSocketService {
String sessionId = sessionRef.getSessionId(); String sessionId = sessionRef.getSessionId();
log.debug("[{}] Processing: {}", sessionId, cmd); log.debug("[{}] Processing: {}", sessionId, cmd);
if (validateSessionMetadata(sessionRef, cmd, sessionId)) {
if (cmd.isUnsubscribe()) { if (cmd.isUnsubscribe()) {
unsubscribe(sessionRef, cmd, sessionId); unsubscribe(sessionRef, cmd, sessionId);
} else if (validateSubscriptionCmd(sessionRef, cmd)) { } else if (validateSubscriptionCmd(sessionRef, cmd)) {
@ -676,7 +656,6 @@ public class DefaultWebSocketService implements WebSocketService {
} }
} }
} }
}
private void handleWsTimeSeriesSubscriptionByKeys(WebSocketSessionRef sessionRef, private void handleWsTimeSeriesSubscriptionByKeys(WebSocketSessionRef sessionRef,
TimeseriesSubscriptionCmd cmd, String sessionId, EntityId entityId) { TimeseriesSubscriptionCmd cmd, String sessionId, EntityId entityId) {
@ -787,7 +766,7 @@ public class DefaultWebSocketService implements WebSocketService {
.build(); .build();
subLock.lock(); subLock.lock();
try{ try {
oldSubService.addSubscription(sub); oldSubService.addSubscription(sub);
sendWsMsg(sessionRef, new TelemetrySubscriptionUpdate(cmd.getCmdId(), data)); sendWsMsg(sessionRef, new TelemetrySubscriptionUpdate(cmd.getCmdId(), data));
} finally { } finally {
@ -1056,23 +1035,23 @@ public class DefaultWebSocketService implements WebSocketService {
} }
public static <W, C> WsCmdHandler<W, C> newCmdHandler(java.util.function.Function<W, C> cmdExtractor, public static <C extends WsCmd> WsCmdHandler<C> newCmdHandler(java.util.function.Function<WsCmdsWrapper, C> cmdExtractor,
BiConsumer<WebSocketSessionRef, C> handler) { BiConsumer<WebSocketSessionRef, C> handler) {
return new WsCmdHandler<>(cmdExtractor, handler); return new WsCmdHandler<>(cmdExtractor, handler);
} }
public static <W, C> WsCmdListHandler<W, C> newCmdsHandler(java.util.function.Function<W, List<C>> cmdsExtractor, public static <C extends WsCmd> WsCmdsHandler<C> newCmdsHandler(java.util.function.Function<WsCmdsWrapper, List<C>> cmdsExtractor,
BiConsumer<WebSocketSessionRef, C> handler) { BiConsumer<WebSocketSessionRef, C> handler) {
return new WsCmdListHandler<>(cmdsExtractor, handler); return new WsCmdsHandler<>(cmdsExtractor, handler);
} }
@RequiredArgsConstructor @RequiredArgsConstructor
public static class WsCmdHandler<W, C> { public static class WsCmdsHandler<C extends WsCmd> {
private final java.util.function.Function<W, C> cmdExtractor; private final java.util.function.Function<WsCmdsWrapper, List<C>> cmdsExtractor;
private final BiConsumer<WebSocketSessionRef, C> handler; protected final BiConsumer<WebSocketSessionRef, C> handler;
public C extractCmd(W cmdsWrapper) { public List<C> extract(WsCmdsWrapper cmdsWrapper) {
return cmdExtractor.apply(cmdsWrapper); return cmdsExtractor.apply(cmdsWrapper);
} }
@SuppressWarnings("unchecked") @SuppressWarnings("unchecked")
@ -1081,20 +1060,12 @@ public class DefaultWebSocketService implements WebSocketService {
} }
} }
@RequiredArgsConstructor public static class WsCmdHandler<C extends WsCmd> extends WsCmdsHandler<C> {
public static class WsCmdListHandler<W, C> { public WsCmdHandler(java.util.function.Function<WsCmdsWrapper, C> cmdExtractor, BiConsumer<WebSocketSessionRef, C> handler) {
private final java.util.function.Function<W, List<C>> cmdsExtractor; super(cmdsWrapper -> {
private final BiConsumer<WebSocketSessionRef, C> handler; C cmd = cmdExtractor.apply(cmdsWrapper);
return cmd != null ? List.of(cmd) : null;
public List<C> extractCmds(W cmdsWrapper) { }, handler);
return cmdsExtractor.apply(cmdsWrapper);
}
@SuppressWarnings("unchecked")
public void handle(WebSocketSessionRef sessionRef, List<?> cmds) {
cmds.forEach(cmd -> {
handler.accept(sessionRef, (C) cmd);
});
} }
} }

4
application/src/main/java/org/thingsboard/server/service/ws/WebSocketSessionType.java
View File

@ -24,8 +24,8 @@ import java.util.Optional;
@RequiredArgsConstructor @RequiredArgsConstructor
@Getter @Getter
public enum WebSocketSessionType { public enum WebSocketSessionType {
TELEMETRY("telemetry"), GENERAL("telemetry"),
NOTIFICATIONS("notifications"); NOTIFICATIONS("notifications"); // deprecated
private final String name; private final String name;

17
application/src/main/java/org/thingsboard/server/service/ws/notification/cmd/NotificationCmdsWrapper.java
View File

@ -15,9 +15,15 @@
*/ */
package org.thingsboard.server.service.ws.notification.cmd; package org.thingsboard.server.service.ws.notification.cmd;
import com.fasterxml.jackson.annotation.JsonIgnore;
import lombok.Data; import lombok.Data;
import org.thingsboard.server.service.ws.telemetry.cmd.WsCmdsWrapper;
/**
* @deprecated Use {@link org.thingsboard.server.service.ws.telemetry.cmd.WsCmdsWrapper}. This class is left for backward compatibility
* */
@Data @Data
@Deprecated
public class NotificationCmdsWrapper { public class NotificationCmdsWrapper {
private NotificationsCountSubCmd unreadCountSubCmd; private NotificationsCountSubCmd unreadCountSubCmd;
@ -30,4 +36,15 @@ public class NotificationCmdsWrapper {
private NotificationsUnsubCmd unsubCmd; private NotificationsUnsubCmd unsubCmd;
@JsonIgnore
public WsCmdsWrapper toCommonCmdsWrapper() {
WsCmdsWrapper wrapper = new WsCmdsWrapper();
wrapper.setUnreadNotificationsCountSubCmd(unreadCountSubCmd);
wrapper.setUnreadNotificationsSubCmd(unreadSubCmd);
wrapper.setMarkNotificationAsReadCmd(markAsReadCmd);
wrapper.setMarkAllNotificationsAsReadCmd(markAllAsReadCmd);
wrapper.setNotificationsUnsubCmd(unsubCmd);
return wrapper;
}
} }

17
application/src/main/java/org/thingsboard/server/service/ws/telemetry/cmd/TelemetryPluginCmdsWrapper.java → application/src/main/java/org/thingsboard/server/service/ws/telemetry/cmd/WsCmdsWrapper.java
View File

@ -16,6 +16,11 @@
package org.thingsboard.server.service.ws.telemetry.cmd; package org.thingsboard.server.service.ws.telemetry.cmd;
import lombok.Data; import lombok.Data;
import org.thingsboard.server.service.ws.notification.cmd.MarkAllNotificationsAsReadCmd;
import org.thingsboard.server.service.ws.notification.cmd.MarkNotificationsAsReadCmd;
import org.thingsboard.server.service.ws.notification.cmd.NotificationsCountSubCmd;
import org.thingsboard.server.service.ws.notification.cmd.NotificationsSubCmd;
import org.thingsboard.server.service.ws.notification.cmd.NotificationsUnsubCmd;
import org.thingsboard.server.service.ws.telemetry.cmd.v1.AttributesSubscriptionCmd; import org.thingsboard.server.service.ws.telemetry.cmd.v1.AttributesSubscriptionCmd;
import org.thingsboard.server.service.ws.telemetry.cmd.v1.GetHistoryCmd; import org.thingsboard.server.service.ws.telemetry.cmd.v1.GetHistoryCmd;
import org.thingsboard.server.service.ws.telemetry.cmd.v1.TimeseriesSubscriptionCmd; import org.thingsboard.server.service.ws.telemetry.cmd.v1.TimeseriesSubscriptionCmd;
@ -34,7 +39,7 @@ import java.util.List;
* @author Andrew Shvayka * @author Andrew Shvayka
*/ */
@Data @Data
public class TelemetryPluginCmdsWrapper { public class WsCmdsWrapper {
private List<AttributesSubscriptionCmd> attrSubCmds; private List<AttributesSubscriptionCmd> attrSubCmds;
@ -58,4 +63,14 @@ public class TelemetryPluginCmdsWrapper {
private List<AlarmCountUnsubscribeCmd> alarmCountUnsubscribeCmds; private List<AlarmCountUnsubscribeCmd> alarmCountUnsubscribeCmds;
private NotificationsCountSubCmd unreadNotificationsCountSubCmd;
private NotificationsSubCmd unreadNotificationsSubCmd;
private MarkNotificationsAsReadCmd markNotificationAsReadCmd;
private MarkAllNotificationsAsReadCmd markAllNotificationsAsReadCmd;
private NotificationsUnsubCmd notificationsUnsubCmd;
} }

4
application/src/main/java/org/thingsboard/server/service/ws/telemetry/cmd/v1/TelemetryPluginCmd.java
View File

@ -15,10 +15,12 @@
*/ */
package org.thingsboard.server.service.ws.telemetry.cmd.v1; package org.thingsboard.server.service.ws.telemetry.cmd.v1;
import org.thingsboard.server.service.ws.notification.cmd.WsCmd;
/** /**
* @author Andrew Shvayka * @author Andrew Shvayka
*/ */
public interface TelemetryPluginCmd { public interface TelemetryPluginCmd extends WsCmd {
int getCmdId(); int getCmdId();

3
application/src/main/java/org/thingsboard/server/service/ws/telemetry/cmd/v2/DataCmd.java
View File

@ -17,9 +17,10 @@ package org.thingsboard.server.service.ws.telemetry.cmd.v2;
import lombok.Data; import lombok.Data;
import lombok.Getter; import lombok.Getter;
import org.thingsboard.server.service.ws.notification.cmd.WsCmd;
@Data @Data
public class DataCmd { public class DataCmd implements WsCmd {
@Getter @Getter
private final int cmdId; private final int cmdId;

4
application/src/main/java/org/thingsboard/server/service/ws/telemetry/cmd/v2/UnsubscribeCmd.java
View File

@ -15,7 +15,9 @@
*/ */
package org.thingsboard.server.service.ws.telemetry.cmd.v2; package org.thingsboard.server.service.ws.telemetry.cmd.v2;
public interface UnsubscribeCmd { import org.thingsboard.server.service.ws.notification.cmd.WsCmd;
public interface UnsubscribeCmd extends WsCmd {
int getCmdId(); int getCmdId();

10
application/src/test/java/org/thingsboard/server/controller/TbTestWebSocketClient.java
View File

@ -28,7 +28,7 @@ import org.thingsboard.server.common.data.query.EntityDataPageLink;
import org.thingsboard.server.common.data.query.EntityDataQuery; import org.thingsboard.server.common.data.query.EntityDataQuery;
import org.thingsboard.server.common.data.query.EntityFilter; import org.thingsboard.server.common.data.query.EntityFilter;
import org.thingsboard.server.common.data.query.EntityKey; import org.thingsboard.server.common.data.query.EntityKey;
import org.thingsboard.server.service.ws.telemetry.cmd.TelemetryPluginCmdsWrapper; import org.thingsboard.server.service.ws.telemetry.cmd.WsCmdsWrapper;
import org.thingsboard.server.service.ws.telemetry.cmd.v1.AttributesSubscriptionCmd; import org.thingsboard.server.service.ws.telemetry.cmd.v1.AttributesSubscriptionCmd;
import org.thingsboard.server.service.ws.telemetry.cmd.v2.AlarmCountCmd; import org.thingsboard.server.service.ws.telemetry.cmd.v2.AlarmCountCmd;
import org.thingsboard.server.service.ws.telemetry.cmd.v2.AlarmCountUpdate; import org.thingsboard.server.service.ws.telemetry.cmd.v2.AlarmCountUpdate;
@ -106,19 +106,19 @@ public class TbTestWebSocketClient extends WebSocketClient {
} }
public void send(EntityDataCmd cmd) throws NotYetConnectedException { public void send(EntityDataCmd cmd) throws NotYetConnectedException {
TelemetryPluginCmdsWrapper wrapper = new TelemetryPluginCmdsWrapper(); WsCmdsWrapper wrapper = new WsCmdsWrapper();
wrapper.setEntityDataCmds(Collections.singletonList(cmd)); wrapper.setEntityDataCmds(Collections.singletonList(cmd));
this.send(JacksonUtil.toString(wrapper)); this.send(JacksonUtil.toString(wrapper));
} }
public void send(EntityCountCmd cmd) throws NotYetConnectedException { public void send(EntityCountCmd cmd) throws NotYetConnectedException {
TelemetryPluginCmdsWrapper wrapper = new TelemetryPluginCmdsWrapper(); WsCmdsWrapper wrapper = new WsCmdsWrapper();
wrapper.setEntityCountCmds(Collections.singletonList(cmd)); wrapper.setEntityCountCmds(Collections.singletonList(cmd));
this.send(JacksonUtil.toString(wrapper)); this.send(JacksonUtil.toString(wrapper));
} }
public void send(AlarmCountCmd cmd) throws NotYetConnectedException { public void send(AlarmCountCmd cmd) throws NotYetConnectedException {
TelemetryPluginCmdsWrapper wrapper = new TelemetryPluginCmdsWrapper(); WsCmdsWrapper wrapper = new WsCmdsWrapper();
wrapper.setAlarmCountCmds(Collections.singletonList(cmd)); wrapper.setAlarmCountCmds(Collections.singletonList(cmd));
this.send(JacksonUtil.toString(wrapper)); this.send(JacksonUtil.toString(wrapper));
} }
@ -240,7 +240,7 @@ public class TbTestWebSocketClient extends WebSocketClient {
cmd.setEntityId(entityId.getId().toString()); cmd.setEntityId(entityId.getId().toString());
cmd.setScope(scope); cmd.setScope(scope);
cmd.setKeys(String.join(",", keys)); cmd.setKeys(String.join(",", keys));
TelemetryPluginCmdsWrapper cmdsWrapper = new TelemetryPluginCmdsWrapper(); WsCmdsWrapper cmdsWrapper = new WsCmdsWrapper();
cmdsWrapper.setAttrSubCmds(List.of(cmd)); cmdsWrapper.setAttrSubCmds(List.of(cmd));
JsonNode msg = JacksonUtil.valueToTree(cmdsWrapper); JsonNode msg = JacksonUtil.valueToTree(cmdsWrapper);
((ObjectNode) msg.get("attrSubCmds").get(0)).remove("type"); ((ObjectNode) msg.get("attrSubCmds").get(0)).remove("type");

22
application/src/test/java/org/thingsboard/server/service/notification/NotificationApiWsClient.java
View File

@ -24,11 +24,11 @@ import org.thingsboard.server.common.data.notification.Notification;
import org.thingsboard.server.controller.TbTestWebSocketClient; import org.thingsboard.server.controller.TbTestWebSocketClient;
import org.thingsboard.server.service.ws.notification.cmd.MarkAllNotificationsAsReadCmd; import org.thingsboard.server.service.ws.notification.cmd.MarkAllNotificationsAsReadCmd;
import org.thingsboard.server.service.ws.notification.cmd.MarkNotificationsAsReadCmd; import org.thingsboard.server.service.ws.notification.cmd.MarkNotificationsAsReadCmd;
import org.thingsboard.server.service.ws.notification.cmd.NotificationCmdsWrapper;
import org.thingsboard.server.service.ws.notification.cmd.NotificationsCountSubCmd; import org.thingsboard.server.service.ws.notification.cmd.NotificationsCountSubCmd;
import org.thingsboard.server.service.ws.notification.cmd.NotificationsSubCmd; import org.thingsboard.server.service.ws.notification.cmd.NotificationsSubCmd;
import org.thingsboard.server.service.ws.notification.cmd.UnreadNotificationsCountUpdate; import org.thingsboard.server.service.ws.notification.cmd.UnreadNotificationsCountUpdate;
import org.thingsboard.server.service.ws.notification.cmd.UnreadNotificationsUpdate; import org.thingsboard.server.service.ws.notification.cmd.UnreadNotificationsUpdate;
import org.thingsboard.server.service.ws.telemetry.cmd.WsCmdsWrapper;
import org.thingsboard.server.service.ws.telemetry.cmd.v2.CmdUpdateType; import org.thingsboard.server.service.ws.telemetry.cmd.v2.CmdUpdateType;
import java.net.URI; import java.net.URI;
@ -50,37 +50,37 @@ public class NotificationApiWsClient extends TbTestWebSocketClient {
private List<Notification> notifications; private List<Notification> notifications;
public NotificationApiWsClient(String wsUrl, String token) throws URISyntaxException { public NotificationApiWsClient(String wsUrl, String token) throws URISyntaxException {
super(new URI(wsUrl + "/api/ws/plugins/notifications?token=" + token)); super(new URI(wsUrl + "/api/ws/plugins/telemetry?token=" + token));
} }
public NotificationApiWsClient subscribeForUnreadNotifications(int limit) { public NotificationApiWsClient subscribeForUnreadNotifications(int limit) {
NotificationCmdsWrapper cmdsWrapper = new NotificationCmdsWrapper(); WsCmdsWrapper cmdsWrapper = new WsCmdsWrapper();
cmdsWrapper.setUnreadSubCmd(new NotificationsSubCmd(1, limit)); cmdsWrapper.setUnreadNotificationsSubCmd(new NotificationsSubCmd(1, limit));
sendCmd(cmdsWrapper); sendCmd(cmdsWrapper);
this.limit = limit; this.limit = limit;
return this; return this;
} }
public NotificationApiWsClient subscribeForUnreadNotificationsCount() { public NotificationApiWsClient subscribeForUnreadNotificationsCount() {
NotificationCmdsWrapper cmdsWrapper = new NotificationCmdsWrapper(); WsCmdsWrapper cmdsWrapper = new WsCmdsWrapper();
cmdsWrapper.setUnreadCountSubCmd(new NotificationsCountSubCmd(2)); cmdsWrapper.setUnreadNotificationsCountSubCmd(new NotificationsCountSubCmd(2));
sendCmd(cmdsWrapper); sendCmd(cmdsWrapper);
return this; return this;
} }
public void markNotificationAsRead(UUID... notifications) { public void markNotificationAsRead(UUID... notifications) {
NotificationCmdsWrapper cmdsWrapper = new NotificationCmdsWrapper(); WsCmdsWrapper cmdsWrapper = new WsCmdsWrapper();
cmdsWrapper.setMarkAsReadCmd(new MarkNotificationsAsReadCmd(newCmdId(), Arrays.asList(notifications))); cmdsWrapper.setMarkNotificationAsReadCmd(new MarkNotificationsAsReadCmd(newCmdId(), Arrays.asList(notifications)));
sendCmd(cmdsWrapper); sendCmd(cmdsWrapper);
} }
public void markAllNotificationsAsRead() { public void markAllNotificationsAsRead() {
NotificationCmdsWrapper cmdsWrapper = new NotificationCmdsWrapper(); WsCmdsWrapper cmdsWrapper = new WsCmdsWrapper();
cmdsWrapper.setMarkAllAsReadCmd(new MarkAllNotificationsAsReadCmd(newCmdId())); cmdsWrapper.setMarkAllNotificationsAsReadCmd(new MarkAllNotificationsAsReadCmd(newCmdId()));
sendCmd(cmdsWrapper); sendCmd(cmdsWrapper);
} }
public void sendCmd(NotificationCmdsWrapper cmdsWrapper) { public void sendCmd(WsCmdsWrapper cmdsWrapper) {
String cmd = JacksonUtil.toString(cmdsWrapper); String cmd = JacksonUtil.toString(cmdsWrapper);
send(cmd); send(cmd);
} }

4
application/src/test/java/org/thingsboard/server/transport/lwm2m/AbstractLwM2MIntegrationTest.java
View File

@ -63,7 +63,7 @@ import org.thingsboard.server.common.data.query.SingleEntityFilter;
import org.thingsboard.server.common.data.security.DeviceCredentials; import org.thingsboard.server.common.data.security.DeviceCredentials;
import org.thingsboard.server.common.data.security.DeviceCredentialsType; import org.thingsboard.server.common.data.security.DeviceCredentialsType;
import org.thingsboard.server.dao.service.DaoSqlTest; import org.thingsboard.server.dao.service.DaoSqlTest;
import org.thingsboard.server.service.ws.telemetry.cmd.TelemetryPluginCmdsWrapper; import org.thingsboard.server.service.ws.telemetry.cmd.WsCmdsWrapper;
import org.thingsboard.server.service.ws.telemetry.cmd.v2.EntityDataCmd; import org.thingsboard.server.service.ws.telemetry.cmd.v2.EntityDataCmd;
import org.thingsboard.server.service.ws.telemetry.cmd.v2.EntityDataUpdate; import org.thingsboard.server.service.ws.telemetry.cmd.v2.EntityDataUpdate;
import org.thingsboard.server.service.ws.telemetry.cmd.v2.LatestValueCmd; import org.thingsboard.server.service.ws.telemetry.cmd.v2.LatestValueCmd;
@ -229,7 +229,7 @@ public abstract class AbstractLwM2MIntegrationTest extends AbstractTransportInte
Collections.emptyList(), Collections.emptyList(), Collections.emptyList()); Collections.emptyList(), Collections.emptyList(), Collections.emptyList());
EntityDataCmd cmd = new EntityDataCmd(1, edq, null, latestCmd, null); EntityDataCmd cmd = new EntityDataCmd(1, edq, null, latestCmd, null);
TelemetryPluginCmdsWrapper wrapper = new TelemetryPluginCmdsWrapper(); WsCmdsWrapper wrapper = new WsCmdsWrapper();
wrapper.setEntityDataCmds(Collections.singletonList(cmd)); wrapper.setEntityDataCmds(Collections.singletonList(cmd));
getWsClient().send(JacksonUtil.toString(wrapper)); getWsClient().send(JacksonUtil.toString(wrapper));