|
|
|
@ -47,6 +47,7 @@ import org.thingsboard.server.dao.attributes.AttributesService; |
|
|
|
import org.thingsboard.server.dao.tenant.TbTenantProfileCache; |
|
|
|
import org.thingsboard.server.dao.timeseries.TimeseriesService; |
|
|
|
import org.thingsboard.server.dao.util.TenantRateLimitException; |
|
|
|
import org.thingsboard.server.exception.UnauthorizedException; |
|
|
|
import org.thingsboard.server.queue.discovery.TbServiceInfoProvider; |
|
|
|
import org.thingsboard.server.queue.util.TbCoreComponent; |
|
|
|
import org.thingsboard.server.service.security.AccessValidator; |
|
|
|
@ -55,26 +56,35 @@ import org.thingsboard.server.service.security.ValidationResult; |
|
|
|
import org.thingsboard.server.service.security.ValidationResultCode; |
|
|
|
import org.thingsboard.server.service.security.model.UserPrincipal; |
|
|
|
import org.thingsboard.server.service.security.permission.Operation; |
|
|
|
import org.thingsboard.server.service.subscription.SubscriptionErrorCode; |
|
|
|
import org.thingsboard.server.service.subscription.TbAttributeSubscription; |
|
|
|
import org.thingsboard.server.service.subscription.TbAttributeSubscriptionScope; |
|
|
|
import org.thingsboard.server.service.subscription.TbEntityDataSubscriptionService; |
|
|
|
import org.thingsboard.server.service.subscription.TbLocalSubscriptionService; |
|
|
|
import org.thingsboard.server.service.subscription.TbTimeseriesSubscription; |
|
|
|
import org.thingsboard.server.service.telemetry.cmd.TelemetryPluginCmdsWrapper; |
|
|
|
import org.thingsboard.server.service.telemetry.cmd.v1.AttributesSubscriptionCmd; |
|
|
|
import org.thingsboard.server.service.telemetry.cmd.v1.GetHistoryCmd; |
|
|
|
import org.thingsboard.server.service.telemetry.cmd.v1.SubscriptionCmd; |
|
|
|
import org.thingsboard.server.service.telemetry.cmd.v1.TelemetryPluginCmd; |
|
|
|
import org.thingsboard.server.service.telemetry.cmd.v1.TimeseriesSubscriptionCmd; |
|
|
|
import org.thingsboard.server.service.telemetry.cmd.v2.AlarmDataCmd; |
|
|
|
import org.thingsboard.server.service.telemetry.cmd.v2.CmdUpdate; |
|
|
|
import org.thingsboard.server.service.telemetry.cmd.v2.EntityCountCmd; |
|
|
|
import org.thingsboard.server.service.telemetry.cmd.v2.EntityDataCmd; |
|
|
|
import org.thingsboard.server.service.telemetry.cmd.v2.EntityDataUpdate; |
|
|
|
import org.thingsboard.server.service.telemetry.cmd.v2.UnsubscribeCmd; |
|
|
|
import org.thingsboard.server.service.telemetry.exception.UnauthorizedException; |
|
|
|
import org.thingsboard.server.service.telemetry.sub.SubscriptionErrorCode; |
|
|
|
import org.thingsboard.server.service.telemetry.sub.TelemetrySubscriptionUpdate; |
|
|
|
import org.thingsboard.server.service.ws.SessionEvent; |
|
|
|
import org.thingsboard.server.service.ws.WebSocketMsgEndpoint; |
|
|
|
import org.thingsboard.server.service.ws.WebSocketSessionRef; |
|
|
|
import org.thingsboard.server.service.ws.WsSessionMetaData; |
|
|
|
import org.thingsboard.server.service.ws.notification.cmd.NotificationCmdsWrapper; |
|
|
|
import org.thingsboard.server.service.ws.notification.cmd.MarkNotificationAsReadCmd; |
|
|
|
import org.thingsboard.server.service.ws.notification.cmd.NotificationsSubCmd; |
|
|
|
import org.thingsboard.server.service.ws.notification.cmd.NotificationsUnsubCmd; |
|
|
|
import org.thingsboard.server.service.ws.notification.sub.DefaultNotificationsSubscriptionService; |
|
|
|
import org.thingsboard.server.service.ws.telemetry.WebSocketService; |
|
|
|
import org.thingsboard.server.service.ws.telemetry.cmd.TelemetryPluginCmdsWrapper; |
|
|
|
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.SubscriptionCmd; |
|
|
|
import org.thingsboard.server.service.ws.telemetry.cmd.v1.TelemetryPluginCmd; |
|
|
|
import org.thingsboard.server.service.ws.telemetry.cmd.v1.TimeseriesSubscriptionCmd; |
|
|
|
import org.thingsboard.server.service.ws.telemetry.cmd.v2.AlarmDataCmd; |
|
|
|
import org.thingsboard.server.service.ws.telemetry.cmd.v2.CmdUpdate; |
|
|
|
import org.thingsboard.server.service.ws.telemetry.cmd.v2.EntityCountCmd; |
|
|
|
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.UnsubscribeCmd; |
|
|
|
import org.thingsboard.server.service.ws.telemetry.sub.TelemetrySubscriptionUpdate; |
|
|
|
|
|
|
|
import javax.annotation.Nullable; |
|
|
|
import javax.annotation.PostConstruct; |
|
|
|
@ -103,7 +113,7 @@ import java.util.stream.Collectors; |
|
|
|
@Service |
|
|
|
@TbCoreComponent |
|
|
|
@Slf4j |
|
|
|
public class DefaultTelemetryWebSocketService implements TelemetryWebSocketService { |
|
|
|
public class DefaultWebSocketService implements WebSocketService { |
|
|
|
|
|
|
|
public static final int NUMBER_OF_PING_ATTEMPTS = 3; |
|
|
|
|
|
|
|
@ -126,7 +136,10 @@ public class DefaultTelemetryWebSocketService implements TelemetryWebSocketServi |
|
|
|
private TbEntityDataSubscriptionService entityDataSubService; |
|
|
|
|
|
|
|
@Autowired |
|
|
|
private TelemetryWebSocketMsgEndpoint msgEndpoint; |
|
|
|
private DefaultNotificationsSubscriptionService notificationsSubService; |
|
|
|
|
|
|
|
@Autowired |
|
|
|
private WebSocketMsgEndpoint msgEndpoint; |
|
|
|
|
|
|
|
@Autowired |
|
|
|
private AccessValidator accessValidator; |
|
|
|
@ -177,7 +190,7 @@ public class DefaultTelemetryWebSocketService implements TelemetryWebSocketServi |
|
|
|
} |
|
|
|
|
|
|
|
@Override |
|
|
|
public void handleWebSocketSessionEvent(TelemetryWebSocketSessionRef sessionRef, SessionEvent event) { |
|
|
|
public void handleWebSocketSessionEvent(WebSocketSessionRef sessionRef, SessionEvent event) { |
|
|
|
String sessionId = sessionRef.getSessionId(); |
|
|
|
log.debug(PROCESSING_MSG, sessionId, event); |
|
|
|
switch (event.getEventType()) { |
|
|
|
@ -197,49 +210,19 @@ public class DefaultTelemetryWebSocketService implements TelemetryWebSocketServi |
|
|
|
} |
|
|
|
|
|
|
|
@Override |
|
|
|
public void handleWebSocketMsg(TelemetryWebSocketSessionRef sessionRef, String msg) { |
|
|
|
public void handleWebSocketMsg(WebSocketSessionRef sessionRef, String msg) { |
|
|
|
if (log.isTraceEnabled()) { |
|
|
|
log.trace("[{}] Processing: {}", sessionRef.getSessionId(), msg); |
|
|
|
} |
|
|
|
|
|
|
|
try { |
|
|
|
TelemetryPluginCmdsWrapper cmdsWrapper = jsonMapper.readValue(msg, TelemetryPluginCmdsWrapper.class); |
|
|
|
if (cmdsWrapper != null) { |
|
|
|
if (cmdsWrapper.getAttrSubCmds() != null) { |
|
|
|
cmdsWrapper.getAttrSubCmds().forEach(cmd -> { |
|
|
|
if (processSubscription(sessionRef, cmd)) { |
|
|
|
handleWsAttributesSubscriptionCmd(sessionRef, cmd); |
|
|
|
} |
|
|
|
}); |
|
|
|
} |
|
|
|
if (cmdsWrapper.getTsSubCmds() != null) { |
|
|
|
cmdsWrapper.getTsSubCmds().forEach(cmd -> { |
|
|
|
if (processSubscription(sessionRef, cmd)) { |
|
|
|
handleWsTimeseriesSubscriptionCmd(sessionRef, cmd); |
|
|
|
} |
|
|
|
}); |
|
|
|
} |
|
|
|
if (cmdsWrapper.getHistoryCmds() != null) { |
|
|
|
cmdsWrapper.getHistoryCmds().forEach(cmd -> handleWsHistoryCmd(sessionRef, cmd)); |
|
|
|
} |
|
|
|
if (cmdsWrapper.getEntityDataCmds() != null) { |
|
|
|
cmdsWrapper.getEntityDataCmds().forEach(cmd -> handleWsEntityDataCmd(sessionRef, cmd)); |
|
|
|
} |
|
|
|
if (cmdsWrapper.getAlarmDataCmds() != null) { |
|
|
|
cmdsWrapper.getAlarmDataCmds().forEach(cmd -> handleWsAlarmDataCmd(sessionRef, cmd)); |
|
|
|
} |
|
|
|
if (cmdsWrapper.getEntityCountCmds() != null) { |
|
|
|
cmdsWrapper.getEntityCountCmds().forEach(cmd -> handleWsEntityCountCmd(sessionRef, cmd)); |
|
|
|
} |
|
|
|
if (cmdsWrapper.getEntityDataUnsubscribeCmds() != null) { |
|
|
|
cmdsWrapper.getEntityDataUnsubscribeCmds().forEach(cmd -> handleWsDataUnsubscribeCmd(sessionRef, cmd)); |
|
|
|
} |
|
|
|
if (cmdsWrapper.getAlarmDataUnsubscribeCmds() != null) { |
|
|
|
cmdsWrapper.getAlarmDataUnsubscribeCmds().forEach(cmd -> handleWsDataUnsubscribeCmd(sessionRef, cmd)); |
|
|
|
} |
|
|
|
if (cmdsWrapper.getEntityCountUnsubscribeCmds() != null) { |
|
|
|
cmdsWrapper.getEntityCountUnsubscribeCmds().forEach(cmd -> handleWsDataUnsubscribeCmd(sessionRef, cmd)); |
|
|
|
} |
|
|
|
switch (sessionRef.getSessionType()) { |
|
|
|
case TELEMETRY: |
|
|
|
processTelemetryCmds(sessionRef, msg); |
|
|
|
break; |
|
|
|
case NOTIFICATIONS: |
|
|
|
processNotificationCmds(sessionRef, msg); |
|
|
|
break; |
|
|
|
} |
|
|
|
} catch (IOException e) { |
|
|
|
log.warn("Failed to decode subscription cmd: {}", e.getMessage(), e); |
|
|
|
@ -247,7 +230,81 @@ public class DefaultTelemetryWebSocketService implements TelemetryWebSocketServi |
|
|
|
} |
|
|
|
} |
|
|
|
|
|
|
|
private void handleWsEntityDataCmd(TelemetryWebSocketSessionRef sessionRef, EntityDataCmd cmd) { |
|
|
|
private void processTelemetryCmds(WebSocketSessionRef sessionRef, String msg) throws JsonProcessingException { |
|
|
|
TelemetryPluginCmdsWrapper cmdsWrapper = jsonMapper.readValue(msg, TelemetryPluginCmdsWrapper.class); |
|
|
|
if (cmdsWrapper == null) { |
|
|
|
return; |
|
|
|
} |
|
|
|
if (cmdsWrapper.getAttrSubCmds() != null) { |
|
|
|
cmdsWrapper.getAttrSubCmds().forEach(cmd -> { |
|
|
|
if (processSubscription(sessionRef, cmd)) { |
|
|
|
handleWsAttributesSubscriptionCmd(sessionRef, cmd); |
|
|
|
} |
|
|
|
}); |
|
|
|
} |
|
|
|
if (cmdsWrapper.getTsSubCmds() != null) { |
|
|
|
cmdsWrapper.getTsSubCmds().forEach(cmd -> { |
|
|
|
if (processSubscription(sessionRef, cmd)) { |
|
|
|
handleWsTimeseriesSubscriptionCmd(sessionRef, cmd); |
|
|
|
} |
|
|
|
}); |
|
|
|
} |
|
|
|
if (cmdsWrapper.getHistoryCmds() != null) { |
|
|
|
cmdsWrapper.getHistoryCmds().forEach(cmd -> handleWsHistoryCmd(sessionRef, cmd)); |
|
|
|
} |
|
|
|
if (cmdsWrapper.getEntityDataCmds() != null) { |
|
|
|
cmdsWrapper.getEntityDataCmds().forEach(cmd -> handleWsEntityDataCmd(sessionRef, cmd)); |
|
|
|
} |
|
|
|
if (cmdsWrapper.getAlarmDataCmds() != null) { |
|
|
|
cmdsWrapper.getAlarmDataCmds().forEach(cmd -> handleWsAlarmDataCmd(sessionRef, cmd)); |
|
|
|
} |
|
|
|
if (cmdsWrapper.getEntityCountCmds() != null) { |
|
|
|
cmdsWrapper.getEntityCountCmds().forEach(cmd -> handleWsEntityCountCmd(sessionRef, cmd)); |
|
|
|
} |
|
|
|
if (cmdsWrapper.getEntityDataUnsubscribeCmds() != null) { |
|
|
|
cmdsWrapper.getEntityDataUnsubscribeCmds().forEach(cmd -> handleWsDataUnsubscribeCmd(sessionRef, cmd)); |
|
|
|
} |
|
|
|
if (cmdsWrapper.getAlarmDataUnsubscribeCmds() != null) { |
|
|
|
cmdsWrapper.getAlarmDataUnsubscribeCmds().forEach(cmd -> handleWsDataUnsubscribeCmd(sessionRef, cmd)); |
|
|
|
} |
|
|
|
if (cmdsWrapper.getEntityCountUnsubscribeCmds() != null) { |
|
|
|
cmdsWrapper.getEntityCountUnsubscribeCmds().forEach(cmd -> handleWsDataUnsubscribeCmd(sessionRef, cmd)); |
|
|
|
} |
|
|
|
} |
|
|
|
|
|
|
|
private void processNotificationCmds(WebSocketSessionRef sessionRef, String msg) throws IOException { |
|
|
|
NotificationCmdsWrapper cmdsWrapper = jsonMapper.readValue(msg, NotificationCmdsWrapper.class); |
|
|
|
if (cmdsWrapper.getUnreadSubCmd() != null) { |
|
|
|
handleUnreadNotificationsSubCmd(sessionRef, cmdsWrapper.getUnreadSubCmd()); |
|
|
|
} else if (cmdsWrapper.getUnreadUnsubCmd() != null) { |
|
|
|
handleUnreadNotificationsUnsubCmd(sessionRef, cmdsWrapper.getUnreadUnsubCmd()); |
|
|
|
} else if (cmdsWrapper.getMarkAsReadCmd() != null) { |
|
|
|
handleMarkNotificationAsReadCmd(sessionRef, cmdsWrapper.getMarkAsReadCmd()); |
|
|
|
} |
|
|
|
} |
|
|
|
|
|
|
|
private void handleUnreadNotificationsSubCmd(WebSocketSessionRef sessionRef, NotificationsSubCmd cmd) { |
|
|
|
String sessionId = sessionRef.getSessionId(); |
|
|
|
if (validateSessionMetadata(sessionRef, cmd.getCmdId(), sessionId)) { |
|
|
|
notificationsSubService.handleUnreadNotificationsSubCmd(sessionRef, cmd); |
|
|
|
} |
|
|
|
} |
|
|
|
|
|
|
|
private void handleUnreadNotificationsUnsubCmd(WebSocketSessionRef sessionRef, NotificationsUnsubCmd cmd) { |
|
|
|
String sessionId = sessionRef.getSessionId(); |
|
|
|
if (validateSessionMetadata(sessionRef, cmd.getCmdId(), sessionId)) { |
|
|
|
notificationsSubService.handleUnsubCmd(sessionRef, cmd); |
|
|
|
} |
|
|
|
} |
|
|
|
|
|
|
|
private void handleMarkNotificationAsReadCmd(WebSocketSessionRef sessionRef, MarkNotificationAsReadCmd cmd) { |
|
|
|
String sessionId = sessionRef.getSessionId(); |
|
|
|
if (validateSessionMetadata(sessionRef, cmd.getCmdId(), sessionId)) { |
|
|
|
notificationsSubService.handleMarkAsReadCmd(sessionRef, cmd); |
|
|
|
} |
|
|
|
} |
|
|
|
|
|
|
|
private void handleWsEntityDataCmd(WebSocketSessionRef sessionRef, EntityDataCmd cmd) { |
|
|
|
String sessionId = sessionRef.getSessionId(); |
|
|
|
log.debug("[{}] Processing: {}", sessionId, cmd); |
|
|
|
|
|
|
|
@ -257,7 +314,7 @@ public class DefaultTelemetryWebSocketService implements TelemetryWebSocketServi |
|
|
|
} |
|
|
|
} |
|
|
|
|
|
|
|
private void handleWsEntityCountCmd(TelemetryWebSocketSessionRef sessionRef, EntityCountCmd cmd) { |
|
|
|
private void handleWsEntityCountCmd(WebSocketSessionRef sessionRef, EntityCountCmd cmd) { |
|
|
|
String sessionId = sessionRef.getSessionId(); |
|
|
|
log.debug("[{}] Processing: {}", sessionId, cmd); |
|
|
|
|
|
|
|
@ -267,7 +324,7 @@ public class DefaultTelemetryWebSocketService implements TelemetryWebSocketServi |
|
|
|
} |
|
|
|
} |
|
|
|
|
|
|
|
private void handleWsAlarmDataCmd(TelemetryWebSocketSessionRef sessionRef, AlarmDataCmd cmd) { |
|
|
|
private void handleWsAlarmDataCmd(WebSocketSessionRef sessionRef, AlarmDataCmd cmd) { |
|
|
|
String sessionId = sessionRef.getSessionId(); |
|
|
|
log.debug("[{}] Processing: {}", sessionId, cmd); |
|
|
|
|
|
|
|
@ -277,7 +334,7 @@ public class DefaultTelemetryWebSocketService implements TelemetryWebSocketServi |
|
|
|
} |
|
|
|
} |
|
|
|
|
|
|
|
private void handleWsDataUnsubscribeCmd(TelemetryWebSocketSessionRef sessionRef, UnsubscribeCmd cmd) { |
|
|
|
private void handleWsDataUnsubscribeCmd(WebSocketSessionRef sessionRef, UnsubscribeCmd cmd) { |
|
|
|
String sessionId = sessionRef.getSessionId(); |
|
|
|
log.debug("[{}] Processing: {}", sessionId, cmd); |
|
|
|
|
|
|
|
@ -315,7 +372,7 @@ public class DefaultTelemetryWebSocketService implements TelemetryWebSocketServi |
|
|
|
} |
|
|
|
} |
|
|
|
|
|
|
|
private void processSessionClose(TelemetryWebSocketSessionRef sessionRef) { |
|
|
|
private void processSessionClose(WebSocketSessionRef sessionRef) { |
|
|
|
var tenantProfileConfiguration = tenantProfileCache.get(sessionRef.getSecurityCtx().getTenantId()).getDefaultProfileConfiguration(); |
|
|
|
if (tenantProfileConfiguration != null) { |
|
|
|
String sessionId = "[" + sessionRef.getSessionId() + "]"; |
|
|
|
@ -349,7 +406,7 @@ public class DefaultTelemetryWebSocketService implements TelemetryWebSocketServi |
|
|
|
} |
|
|
|
} |
|
|
|
|
|
|
|
private boolean processSubscription(TelemetryWebSocketSessionRef sessionRef, SubscriptionCmd cmd) { |
|
|
|
private boolean processSubscription(WebSocketSessionRef sessionRef, SubscriptionCmd cmd) { |
|
|
|
var tenantProfileConfiguration = (DefaultTenantProfileConfiguration) tenantProfileCache.get(sessionRef.getSecurityCtx().getTenantId()).getDefaultProfileConfiguration(); |
|
|
|
|
|
|
|
String subId = "[" + sessionRef.getSessionId() + "]:[" + cmd.getCmdId() + "]"; |
|
|
|
@ -420,7 +477,7 @@ public class DefaultTelemetryWebSocketService implements TelemetryWebSocketServi |
|
|
|
return true; |
|
|
|
} |
|
|
|
|
|
|
|
private void handleWsAttributesSubscriptionCmd(TelemetryWebSocketSessionRef sessionRef, AttributesSubscriptionCmd cmd) { |
|
|
|
private void handleWsAttributesSubscriptionCmd(WebSocketSessionRef sessionRef, AttributesSubscriptionCmd cmd) { |
|
|
|
String sessionId = sessionRef.getSessionId(); |
|
|
|
log.debug("[{}] Processing: {}", sessionId, cmd); |
|
|
|
|
|
|
|
@ -441,7 +498,7 @@ public class DefaultTelemetryWebSocketService implements TelemetryWebSocketServi |
|
|
|
} |
|
|
|
} |
|
|
|
|
|
|
|
private void handleWsAttributesSubscriptionByKeys(TelemetryWebSocketSessionRef sessionRef, |
|
|
|
private void handleWsAttributesSubscriptionByKeys(WebSocketSessionRef sessionRef, |
|
|
|
AttributesSubscriptionCmd cmd, String sessionId, EntityId entityId, |
|
|
|
List<String> keys) { |
|
|
|
FutureCallback<List<AttributeKvEntry>> callback = new FutureCallback<>() { |
|
|
|
@ -465,7 +522,7 @@ public class DefaultTelemetryWebSocketService implements TelemetryWebSocketServi |
|
|
|
.allKeys(false) |
|
|
|
.keyStates(subState) |
|
|
|
.scope(scope) |
|
|
|
.updateConsumer(DefaultTelemetryWebSocketService.this::sendWsMsg) |
|
|
|
.updateProcessor((subscription, update) -> sendWsMsg(subscription.getSessionId(), update)) |
|
|
|
.build(); |
|
|
|
oldSubService.addSubscription(sub); |
|
|
|
} |
|
|
|
@ -492,7 +549,7 @@ public class DefaultTelemetryWebSocketService implements TelemetryWebSocketServi |
|
|
|
} |
|
|
|
} |
|
|
|
|
|
|
|
private void handleWsHistoryCmd(TelemetryWebSocketSessionRef sessionRef, GetHistoryCmd cmd) { |
|
|
|
private void handleWsHistoryCmd(WebSocketSessionRef sessionRef, GetHistoryCmd cmd) { |
|
|
|
String sessionId = sessionRef.getSessionId(); |
|
|
|
WsSessionMetaData sessionMD = wsSessionsMap.get(sessionId); |
|
|
|
if (sessionMD == null) { |
|
|
|
@ -542,7 +599,7 @@ public class DefaultTelemetryWebSocketService implements TelemetryWebSocketServi |
|
|
|
on(r -> Futures.addCallback(tsService.findAll(sessionRef.getSecurityCtx().getTenantId(), entityId, queries), callback, executor), callback::onFailure)); |
|
|
|
} |
|
|
|
|
|
|
|
private void handleWsAttributesSubscription(TelemetryWebSocketSessionRef sessionRef, |
|
|
|
private void handleWsAttributesSubscription(WebSocketSessionRef sessionRef, |
|
|
|
AttributesSubscriptionCmd cmd, String sessionId, EntityId entityId) { |
|
|
|
FutureCallback<List<AttributeKvEntry>> callback = new FutureCallback<>() { |
|
|
|
@Override |
|
|
|
@ -563,7 +620,7 @@ public class DefaultTelemetryWebSocketService implements TelemetryWebSocketServi |
|
|
|
.entityId(entityId) |
|
|
|
.allKeys(true) |
|
|
|
.keyStates(subState) |
|
|
|
.updateConsumer(DefaultTelemetryWebSocketService.this::sendWsMsg) |
|
|
|
.updateProcessor((subscription, update) -> sendWsMsg(subscription.getSessionId(), update)) |
|
|
|
.scope(scope).build(); |
|
|
|
oldSubService.addSubscription(sub); |
|
|
|
} |
|
|
|
@ -585,7 +642,7 @@ public class DefaultTelemetryWebSocketService implements TelemetryWebSocketServi |
|
|
|
} |
|
|
|
} |
|
|
|
|
|
|
|
private void handleWsTimeseriesSubscriptionCmd(TelemetryWebSocketSessionRef sessionRef, TimeseriesSubscriptionCmd cmd) { |
|
|
|
private void handleWsTimeseriesSubscriptionCmd(WebSocketSessionRef sessionRef, TimeseriesSubscriptionCmd cmd) { |
|
|
|
String sessionId = sessionRef.getSessionId(); |
|
|
|
log.debug("[{}] Processing: {}", sessionId, cmd); |
|
|
|
|
|
|
|
@ -605,7 +662,7 @@ public class DefaultTelemetryWebSocketService implements TelemetryWebSocketServi |
|
|
|
} |
|
|
|
} |
|
|
|
|
|
|
|
private void handleWsTimeseriesSubscriptionByKeys(TelemetryWebSocketSessionRef sessionRef, |
|
|
|
private void handleWsTimeseriesSubscriptionByKeys(WebSocketSessionRef sessionRef, |
|
|
|
TimeseriesSubscriptionCmd cmd, String sessionId, EntityId entityId) { |
|
|
|
long startTs; |
|
|
|
if (cmd.getTimeWindow() > 0) { |
|
|
|
@ -629,7 +686,7 @@ public class DefaultTelemetryWebSocketService implements TelemetryWebSocketServi |
|
|
|
} |
|
|
|
} |
|
|
|
|
|
|
|
private void handleWsTimeseriesSubscription(TelemetryWebSocketSessionRef sessionRef, |
|
|
|
private void handleWsTimeseriesSubscription(WebSocketSessionRef sessionRef, |
|
|
|
TimeseriesSubscriptionCmd cmd, String sessionId, EntityId entityId) { |
|
|
|
FutureCallback<List<TsKvEntry>> callback = new FutureCallback<List<TsKvEntry>>() { |
|
|
|
@Override |
|
|
|
@ -644,7 +701,7 @@ public class DefaultTelemetryWebSocketService implements TelemetryWebSocketServi |
|
|
|
.subscriptionId(cmd.getCmdId()) |
|
|
|
.tenantId(sessionRef.getSecurityCtx().getTenantId()) |
|
|
|
.entityId(entityId) |
|
|
|
.updateConsumer(DefaultTelemetryWebSocketService.this::sendWsMsg) |
|
|
|
.updateProcessor((subscription, update) -> sendWsMsg(subscription.getSessionId(), update)) |
|
|
|
.allKeys(true) |
|
|
|
.keyStates(subState).build(); |
|
|
|
oldSubService.addSubscription(sub); |
|
|
|
@ -667,7 +724,7 @@ public class DefaultTelemetryWebSocketService implements TelemetryWebSocketServi |
|
|
|
on(r -> Futures.addCallback(tsService.findAllLatest(sessionRef.getSecurityCtx().getTenantId(), entityId), callback, executor), callback::onFailure)); |
|
|
|
} |
|
|
|
|
|
|
|
private FutureCallback<List<TsKvEntry>> getSubscriptionCallback(final TelemetryWebSocketSessionRef sessionRef, final TimeseriesSubscriptionCmd cmd, final String sessionId, final EntityId entityId, final long startTs, final List<String> keys) { |
|
|
|
private FutureCallback<List<TsKvEntry>> getSubscriptionCallback(final WebSocketSessionRef sessionRef, final TimeseriesSubscriptionCmd cmd, final String sessionId, final EntityId entityId, final long startTs, final List<String> keys) { |
|
|
|
return new FutureCallback<>() { |
|
|
|
@Override |
|
|
|
public void onSuccess(List<TsKvEntry> data) { |
|
|
|
@ -682,7 +739,7 @@ public class DefaultTelemetryWebSocketService implements TelemetryWebSocketServi |
|
|
|
.subscriptionId(cmd.getCmdId()) |
|
|
|
.tenantId(sessionRef.getSecurityCtx().getTenantId()) |
|
|
|
.entityId(entityId) |
|
|
|
.updateConsumer(DefaultTelemetryWebSocketService.this::sendWsMsg) |
|
|
|
.updateProcessor((subscription, update) -> sendWsMsg(subscription.getSessionId(), update)) |
|
|
|
.allKeys(false) |
|
|
|
.keyStates(subState).build(); |
|
|
|
oldSubService.addSubscription(sub); |
|
|
|
@ -702,7 +759,7 @@ public class DefaultTelemetryWebSocketService implements TelemetryWebSocketServi |
|
|
|
}; |
|
|
|
} |
|
|
|
|
|
|
|
private void unsubscribe(TelemetryWebSocketSessionRef sessionRef, SubscriptionCmd cmd, String sessionId) { |
|
|
|
private void unsubscribe(WebSocketSessionRef sessionRef, SubscriptionCmd cmd, String sessionId) { |
|
|
|
if (cmd.getEntityId() == null || cmd.getEntityId().isEmpty()) { |
|
|
|
oldSubService.cancelAllSessionSubscriptions(sessionId); |
|
|
|
} else { |
|
|
|
@ -710,7 +767,7 @@ public class DefaultTelemetryWebSocketService implements TelemetryWebSocketServi |
|
|
|
} |
|
|
|
} |
|
|
|
|
|
|
|
private boolean validateSubscriptionCmd(TelemetryWebSocketSessionRef sessionRef, EntityDataCmd cmd) { |
|
|
|
private boolean validateSubscriptionCmd(WebSocketSessionRef sessionRef, EntityDataCmd cmd) { |
|
|
|
if (cmd.getCmdId() < 0) { |
|
|
|
TelemetrySubscriptionUpdate update = new TelemetrySubscriptionUpdate(cmd.getCmdId(), SubscriptionErrorCode.BAD_REQUEST, |
|
|
|
"Cmd id is negative value!"); |
|
|
|
@ -725,7 +782,7 @@ public class DefaultTelemetryWebSocketService implements TelemetryWebSocketServi |
|
|
|
return true; |
|
|
|
} |
|
|
|
|
|
|
|
private boolean validateSubscriptionCmd(TelemetryWebSocketSessionRef sessionRef, EntityCountCmd cmd) { |
|
|
|
private boolean validateSubscriptionCmd(WebSocketSessionRef sessionRef, EntityCountCmd cmd) { |
|
|
|
if (cmd.getCmdId() < 0) { |
|
|
|
TelemetrySubscriptionUpdate update = new TelemetrySubscriptionUpdate(cmd.getCmdId(), SubscriptionErrorCode.BAD_REQUEST, |
|
|
|
"Cmd id is negative value!"); |
|
|
|
@ -739,7 +796,7 @@ public class DefaultTelemetryWebSocketService implements TelemetryWebSocketServi |
|
|
|
return true; |
|
|
|
} |
|
|
|
|
|
|
|
private boolean validateSubscriptionCmd(TelemetryWebSocketSessionRef sessionRef, AlarmDataCmd cmd) { |
|
|
|
private boolean validateSubscriptionCmd(WebSocketSessionRef sessionRef, AlarmDataCmd cmd) { |
|
|
|
if (cmd.getCmdId() < 0) { |
|
|
|
TelemetrySubscriptionUpdate update = new TelemetrySubscriptionUpdate(cmd.getCmdId(), SubscriptionErrorCode.BAD_REQUEST, |
|
|
|
"Cmd id is negative value!"); |
|
|
|
@ -754,7 +811,7 @@ public class DefaultTelemetryWebSocketService implements TelemetryWebSocketServi |
|
|
|
return true; |
|
|
|
} |
|
|
|
|
|
|
|
private boolean validateSubscriptionCmd(TelemetryWebSocketSessionRef sessionRef, SubscriptionCmd cmd) { |
|
|
|
private boolean validateSubscriptionCmd(WebSocketSessionRef sessionRef, SubscriptionCmd cmd) { |
|
|
|
if (cmd.getEntityId() == null || cmd.getEntityId().isEmpty()) { |
|
|
|
TelemetrySubscriptionUpdate update = new TelemetrySubscriptionUpdate(cmd.getCmdId(), SubscriptionErrorCode.BAD_REQUEST, |
|
|
|
"Device id is empty!"); |
|
|
|
@ -764,11 +821,11 @@ public class DefaultTelemetryWebSocketService implements TelemetryWebSocketServi |
|
|
|
return true; |
|
|
|
} |
|
|
|
|
|
|
|
private boolean validateSessionMetadata(TelemetryWebSocketSessionRef sessionRef, SubscriptionCmd cmd, String sessionId) { |
|
|
|
private boolean validateSessionMetadata(WebSocketSessionRef sessionRef, SubscriptionCmd cmd, String sessionId) { |
|
|
|
return validateSessionMetadata(sessionRef, cmd.getCmdId(), sessionId); |
|
|
|
} |
|
|
|
|
|
|
|
private boolean validateSessionMetadata(TelemetryWebSocketSessionRef sessionRef, int cmdId, String sessionId) { |
|
|
|
private boolean validateSessionMetadata(WebSocketSessionRef sessionRef, int cmdId, String sessionId) { |
|
|
|
WsSessionMetaData sessionMD = wsSessionsMap.get(sessionId); |
|
|
|
if (sessionMD == null) { |
|
|
|
log.warn("[{}] Session meta data not found. ", sessionId); |
|
|
|
@ -781,15 +838,15 @@ public class DefaultTelemetryWebSocketService implements TelemetryWebSocketServi |
|
|
|
} |
|
|
|
} |
|
|
|
|
|
|
|
private void sendWsMsg(TelemetryWebSocketSessionRef sessionRef, EntityDataUpdate update) { |
|
|
|
private void sendWsMsg(WebSocketSessionRef sessionRef, EntityDataUpdate update) { |
|
|
|
sendWsMsg(sessionRef, update.getCmdId(), update); |
|
|
|
} |
|
|
|
|
|
|
|
private void sendWsMsg(TelemetryWebSocketSessionRef sessionRef, TelemetrySubscriptionUpdate update) { |
|
|
|
private void sendWsMsg(WebSocketSessionRef sessionRef, TelemetrySubscriptionUpdate update) { |
|
|
|
sendWsMsg(sessionRef, update.getSubscriptionId(), update); |
|
|
|
} |
|
|
|
|
|
|
|
private void sendWsMsg(TelemetryWebSocketSessionRef sessionRef, int cmdId, Object update) { |
|
|
|
private void sendWsMsg(WebSocketSessionRef sessionRef, int cmdId, Object update) { |
|
|
|
try { |
|
|
|
String msg = jsonMapper.writeValueAsString(update); |
|
|
|
executor.submit(() -> { |