Merge branch '20230615' into 'master_dev'

refactor: RPC命令下发兼容，定时任务和接口 See merge request yunteng/thingskit!198

Merge branch '20230615' into 'master_dev'
refactor: RPC命令下发兼容，定时任务和接口 See merge request yunteng/thingskit!198
xp.Huang
2 parents 40685e07 7ab1f566
Showing 2 changed files with 37 additions and 25 deletions
application/src/main/java/org/thingsboard/server/controller/AbstractRpcController.java
application/src/main/java/org/thingsboard/server/service/rpc/DefaultTbCoreDeviceRpcService.java
--- a/application/src/main/java/org/thingsboard/server/controller/AbstractRpcController.java
View file @f9e06b6
+++ b/application/src/main/java/org/thingsboard/server/controller/AbstractRpcController.java
View file @f9e06b6
@@ -71,8 +71,6 @@ public abstract class AbstractRpcController extends BaseController {
     protected AccessValidator accessValidator;
-    @Autowired
-    protected TkDeviceService tkDeviceService;
     @Value("${server.rest.server_side_rpc.min_timeout:5000}")
     protected long minTimeout;
@@ -83,42 +81,24 @@ public abstract class AbstractRpcController extends BaseController {
     protected DeferredResult<ResponseEntity> handleDeviceRPCRequest(boolean oneWay, DeviceId deviceId, String requestBody, HttpStatus timeoutStatus, HttpStatus noActiveConnectionStatus) throws ThingsboardException {
         try {
             JsonNode rpcRequestBody = JacksonUtil.toJsonNode(requestBody);
+            ToDeviceRpcRequestBody body = new ToDeviceRpcRequestBody(rpcRequestBody.get("method").asText(), JacksonUtil.toString(rpcRequestBody.get("params")));
             SecurityUser currentUser = getCurrentUser();
-
-            //Thingskit function
-            DeviceDTO targetDevice = tkDeviceService.findDeviceInfoByTbDeviceId(currentUser.getCurrentTenantId(), deviceId.getId().toString());
-            ObjectNode additional = (ObjectNode) rpcRequestBody.get(DataConstants.ADDITIONAL_INFO);
-            if(additional == null){
-                additional = JacksonUtil.newObjectNode();
-            }
-            if(!additional.has(FastIotConstants.Rpc.TARGET_ID)){
-                additional.put(FastIotConstants.Rpc.TARGET_ID, deviceId.getId().toString());
-            }
-            String additionalInfo =  JacksonUtil.toString(additional);
-            String methodName = rpcRequestBody.get(FastIotConstants.Rpc.METHOD_NAME).asText();
-            JsonNode params = rpcRequestBody.get(FastIotConstants.Rpc.PARAMS_NAME);
-            if(DeviceTypeEnum.SENSOR == targetDevice.getDeviceType() && params!=null && !params.isTextual()){
-                ObjectNode methodParams = (ObjectNode) rpcRequestBody.get(FastIotConstants.Rpc.PARAMS_NAME);
-                methodParams.put(FastIotConstants.Rpc.TARGET_NAME,targetDevice.getName());
-            }
-            DeviceId realDevice = DeviceTypeEnum.SENSOR == targetDevice.getDeviceType()?new DeviceId(UUID.fromString(targetDevice.getGatewayId())):deviceId;
-
-            ToDeviceRpcRequestBody body = new ToDeviceRpcRequestBody(methodName, JacksonUtil.toString(params));
             TenantId tenantId = currentUser.getTenantId();
             final DeferredResult<ResponseEntity> response = new DeferredResult<>();
             long timeout = rpcRequestBody.has(DataConstants.TIMEOUT) ? rpcRequestBody.get(DataConstants.TIMEOUT).asLong() : defaultTimeout;
             long expTime = rpcRequestBody.has(DataConstants.EXPIRATION_TIME) ? rpcRequestBody.get(DataConstants.EXPIRATION_TIME).asLong() : System.currentTimeMillis() + Math.max(minTimeout, timeout);
             UUID rpcRequestUUID = rpcRequestBody.has("requestUUID") ? UUID.fromString(rpcRequestBody.get("requestUUID").asText()) : UUID.randomUUID();
             boolean persisted = rpcRequestBody.has(DataConstants.PERSISTENT) && rpcRequestBody.get(DataConstants.PERSISTENT).asBoolean();
-
+            String additionalInfo =  JacksonUtil.toString(rpcRequestBody.get(DataConstants.ADDITIONAL_INFO));
             Integer retries = rpcRequestBody.has(DataConstants.RETRIES) ? rpcRequestBody.get(DataConstants.RETRIES).asInt() : null;
-            accessValidator.validate(currentUser, Operation.RPC_CALL, realDevice, new HttpValidationCallback(response, new FutureCallback<>() {
+
+            accessValidator.validate(currentUser, Operation.RPC_CALL, deviceId, new HttpValidationCallback(response, new FutureCallback<>() {
                 @Override
                 public void onSuccess(@Nullable DeferredResult<ResponseEntity> result) {
                     ToDeviceRpcRequest rpcRequest = new ToDeviceRpcRequest(rpcRequestUUID,
                             tenantId,
-                            realDevice,
+                            deviceId,
                             oneWay,
                             expTime,
                             body,
--- a/application/src/main/java/org/thingsboard/server/service/rpc/DefaultTbCoreDeviceRpcService.java
View file @f9e06b6
+++ b/application/src/main/java/org/thingsboard/server/service/rpc/DefaultTbCoreDeviceRpcService.java
View file @f9e06b6
@@ -21,19 +21,27 @@ import com.fasterxml.jackson.databind.node.ObjectNode;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Service;
+import org.thingsboard.common.util.JacksonUtil;
 import org.thingsboard.common.util.ThingsBoardThreadFactory;
 import org.thingsboard.server.actors.ActorSystemContext;
 import org.thingsboard.server.cluster.TbClusterService;
 import org.thingsboard.server.common.data.DataConstants;
 import org.thingsboard.server.common.data.Device;
+import org.thingsboard.server.common.data.StringUtils;
 import org.thingsboard.server.common.data.User;
+import org.thingsboard.server.common.data.id.DeviceId;
 import org.thingsboard.server.common.data.rpc.RpcError;
+import org.thingsboard.server.common.data.rpc.ToDeviceRpcRequestBody;
+import org.thingsboard.server.common.data.yunteng.constant.FastIotConstants;
+import org.thingsboard.server.common.data.yunteng.dto.DeviceDTO;
+import org.thingsboard.server.common.data.yunteng.enums.DeviceTypeEnum;
 import org.thingsboard.server.common.msg.TbMsg;
 import org.thingsboard.server.common.msg.TbMsgDataType;
 import org.thingsboard.server.common.msg.TbMsgMetaData;
 import org.thingsboard.server.common.msg.rpc.FromDeviceRpcResponse;
 import org.thingsboard.server.common.msg.rpc.ToDeviceRpcRequest;
 import org.thingsboard.server.dao.device.DeviceService;
+import org.thingsboard.server.dao.yunteng.service.TkDeviceService;
 import org.thingsboard.server.queue.discovery.TbServiceInfoProvider;
 import org.thingsboard.server.queue.util.TbCoreComponent;
 import org.thingsboard.server.service.security.model.SecurityUser;
@@ -71,6 +79,8 @@ public class DefaultTbCoreDeviceRpcService implements TbCoreDeviceRpcService {
     private ScheduledExecutorService scheduler;
     private String serviceId;
+    @Autowired
+    protected TkDeviceService tkDeviceService;
     public DefaultTbCoreDeviceRpcService(DeviceService deviceService, TbClusterService clusterService, TbServiceInfoProvider serviceInfoProvider,
                                          ActorSystemContext actorContext) {
         this.deviceService = deviceService;
@@ -100,6 +110,28 @@ public class DefaultTbCoreDeviceRpcService implements TbCoreDeviceRpcService {
     @Override
     public void processRestApiRpcRequest(ToDeviceRpcRequest request, Consumer<FromDeviceRpcResponse> responseConsumer, SecurityUser currentUser) {
         log.trace("[{}][{}] Processing REST API call to rule engine [{}]", request.getTenantId(), request.getId(), request.getDeviceId());
+
+        //Thingskit function 修改RPC三要素：设备ID、参数、附加信息
+        DeviceId targetId = request.getDeviceId();
+        DeviceDTO targetDevice = tkDeviceService.findDeviceInfoByTbDeviceId(currentUser.getCurrentTenantId(), targetId.getId().toString());
+        DeviceId netEnableId = DeviceTypeEnum.SENSOR == targetDevice.getDeviceType()?new DeviceId(UUID.fromString(targetDevice.getGatewayId())):targetId;
+        ToDeviceRpcRequestBody body = request.getBody();
+        String paramStr = body.getParams();
+        if(DeviceTypeEnum.SENSOR == targetDevice.getDeviceType() && paramStr.contains("}")){
+            ObjectNode methodParams = (ObjectNode) JacksonUtil.toJsonNode(paramStr);
+            methodParams.put(FastIotConstants.Rpc.TARGET_NAME,targetDevice.getName());
+            body = new ToDeviceRpcRequestBody(body.getMethod(),JacksonUtil.toString(methodParams));
+        }
+
+        ObjectNode additional = JacksonUtil.newObjectNode();
+        if(StringUtils.isNotBlank(request.getAdditionalInfo())){
+            additional = (ObjectNode) JacksonUtil.toJsonNode(request.getAdditionalInfo());
+        }
+        if(!additional.has(FastIotConstants.Rpc.TARGET_ID)){
+            additional.put(FastIotConstants.Rpc.TARGET_ID, targetId.getId().toString());
+        }
+        request = new ToDeviceRpcRequest(request.getId(),request.getTenantId(),netEnableId,request.isOneway(),request.getExpirationTime(),body,request.isPersisted(),request.getRetries(),JacksonUtil.toString(additional));
+
         UUID requestId = request.getId();
         localToRuleEngineRpcRequests.put(requestId, responseConsumer);
         sendRpcRequestToRuleEngine(request, currentUser);