tsai
/
thingsboard
mirror of https://github.com/thingsboard/thingsboard.git
1
0
Fork 0
Code Issues Projects Releases Wiki Activity
Browse Source

Merge pull request #9952 from lsyer/wsRPC 

added sending RPC(Server-side RPC) via websocket.
feature/websocket-rpc
Viacheslav Klimov 2 years ago
committed by GitHub
parent
a411213d6c 778c5e2d00
commit
6addc01ee9
No known key found for this signature in database GPG Key ID: B5690EEEBB952194
5 changed files with 192 additions and 0 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Split View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 137
      application/src/main/java/org/thingsboard/server/service/ws/DefaultWebSocketService.java
  2. 1
      application/src/main/java/org/thingsboard/server/service/ws/WsCmdType.java
  3. 2
      application/src/main/java/org/thingsboard/server/service/ws/WsCommandsWrapper.java
  4. 4
      application/src/main/java/org/thingsboard/server/service/ws/telemetry/cmd/TelemetryCmdsWrapper.java
  5. 48
      application/src/main/java/org/thingsboard/server/service/ws/telemetry/cmd/v1/RpcCmd.java

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

@ -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<WsCmdType, WsCmdHandler<? extends WsCmd>> 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<ValidationResult> 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> 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<String> 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> 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> 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;

1
application/src/main/java/org/thingsboard/server/service/ws/WsCmdType.java
View File

@ -37,5 +37,6 @@ public enum WsCmdType {
ENTITY_DATA_UNSUBSCRIBE,
ENTITY_COUNT_UNSUBSCRIBE,
NOTIFICATIONS_UNSUBSCRIBE,
RPC,
ALARM_STATUS_UNSUBSCRIBE
}

2
application/src/main/java/org/thingsboard/server/service/ws/WsCommandsWrapper.java
View File

@ -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<WsCmd> cmds;

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

@ -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<AlarmCountUnsubscribeCmd> alarmCountUnsubscribeCmds;
private List<RpcCmd> rpcCmds;
@JsonIgnore
public WsCommandsWrapper toCommonCmdsWrapper() {
return new WsCommandsWrapper(null, Stream.of(
rpcCmds,
attrSubCmds, tsSubCmds, historyCmds, entityDataCmds,
entityDataUnsubscribeCmds, alarmDataCmds, alarmDataUnsubscribeCmds,
entityCountCmds, entityCountUnsubscribeCmds,

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

@ -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;
}
}
Write Preview
Loading…
Cancel
Save