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 856c0d60b0..34830a96da 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; @@ -28,6 +29,7 @@ import lombok.Getter; import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; 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; import org.springframework.web.socket.CloseStatus; @@ -37,10 +39,13 @@ import org.thingsboard.common.util.ThingsBoardThreadFactory; import org.thingsboard.server.common.data.AttributeScope; 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; @@ -48,19 +53,27 @@ 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.common.msg.tools.TbRateLimitsException; 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; @@ -70,6 +83,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; @@ -95,6 +109,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; @@ -153,6 +168,14 @@ public class DefaultWebSocketService implements WebSocketService { private Map> cmdsHandlers; + private final TbCoreDeviceRpcService deviceRpcService; + + @Value("${server.rest.server_side_rpc.min_timeout:5000}") + private long rpcMinTimeout; + + @Value("${server.rest.server_side_rpc.default_timeout:10000}") + private long rpcDefaultTimeout; + @PostConstruct public void init() { serviceId = serviceInfoProvider.getServiceId(); @@ -180,6 +203,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 @@ -420,6 +444,119 @@ public class DefaultWebSocketService implements WebSocketService { return true; } + 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 = 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()); + 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(); + logRpcCall(currentUser, rpcRequest, rpcError, null); + sendUpdate(sessionRef, cmdId, "{'msg' : '"+data+"'}"); + } else { + logRpcCall(currentUser, rpcRequest, rpcError, null); + // sendUpdate(sessionRef, cmdId, "{'msg' : '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 1b7a71f35e..fec81eb238 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 @@ -37,5 +37,6 @@ public enum WsCmdType { ENTITY_DATA_UNSUBSCRIBE, ENTITY_COUNT_UNSUBSCRIBE, NOTIFICATIONS_UNSUBSCRIBE, + RPC, ALARM_STATUS_UNSUBSCRIBE } 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 183292bb40..46d823dbd8 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; @@ -68,6 +69,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), @Type(name = "ALARM_STATUS_UNSUBSCRIBE", value = AlarmStatusUnsubscribeCmd.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 5364b5efac..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 @@ -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; @@ -65,9 +66,12 @@ public class TelemetryCmdsWrapper { private List alarmCountUnsubscribeCmds; + private List rpcCmds; + @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..7972ad77f2 --- /dev/null +++ b/application/src/main/java/org/thingsboard/server/service/ws/telemetry/cmd/v1/RpcCmd.java @@ -0,0 +1,48 @@ +/** + * 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. + * 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.common.data.id.EntityId; +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; //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() { + return WsCmdType.RPC; + } +} \ No newline at end of file