From 5ee096093e0011948361b8aa3c84741e9f12b1a3 Mon Sep 17 00:00:00 2001 From: lsyer Date: Sun, 7 Jan 2024 14:18:25 +0800 Subject: [PATCH 1/4] added sending RPC(Server-side RPC) via websocket. --- .../service/ws/DefaultWebSocketService.java | 143 ++++++++++++++++++ .../server/service/ws/WsCmdType.java | 2 + .../server/service/ws/WsCommandsWrapper.java | 2 + .../telemetry/cmd/TelemetryCmdsWrapper.java | 4 + .../service/ws/telemetry/cmd/v1/RpcCmd.java | 46 ++++++ 5 files changed, 197 insertions(+) create mode 100644 application/src/main/java/org/thingsboard/server/service/ws/telemetry/cmd/v1/RpcCmd.java diff --git a/application/src/main/java/org/thingsboard/server/service/ws/DefaultWebSocketService.java b/application/src/main/java/org/thingsboard/server/service/ws/DefaultWebSocketService.java index 3e9ee69cfc..1de2f8e03e 100644 --- a/application/src/main/java/org/thingsboard/server/service/ws/DefaultWebSocketService.java +++ b/application/src/main/java/org/thingsboard/server/service/ws/DefaultWebSocketService.java @@ -16,6 +16,7 @@ package org.thingsboard.server.service.ws; import com.fasterxml.jackson.core.JsonProcessingException; +import com.fasterxml.jackson.databind.JsonNode; import com.google.common.base.Function; import com.google.common.util.concurrent.FutureCallback; import com.google.common.util.concurrent.Futures; @@ -25,6 +26,7 @@ import lombok.Getter; import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; import org.apache.commons.collections.CollectionUtils; +import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Value; import org.springframework.stereotype.Service; import org.springframework.web.socket.CloseStatus; @@ -34,10 +36,13 @@ import org.thingsboard.common.util.ThingsBoardThreadFactory; import org.thingsboard.server.common.data.DataConstants; import org.thingsboard.server.common.data.StringUtils; import org.thingsboard.server.common.data.TenantProfile; +import org.thingsboard.server.common.data.audit.ActionType; import org.thingsboard.server.common.data.id.CustomerId; +import org.thingsboard.server.common.data.id.DeviceId; import org.thingsboard.server.common.data.id.EntityId; import org.thingsboard.server.common.data.id.EntityIdFactory; import org.thingsboard.server.common.data.id.TenantId; +import org.thingsboard.server.common.data.id.UUIDBased; import org.thingsboard.server.common.data.id.UserId; import org.thingsboard.server.common.data.kv.Aggregation; import org.thingsboard.server.common.data.kv.AttributeKvEntry; @@ -45,18 +50,26 @@ import org.thingsboard.server.common.data.kv.BaseReadTsKvQuery; import org.thingsboard.server.common.data.kv.BasicTsKvEntry; import org.thingsboard.server.common.data.kv.ReadTsKvQuery; import org.thingsboard.server.common.data.kv.TsKvEntry; +import org.thingsboard.server.common.data.rpc.RpcError; +import org.thingsboard.server.common.data.rpc.ToDeviceRpcRequestBody; import org.thingsboard.server.common.data.tenant.profile.DefaultTenantProfileConfiguration; +import org.thingsboard.server.common.msg.rpc.FromDeviceRpcResponse; +import org.thingsboard.server.common.msg.rpc.ToDeviceRpcRequest; +import org.thingsboard.server.controller.BaseController; import org.thingsboard.server.dao.attributes.AttributesService; +import org.thingsboard.server.dao.audit.AuditLogService; 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.rpc.TbCoreDeviceRpcService; import org.thingsboard.server.service.security.AccessValidator; import org.thingsboard.server.service.security.ValidationCallback; import org.thingsboard.server.service.security.ValidationResult; import org.thingsboard.server.service.security.ValidationResultCode; +import org.thingsboard.server.service.security.model.SecurityUser; import org.thingsboard.server.service.security.model.UserPrincipal; import org.thingsboard.server.service.security.permission.Operation; import org.thingsboard.server.service.subscription.SubscriptionErrorCode; @@ -66,6 +79,7 @@ import org.thingsboard.server.service.subscription.TbEntityDataSubscriptionServi import org.thingsboard.server.service.subscription.TbLocalSubscriptionService; import org.thingsboard.server.service.subscription.TbTimeSeriesSubscription; import org.thingsboard.server.service.ws.notification.NotificationCommandsHandler; +import org.thingsboard.server.service.ws.telemetry.cmd.v1.RpcCmd; 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; @@ -93,6 +107,7 @@ import java.util.List; import java.util.Map; import java.util.Optional; import java.util.Set; +import java.util.UUID; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentMap; import java.util.concurrent.ExecutorService; @@ -159,6 +174,7 @@ public class DefaultWebSocketService implements WebSocketService { pingExecutor.scheduleWithFixedDelay(this::sendPing, pingTimeout / NUMBER_OF_PING_ATTEMPTS, pingTimeout / NUMBER_OF_PING_ATTEMPTS, TimeUnit.MILLISECONDS); cmdsHandlers = new EnumMap<>(WsCmdType.class); + cmdsHandlers.put(WsCmdType.RPC, newCmdHandler(this::handleWsRpcCmd)); cmdsHandlers.put(WsCmdType.ATTRIBUTES, newCmdHandler(this::handleWsAttributesSubscriptionCmd)); cmdsHandlers.put(WsCmdType.TIMESERIES, newCmdHandler(this::handleWsTimeseriesSubscriptionCmd)); cmdsHandlers.put(WsCmdType.TIMESERIES_HISTORY, newCmdHandler(this::handleWsHistoryCmd)); @@ -399,6 +415,133 @@ public class DefaultWebSocketService implements WebSocketService { return true; } + @Autowired + protected TbCoreDeviceRpcService deviceRpcService; + + @Value("${server.rest.server_side_rpc.min_timeout:5000}") + protected long minTimeout; + + @Value("${server.rest.server_side_rpc.default_timeout:10000}") + protected long defaultTimeout; + + private void handleWsRpcCmd(WebSocketSessionRef sessionRef, RpcCmd cmd) { + if (!validateCmd(sessionRef, cmd, () -> { + if (cmd.getEntityId() == null || cmd.getEntityId().isEmpty() || cmd.getEntityType() == null || cmd.getEntityType().isEmpty()) { + throw new IllegalArgumentException("Device id is empty!"); + } + if (cmd.getRpcjson() == null || cmd.getRpcjson().isEmpty()) { + throw new IllegalArgumentException("RPC JSON is empty!"); + } + })) return; + + UUID rpcRequestUUID = UUID.randomUUID(); + EntityId entityId = EntityIdFactory.getByTypeAndId(cmd.getEntityType(), cmd.getEntityId()); + SecurityUser currentUser = sessionRef.getSecurityCtx(); + TenantId tenantId = currentUser.getTenantId(); + DeviceId deviceId = DeviceId.fromString(cmd.getEntityId()); + boolean oneWay = true; + long timeout = cmd.getTimeout() > 0 ? cmd.getTimeout() : defaultTimeout; + long expTime = cmd.getExpTime() > 0 ? cmd.getExpTime() : System.currentTimeMillis() + Math.max(minTimeout, timeout); + JsonNode rpcRequestBody = JacksonUtil.toJsonNode(cmd.getRpcjson()); + ToDeviceRpcRequestBody body = new ToDeviceRpcRequestBody(rpcRequestBody.get("method").asText(), JacksonUtil.toString(rpcRequestBody.get("params"))); + boolean persisted = cmd.isPersisted(); + int retries = Math.max(cmd.getRetries(), 0); + + FutureCallback callback = new FutureCallback<>() { + @Override + public void onSuccess(@Nullable ValidationResult result) { + ToDeviceRpcRequest rpcRequest = new ToDeviceRpcRequest(rpcRequestUUID, + tenantId, + deviceId, + oneWay, + expTime, + body, + persisted, + retries, + null + ); + deviceRpcService.processRestApiRpcRequest(rpcRequest, fromDeviceRpcResponse -> reply(rpcRequest, currentUser, fromDeviceRpcResponse, sessionRef, cmd.getCmdId()), currentUser); + } + + @Override + public void onFailure(Throwable e) { + TelemetrySubscriptionUpdate update; + if (UnauthorizedException.class.isInstance(e)) { + update = new TelemetrySubscriptionUpdate(cmd.getCmdId(), SubscriptionErrorCode.UNAUTHORIZED, + SubscriptionErrorCode.UNAUTHORIZED.getDefaultMsg()); + } else { + update = new TelemetrySubscriptionUpdate(cmd.getCmdId(), SubscriptionErrorCode.INTERNAL_ERROR, + FAILED_TO_FETCH_DATA); + } + sendUpdate(sessionRef, update); + } + }; + accessValidator.validate(sessionRef.getSecurityCtx(), Operation.RPC_CALL, entityId, callback); + } + + public void reply(ToDeviceRpcRequest rpcRequest, SecurityUser currentUser, FromDeviceRpcResponse response, WebSocketSessionRef sessionRef, int cmdId) { + Optional rpcError = response.getError(); + if (rpcError.isPresent()) { + logRpcCall(currentUser, rpcRequest, rpcError, null); + RpcError error = rpcError.get(); + switch (error) { + case TIMEOUT: + sendUpdate(sessionRef, new TelemetrySubscriptionUpdate(cmdId, SubscriptionErrorCode.INTERNAL_ERROR, "TIMEOUT")); + break; + case NO_ACTIVE_CONNECTION: + sendUpdate(sessionRef, new TelemetrySubscriptionUpdate(cmdId, SubscriptionErrorCode.INTERNAL_ERROR, "NO_ACTIVE_CONNECTION")); + break; + default: + sendUpdate(sessionRef, new TelemetrySubscriptionUpdate(cmdId, SubscriptionErrorCode.INTERNAL_ERROR, error.toString())); + break; + } + } else { + Optional responseData = response.getResponse(); + if (responseData.isPresent() && !StringUtils.isEmpty(responseData.get())) { + String data = responseData.get(); + try { + logRpcCall(currentUser, rpcRequest, rpcError, null); + sendUpdate(sessionRef, new TelemetrySubscriptionUpdate(cmdId, SubscriptionErrorCode.NO_ERROR, data)); + } catch (IllegalArgumentException e) { + log.debug("Failed to decode device response: {}", data, e); + logRpcCall(currentUser, rpcRequest, rpcError, e); + } + } else { + logRpcCall(currentUser, rpcRequest, rpcError, null); + sendUpdate(sessionRef, new TelemetrySubscriptionUpdate(cmdId, SubscriptionErrorCode.NO_ERROR, "RPC SUCCESS!")); + } + } + } + + private void logRpcCall(SecurityUser currentUser, ToDeviceRpcRequest rpcRequest, Optional rpcError, Throwable e) { + logRpcCall(currentUser, rpcRequest.getDeviceId(), rpcRequest.getBody(), rpcRequest.isOneway(), rpcError, null); + } + + @Autowired + protected AuditLogService auditLogService; + private void logRpcCall(SecurityUser user, EntityId entityId, ToDeviceRpcRequestBody body, boolean oneWay, Optional rpcError, Throwable e) { + String rpcErrorStr = ""; + if (rpcError.isPresent()) { + rpcErrorStr = "RPC Error: " + rpcError.get().name(); + } + String method = body.getMethod(); + String params = body.getParams(); + + auditLogService.logEntityAction( + user.getTenantId(), + user.getCustomerId(), + user.getId(), + user.getName(), + (UUIDBased & EntityId) entityId, + null, + ActionType.RPC_CALL, + BaseController.toException(e), + rpcErrorStr, + oneWay, + method, + params); + } + private void handleWsAttributesSubscriptionCmd(WebSocketSessionRef sessionRef, AttributesSubscriptionCmd cmd) { if (!processSubscription(sessionRef, cmd)) { return; diff --git a/application/src/main/java/org/thingsboard/server/service/ws/WsCmdType.java b/application/src/main/java/org/thingsboard/server/service/ws/WsCmdType.java index 285dd88efe..a255184962 100644 --- a/application/src/main/java/org/thingsboard/server/service/ws/WsCmdType.java +++ b/application/src/main/java/org/thingsboard/server/service/ws/WsCmdType.java @@ -18,6 +18,8 @@ package org.thingsboard.server.service.ws; public enum WsCmdType { AUTH, + RPC, + ATTRIBUTES, TIMESERIES, TIMESERIES_HISTORY, diff --git a/application/src/main/java/org/thingsboard/server/service/ws/WsCommandsWrapper.java b/application/src/main/java/org/thingsboard/server/service/ws/WsCommandsWrapper.java index 5704f3b2dc..47ab8b508d 100644 --- a/application/src/main/java/org/thingsboard/server/service/ws/WsCommandsWrapper.java +++ b/application/src/main/java/org/thingsboard/server/service/ws/WsCommandsWrapper.java @@ -26,6 +26,7 @@ import org.thingsboard.server.service.ws.notification.cmd.MarkNotificationsAsRea 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.RpcCmd; 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.TimeseriesSubscriptionCmd; @@ -49,6 +50,7 @@ public class WsCommandsWrapper { @JsonTypeInfo(use = JsonTypeInfo.Id.NAME, property = "type") @JsonSubTypes({ + @Type(name = "RPC", value = RpcCmd.class), @Type(name = "ATTRIBUTES", value = AttributesSubscriptionCmd.class), @Type(name = "TIMESERIES", value = TimeseriesSubscriptionCmd.class), @Type(name = "TIMESERIES_HISTORY", value = GetHistoryCmd.class), diff --git a/application/src/main/java/org/thingsboard/server/service/ws/telemetry/cmd/TelemetryCmdsWrapper.java b/application/src/main/java/org/thingsboard/server/service/ws/telemetry/cmd/TelemetryCmdsWrapper.java index 1019124e6d..1a7b2395de 100644 --- a/application/src/main/java/org/thingsboard/server/service/ws/telemetry/cmd/TelemetryCmdsWrapper.java +++ b/application/src/main/java/org/thingsboard/server/service/ws/telemetry/cmd/TelemetryCmdsWrapper.java @@ -18,6 +18,7 @@ package org.thingsboard.server.service.ws.telemetry.cmd; import com.fasterxml.jackson.annotation.JsonIgnore; import lombok.Data; import org.thingsboard.server.service.ws.WsCommandsWrapper; +import org.thingsboard.server.service.ws.telemetry.cmd.v1.RpcCmd; 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.TimeseriesSubscriptionCmd; @@ -43,6 +44,8 @@ import java.util.stream.Stream; @Deprecated public class TelemetryCmdsWrapper { + private List rpcCmds; + private List attrSubCmds; private List tsSubCmds; @@ -68,6 +71,7 @@ public class TelemetryCmdsWrapper { @JsonIgnore public WsCommandsWrapper toCommonCmdsWrapper() { return new WsCommandsWrapper(null, Stream.of( + rpcCmds, attrSubCmds, tsSubCmds, historyCmds, entityDataCmds, entityDataUnsubscribeCmds, alarmDataCmds, alarmDataUnsubscribeCmds, entityCountCmds, entityCountUnsubscribeCmds, diff --git a/application/src/main/java/org/thingsboard/server/service/ws/telemetry/cmd/v1/RpcCmd.java b/application/src/main/java/org/thingsboard/server/service/ws/telemetry/cmd/v1/RpcCmd.java new file mode 100644 index 0000000000..8230730926 --- /dev/null +++ b/application/src/main/java/org/thingsboard/server/service/ws/telemetry/cmd/v1/RpcCmd.java @@ -0,0 +1,46 @@ +/** + * Copyright © 2016-2023 The Thingsboard Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.thingsboard.server.service.ws.telemetry.cmd.v1; + +import lombok.AllArgsConstructor; +import lombok.Data; +import lombok.NoArgsConstructor; + +import org.thingsboard.server.service.ws.WsCmd; +import org.thingsboard.server.service.ws.WsCmdType; + +/** + * @author lsyer + */ +@NoArgsConstructor +@AllArgsConstructor +@Data +public class RpcCmd implements WsCmd { + + private int cmdId; + private String entityType; + private String entityId; + private String rpcjson; + private long timeout; + private long expTime; + private boolean persisted; + private int retries; + + @Override + public WsCmdType getType() { + return WsCmdType.RPC; + } +} \ No newline at end of file From 9f859a3704d96414707564e9887c0ec0c6310693 Mon Sep 17 00:00:00 2001 From: lsyer Date: Sun, 7 Jan 2024 18:28:09 +0800 Subject: [PATCH 2/4] improve sending RPC(Server-side RPC) via websocket --- .../server/service/ws/DefaultWebSocketService.java | 13 ++++--------- .../server/service/ws/telemetry/cmd/v1/RpcCmd.java | 9 +++++---- 2 files changed, 9 insertions(+), 13 deletions(-) diff --git a/application/src/main/java/org/thingsboard/server/service/ws/DefaultWebSocketService.java b/application/src/main/java/org/thingsboard/server/service/ws/DefaultWebSocketService.java index 1de2f8e03e..2acb575be6 100644 --- a/application/src/main/java/org/thingsboard/server/service/ws/DefaultWebSocketService.java +++ b/application/src/main/java/org/thingsboard/server/service/ws/DefaultWebSocketService.java @@ -439,7 +439,7 @@ public class DefaultWebSocketService implements WebSocketService { SecurityUser currentUser = sessionRef.getSecurityCtx(); TenantId tenantId = currentUser.getTenantId(); DeviceId deviceId = DeviceId.fromString(cmd.getEntityId()); - boolean oneWay = true; + boolean oneWay = cmd.isOneway(); long timeout = cmd.getTimeout() > 0 ? cmd.getTimeout() : defaultTimeout; long expTime = cmd.getExpTime() > 0 ? cmd.getExpTime() : System.currentTimeMillis() + Math.max(minTimeout, timeout); JsonNode rpcRequestBody = JacksonUtil.toJsonNode(cmd.getRpcjson()); @@ -499,16 +499,11 @@ public class DefaultWebSocketService implements WebSocketService { Optional responseData = response.getResponse(); if (responseData.isPresent() && !StringUtils.isEmpty(responseData.get())) { String data = responseData.get(); - try { - logRpcCall(currentUser, rpcRequest, rpcError, null); - sendUpdate(sessionRef, new TelemetrySubscriptionUpdate(cmdId, SubscriptionErrorCode.NO_ERROR, data)); - } catch (IllegalArgumentException e) { - log.debug("Failed to decode device response: {}", data, e); - logRpcCall(currentUser, rpcRequest, rpcError, e); - } + logRpcCall(currentUser, rpcRequest, rpcError, null); + sendUpdate(sessionRef, cmdId, "{'msg' : '"+data+"'}"); } else { logRpcCall(currentUser, rpcRequest, rpcError, null); - sendUpdate(sessionRef, new TelemetrySubscriptionUpdate(cmdId, SubscriptionErrorCode.NO_ERROR, "RPC SUCCESS!")); + // sendUpdate(sessionRef, cmdId, "{'msg' : 'RPC Success!'}"); } } } diff --git a/application/src/main/java/org/thingsboard/server/service/ws/telemetry/cmd/v1/RpcCmd.java b/application/src/main/java/org/thingsboard/server/service/ws/telemetry/cmd/v1/RpcCmd.java index 8230730926..ac12ca8626 100644 --- a/application/src/main/java/org/thingsboard/server/service/ws/telemetry/cmd/v1/RpcCmd.java +++ b/application/src/main/java/org/thingsboard/server/service/ws/telemetry/cmd/v1/RpcCmd.java @@ -30,14 +30,15 @@ import org.thingsboard.server.service.ws.WsCmdType; @Data public class RpcCmd implements WsCmd { - private int cmdId; - private String entityType; - private String entityId; - private String rpcjson; + private int cmdId; //required + private String entityType; //required + private String entityId; //required + private String rpcjson; //required private long timeout; private long expTime; private boolean persisted; private int retries; + private boolean oneway; //default: false @Override public WsCmdType getType() { From 04c11fcb194b87f04a850973dfd1bc858ed7a710 Mon Sep 17 00:00:00 2001 From: lsyer Date: Fri, 20 Sep 2024 18:06:07 +0800 Subject: [PATCH 3/4] improve sending RPC(Server-side RPC) via websocket --- .../service/ws/DefaultWebSocketService.java | 25 +++++++++---------- .../server/service/ws/WsCmdType.java | 6 ++--- .../server/service/ws/WsCommandsWrapper.java | 2 +- .../telemetry/cmd/TelemetryCmdsWrapper.java | 4 +-- .../service/ws/telemetry/cmd/v1/RpcCmd.java | 3 ++- 5 files changed, 20 insertions(+), 20 deletions(-) diff --git a/application/src/main/java/org/thingsboard/server/service/ws/DefaultWebSocketService.java b/application/src/main/java/org/thingsboard/server/service/ws/DefaultWebSocketService.java index 649c91cb79..f878c4e8fc 100644 --- a/application/src/main/java/org/thingsboard/server/service/ws/DefaultWebSocketService.java +++ b/application/src/main/java/org/thingsboard/server/service/ws/DefaultWebSocketService.java @@ -28,7 +28,7 @@ import jakarta.annotation.PreDestroy; import lombok.Getter; import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; -import org.apache.commons.collections.CollectionUtils; +import org.apache.commons.collections4.CollectionUtils; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Value; import org.springframework.stereotype.Service; @@ -167,6 +167,14 @@ public class DefaultWebSocketService implements WebSocketService { private Map> cmdsHandlers; + private final TbCoreDeviceRpcService deviceRpcService; + + @Value("${server.rest.server_side_rpc.min_timeout:5000}") + private final long rpcMinTimeout; + + @Value("${server.rest.server_side_rpc.default_timeout:10000}") + private final long rpcDefaultTimeout; + @PostConstruct public void init() { serviceId = serviceInfoProvider.getServiceId(); @@ -176,7 +184,6 @@ public class DefaultWebSocketService implements WebSocketService { pingExecutor.scheduleWithFixedDelay(this::sendPing, pingTimeout / NUMBER_OF_PING_ATTEMPTS, pingTimeout / NUMBER_OF_PING_ATTEMPTS, TimeUnit.MILLISECONDS); cmdsHandlers = new EnumMap<>(WsCmdType.class); - cmdsHandlers.put(WsCmdType.RPC, newCmdHandler(this::handleWsRpcCmd)); cmdsHandlers.put(WsCmdType.ATTRIBUTES, newCmdHandler(this::handleWsAttributesSubscriptionCmd)); cmdsHandlers.put(WsCmdType.TIMESERIES, newCmdHandler(this::handleWsTimeseriesSubscriptionCmd)); cmdsHandlers.put(WsCmdType.TIMESERIES_HISTORY, newCmdHandler(this::handleWsHistoryCmd)); @@ -193,6 +200,7 @@ public class DefaultWebSocketService implements WebSocketService { cmdsHandlers.put(WsCmdType.MARK_NOTIFICATIONS_AS_READ, newCmdHandler(notificationCmdsHandler::handleMarkAsReadCmd)); cmdsHandlers.put(WsCmdType.MARK_ALL_NOTIFICATIONS_AS_READ, newCmdHandler(notificationCmdsHandler::handleMarkAllAsReadCmd)); cmdsHandlers.put(WsCmdType.NOTIFICATIONS_UNSUBSCRIBE, newCmdHandler(notificationCmdsHandler::handleUnsubCmd)); + cmdsHandlers.put(WsCmdType.RPC, newCmdHandler(this::handleWsRpcCmd)); } @PreDestroy @@ -427,15 +435,6 @@ public class DefaultWebSocketService implements WebSocketService { return true; } - @Autowired - protected TbCoreDeviceRpcService deviceRpcService; - - @Value("${server.rest.server_side_rpc.min_timeout:5000}") - protected long minTimeout; - - @Value("${server.rest.server_side_rpc.default_timeout:10000}") - protected long defaultTimeout; - private void handleWsRpcCmd(WebSocketSessionRef sessionRef, RpcCmd cmd) { if (!validateCmd(sessionRef, cmd, () -> { if (cmd.getEntityId() == null || cmd.getEntityId().isEmpty() || cmd.getEntityType() == null || cmd.getEntityType().isEmpty()) { @@ -452,8 +451,8 @@ public class DefaultWebSocketService implements WebSocketService { TenantId tenantId = currentUser.getTenantId(); DeviceId deviceId = DeviceId.fromString(cmd.getEntityId()); boolean oneWay = cmd.isOneway(); - long timeout = cmd.getTimeout() > 0 ? cmd.getTimeout() : defaultTimeout; - long expTime = cmd.getExpTime() > 0 ? cmd.getExpTime() : System.currentTimeMillis() + Math.max(minTimeout, timeout); + long timeout = cmd.getTimeout() > 0 ? cmd.getTimeout() : rpcDefaultTimeout; + long expTime = cmd.getExpTime() > 0 ? cmd.getExpTime() : System.currentTimeMillis() + Math.max(rpcMinTimeout, timeout); JsonNode rpcRequestBody = JacksonUtil.toJsonNode(cmd.getRpcjson()); ToDeviceRpcRequestBody body = new ToDeviceRpcRequestBody(rpcRequestBody.get("method").asText(), JacksonUtil.toString(rpcRequestBody.get("params"))); boolean persisted = cmd.isPersisted(); diff --git a/application/src/main/java/org/thingsboard/server/service/ws/WsCmdType.java b/application/src/main/java/org/thingsboard/server/service/ws/WsCmdType.java index b2d697ce55..282646f6a2 100644 --- a/application/src/main/java/org/thingsboard/server/service/ws/WsCmdType.java +++ b/application/src/main/java/org/thingsboard/server/service/ws/WsCmdType.java @@ -18,8 +18,6 @@ package org.thingsboard.server.service.ws; public enum WsCmdType { AUTH, - RPC, - ATTRIBUTES, TIMESERIES, TIMESERIES_HISTORY, @@ -37,5 +35,7 @@ public enum WsCmdType { ALARM_COUNT_UNSUBSCRIBE, ENTITY_DATA_UNSUBSCRIBE, ENTITY_COUNT_UNSUBSCRIBE, - NOTIFICATIONS_UNSUBSCRIBE + NOTIFICATIONS_UNSUBSCRIBE, + + RPC, } diff --git a/application/src/main/java/org/thingsboard/server/service/ws/WsCommandsWrapper.java b/application/src/main/java/org/thingsboard/server/service/ws/WsCommandsWrapper.java index 6e0c9e2e5b..a256734aa7 100644 --- a/application/src/main/java/org/thingsboard/server/service/ws/WsCommandsWrapper.java +++ b/application/src/main/java/org/thingsboard/server/service/ws/WsCommandsWrapper.java @@ -50,7 +50,6 @@ public class WsCommandsWrapper { @JsonTypeInfo(use = JsonTypeInfo.Id.NAME, property = "type") @JsonSubTypes({ - @Type(name = "RPC", value = RpcCmd.class), @Type(name = "ATTRIBUTES", value = AttributesSubscriptionCmd.class), @Type(name = "TIMESERIES", value = TimeseriesSubscriptionCmd.class), @Type(name = "TIMESERIES_HISTORY", value = GetHistoryCmd.class), @@ -67,6 +66,7 @@ public class WsCommandsWrapper { @Type(name = "ENTITY_DATA_UNSUBSCRIBE", value = EntityDataUnsubscribeCmd.class), @Type(name = "ENTITY_COUNT_UNSUBSCRIBE", value = EntityCountUnsubscribeCmd.class), @Type(name = "NOTIFICATIONS_UNSUBSCRIBE", value = NotificationsUnsubCmd.class), + @Type(name = "RPC", value = RpcCmd.class), }) private List cmds; diff --git a/application/src/main/java/org/thingsboard/server/service/ws/telemetry/cmd/TelemetryCmdsWrapper.java b/application/src/main/java/org/thingsboard/server/service/ws/telemetry/cmd/TelemetryCmdsWrapper.java index 374536fc6f..2e09fcdac6 100644 --- a/application/src/main/java/org/thingsboard/server/service/ws/telemetry/cmd/TelemetryCmdsWrapper.java +++ b/application/src/main/java/org/thingsboard/server/service/ws/telemetry/cmd/TelemetryCmdsWrapper.java @@ -44,8 +44,6 @@ import java.util.stream.Stream; @Deprecated public class TelemetryCmdsWrapper { - private List rpcCmds; - private List attrSubCmds; private List tsSubCmds; @@ -68,6 +66,8 @@ public class TelemetryCmdsWrapper { private List alarmCountUnsubscribeCmds; + private List rpcCmds; + @JsonIgnore public WsCommandsWrapper toCommonCmdsWrapper() { return new WsCommandsWrapper(null, Stream.of( diff --git a/application/src/main/java/org/thingsboard/server/service/ws/telemetry/cmd/v1/RpcCmd.java b/application/src/main/java/org/thingsboard/server/service/ws/telemetry/cmd/v1/RpcCmd.java index ac12ca8626..023cd40884 100644 --- a/application/src/main/java/org/thingsboard/server/service/ws/telemetry/cmd/v1/RpcCmd.java +++ b/application/src/main/java/org/thingsboard/server/service/ws/telemetry/cmd/v1/RpcCmd.java @@ -1,5 +1,5 @@ /** - * Copyright © 2016-2023 The Thingsboard Authors + * Copyright © 2016-2024 The Thingsboard Authors * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -19,6 +19,7 @@ import lombok.AllArgsConstructor; import lombok.Data; import lombok.NoArgsConstructor; +import org.thingsboard.server.common.data.id.EntityId; import org.thingsboard.server.service.ws.WsCmd; import org.thingsboard.server.service.ws.WsCmdType; From 79d2ac89646c6a156faa0403eb998ce46bac228d Mon Sep 17 00:00:00 2001 From: lsyer Date: Fri, 15 Nov 2024 12:27:48 +0800 Subject: [PATCH 4/4] [Complete]Sending RPC(Server-side RPC) via websocket --- .../server/service/ws/DefaultWebSocketService.java | 8 ++++---- .../server/service/ws/telemetry/cmd/v1/RpcCmd.java | 2 +- 2 files changed, 5 insertions(+), 5 deletions(-) diff --git a/application/src/main/java/org/thingsboard/server/service/ws/DefaultWebSocketService.java b/application/src/main/java/org/thingsboard/server/service/ws/DefaultWebSocketService.java index f878c4e8fc..966f64cbca 100644 --- a/application/src/main/java/org/thingsboard/server/service/ws/DefaultWebSocketService.java +++ b/application/src/main/java/org/thingsboard/server/service/ws/DefaultWebSocketService.java @@ -170,10 +170,10 @@ public class DefaultWebSocketService implements WebSocketService { private final TbCoreDeviceRpcService deviceRpcService; @Value("${server.rest.server_side_rpc.min_timeout:5000}") - private final long rpcMinTimeout; + private long rpcMinTimeout; @Value("${server.rest.server_side_rpc.default_timeout:10000}") - private final long rpcDefaultTimeout; + private long rpcDefaultTimeout; @PostConstruct public void init() { @@ -440,7 +440,7 @@ public class DefaultWebSocketService implements WebSocketService { if (cmd.getEntityId() == null || cmd.getEntityId().isEmpty() || cmd.getEntityType() == null || cmd.getEntityType().isEmpty()) { throw new IllegalArgumentException("Device id is empty!"); } - if (cmd.getRpcjson() == null || cmd.getRpcjson().isEmpty()) { + if (cmd.getRpcJson() == null || cmd.getRpcJson().isEmpty()) { throw new IllegalArgumentException("RPC JSON is empty!"); } })) return; @@ -453,7 +453,7 @@ public class DefaultWebSocketService implements WebSocketService { boolean oneWay = cmd.isOneway(); long timeout = cmd.getTimeout() > 0 ? cmd.getTimeout() : rpcDefaultTimeout; long expTime = cmd.getExpTime() > 0 ? cmd.getExpTime() : System.currentTimeMillis() + Math.max(rpcMinTimeout, timeout); - JsonNode rpcRequestBody = JacksonUtil.toJsonNode(cmd.getRpcjson()); + JsonNode rpcRequestBody = JacksonUtil.toJsonNode(cmd.getRpcJson()); ToDeviceRpcRequestBody body = new ToDeviceRpcRequestBody(rpcRequestBody.get("method").asText(), JacksonUtil.toString(rpcRequestBody.get("params"))); boolean persisted = cmd.isPersisted(); int retries = Math.max(cmd.getRetries(), 0); diff --git a/application/src/main/java/org/thingsboard/server/service/ws/telemetry/cmd/v1/RpcCmd.java b/application/src/main/java/org/thingsboard/server/service/ws/telemetry/cmd/v1/RpcCmd.java index 023cd40884..7972ad77f2 100644 --- a/application/src/main/java/org/thingsboard/server/service/ws/telemetry/cmd/v1/RpcCmd.java +++ b/application/src/main/java/org/thingsboard/server/service/ws/telemetry/cmd/v1/RpcCmd.java @@ -34,7 +34,7 @@ public class RpcCmd implements WsCmd { private int cmdId; //required private String entityType; //required private String entityId; //required - private String rpcjson; //required + private String rpcJson; //required private long timeout; private long expTime; private boolean persisted;