added removing rpc request from pending map after rpc deleting

YevhenBondarenko · Andrew Shvayka
1 parent 9093206b
Showing 13 changed files with 103 additions and 4 deletions
application/src/main/java/org/thingsboard/server/actors/app/AppActor.java
application/src/main/java/org/thingsboard/server/actors/device/DeviceActor.java
application/src/main/java/org/thingsboard/server/actors/device/DeviceActorMessageProcessor.java
application/src/main/java/org/thingsboard/server/actors/tenant/TenantActor.java
application/src/main/java/org/thingsboard/server/controller/AbstractRpcController.java
application/src/main/java/org/thingsboard/server/controller/RpcV2Controller.java
application/src/main/java/org/thingsboard/server/service/rpc/DefaultTbCoreDeviceRpcService.java
application/src/main/java/org/thingsboard/server/service/rpc/RemoveRpcActorMsg.java
application/src/main/java/org/thingsboard/server/service/rpc/TbCoreDeviceRpcService.java
common/data/src/main/java/org/thingsboard/server/common/data/DataConstants.java
common/message/src/main/java/org/thingsboard/server/common/msg/MsgType.java
rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/filter/TbMsgTypeSwitchNode.java
rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/rpc/TbSendRPCRequestNode.java
--- a/application/src/main/java/org/thingsboard/server/actors/app/AppActor.java
View file @d44adcf
+++ b/application/src/main/java/org/thingsboard/server/actors/app/AppActor.java
View file @d44adcf
@@ -95,6 +95,7 @@ public class AppActor extends ContextAwareActor {
             case DEVICE_RPC_REQUEST_TO_DEVICE_ACTOR_MSG:
             case DEVICE_RPC_RESPONSE_TO_DEVICE_ACTOR_MSG:
             case SERVER_RPC_RESPONSE_TO_DEVICE_ACTOR_MSG:
+            case REMOVE_RPC_TO_DEVICE_ACTOR_MSG:
                 onToDeviceActorMsg((TenantAwareMsg) msg, true);
                 break;
             case EDGE_EVENT_UPDATE_TO_EDGE_SESSION_MSG:
--- a/application/src/main/java/org/thingsboard/server/actors/device/DeviceActor.java
View file @d44adcf
+++ b/application/src/main/java/org/thingsboard/server/actors/device/DeviceActor.java
View file @d44adcf
@@ -28,6 +28,7 @@ import org.thingsboard.server.common.data.id.TenantId;
 import org.thingsboard.server.common.msg.TbActorMsg;
 import org.thingsboard.server.common.msg.timeout.DeviceActorServerSideRpcTimeoutMsg;
 import org.thingsboard.server.service.rpc.FromDeviceRpcResponseActorMsg;
+import org.thingsboard.server.service.rpc.RemoveRpcActorMsg;
 import org.thingsboard.server.service.rpc.ToDeviceRpcRequestActorMsg;
 import org.thingsboard.server.service.transport.msg.TransportToDeviceActorMsgWrapper;
@@ -84,6 +85,9 @@ public class DeviceActor extends ContextAwareActor {
             case DEVICE_EDGE_UPDATE_TO_DEVICE_ACTOR_MSG:
                 processor.processEdgeUpdate((DeviceEdgeUpdateMsg) msg);
                 break;
+            case REMOVE_RPC_TO_DEVICE_ACTOR_MSG:
+                processor.processRemoveRpc(ctx, (RemoveRpcActorMsg) msg);
+                break;
             default:
                 return false;
         }
--- a/application/src/main/java/org/thingsboard/server/actors/device/DeviceActorMessageProcessor.java
View file @d44adcf
+++ b/application/src/main/java/org/thingsboard/server/actors/device/DeviceActorMessageProcessor.java
View file @d44adcf
@@ -88,6 +88,7 @@ import org.thingsboard.server.gen.transport.TransportProtos.TransportToDeviceAct
 import org.thingsboard.server.gen.transport.TransportProtos.TsKvProto;
 import org.thingsboard.server.common.msg.rpc.FromDeviceRpcResponse;
 import org.thingsboard.server.service.rpc.FromDeviceRpcResponseActorMsg;
+import org.thingsboard.server.service.rpc.RemoveRpcActorMsg;
 import org.thingsboard.server.service.rpc.ToDeviceRpcRequestActorMsg;
 import org.thingsboard.server.service.transport.msg.TransportToDeviceActorMsgWrapper;
@@ -263,6 +264,21 @@ class DeviceActorMessageProcessor extends AbstractContextAwareMsgProcessor {
         }
     }
+    void processRemoveRpc(TbActorCtx context, RemoveRpcActorMsg msg) {
+        log.debug("[{}] Processing remove rpc command", msg.getRequestId());
+        Integer requestId = null;
+        for (Map.Entry<Integer, ToDeviceRpcRequestMetadata> entry : toDeviceRpcPendingMap.entrySet()) {
+            if (entry.getValue().getMsg().getMsg().getId().equals(msg.getRequestId())) {
+                requestId = entry.getKey();
+                break;
+            }
+        }
+
+        if (requestId != null) {
+            toDeviceRpcPendingMap.remove(requestId);
+        }
+    }
+
     private void registerPendingRpcRequest(TbActorCtx context, ToDeviceRpcRequestActorMsg msg, boolean sent, ToDeviceRpcRequestMsg rpcRequest, long timeout) {
         toDeviceRpcPendingMap.put(rpcRequest.getRequestId(), new ToDeviceRpcRequestMetadata(msg, sent));
         DeviceActorServerSideRpcTimeoutMsg timeoutMsg = new DeviceActorServerSideRpcTimeoutMsg(rpcRequest.getRequestId(), timeout);
--- a/application/src/main/java/org/thingsboard/server/actors/tenant/TenantActor.java
View file @d44adcf
+++ b/application/src/main/java/org/thingsboard/server/actors/tenant/TenantActor.java
View file @d44adcf
@@ -165,6 +165,7 @@ public class TenantActor extends RuleChainManagerActor {
             case DEVICE_RPC_REQUEST_TO_DEVICE_ACTOR_MSG:
             case DEVICE_RPC_RESPONSE_TO_DEVICE_ACTOR_MSG:
             case SERVER_RPC_RESPONSE_TO_DEVICE_ACTOR_MSG:
+            case REMOVE_RPC_TO_DEVICE_ACTOR_MSG:
                 onToDeviceActorMsg((DeviceAwareMsg) msg, true);
                 break;
             case RULE_CHAIN_TO_RULE_CHAIN_MSG:
--- a/application/src/main/java/org/thingsboard/server/controller/AbstractRpcController.java
View file @d44adcf
+++ b/application/src/main/java/org/thingsboard/server/controller/AbstractRpcController.java
View file @d44adcf
@@ -57,7 +57,7 @@ import java.util.UUID;
 public abstract class AbstractRpcController extends BaseController {
     @Autowired
-    private TbCoreDeviceRpcService deviceRpcService;
+    protected TbCoreDeviceRpcService deviceRpcService;
     @Autowired
     private AccessValidator accessValidator;
--- a/application/src/main/java/org/thingsboard/server/controller/RpcV2Controller.java
View file @d44adcf
+++ b/application/src/main/java/org/thingsboard/server/controller/RpcV2Controller.java
View file @d44adcf
@@ -27,6 +27,7 @@ import org.springframework.web.bind.annotation.RequestParam;
 import org.springframework.web.bind.annotation.ResponseBody;
 import org.springframework.web.bind.annotation.RestController;
 import org.springframework.web.context.request.async.DeferredResult;
+import org.thingsboard.common.util.JacksonUtil;
 import org.thingsboard.server.common.data.exception.ThingsboardException;
 import org.thingsboard.server.common.data.id.DeviceId;
 import org.thingsboard.server.common.data.id.RpcId;
@@ -35,11 +36,16 @@ import org.thingsboard.server.common.data.page.PageData;
 import org.thingsboard.server.common.data.page.PageLink;
 import org.thingsboard.server.common.data.rpc.Rpc;
 import org.thingsboard.server.common.data.rpc.RpcStatus;
+import org.thingsboard.server.common.msg.TbMsg;
+import org.thingsboard.server.common.msg.TbMsgMetaData;
 import org.thingsboard.server.queue.util.TbCoreComponent;
+import org.thingsboard.server.service.rpc.RemoveRpcActorMsg;
 import org.thingsboard.server.service.security.permission.Operation;
 import java.util.UUID;
+import static org.thingsboard.server.common.data.DataConstants.RPC_DELETED;
+
 @RestController
 @TbCoreComponent
 @RequestMapping(TbUrlConstants.RPC_V2_URL_PREFIX)
@@ -100,7 +106,21 @@ public class RpcV2Controller extends AbstractRpcController {
     public void deleteResource(@PathVariable("rpcId") String strRpc) throws ThingsboardException {
         checkParameter("RpcId", strRpc);
         try {
-            rpcService.deleteRpc(getTenantId(), new RpcId(UUID.fromString(strRpc)));
+            RpcId rpcId = new RpcId(UUID.fromString(strRpc));
+            Rpc rpc = checkRpcId(rpcId, Operation.DELETE);
+
+            if (rpc != null) {
+                if (rpc.getStatus().equals(RpcStatus.QUEUED)) {
+                    RemoveRpcActorMsg removeMsg = new RemoveRpcActorMsg(getTenantId(), rpc.getDeviceId(), rpc.getUuidId());
+                    log.trace("[{}] Forwarding msg {} to queue actor!", rpc.getDeviceId(), rpc);
+                    tbClusterService.pushMsgToCore(removeMsg, null);
+                }
+
+                rpcService.deleteRpc(getTenantId(), rpcId);
+
+                TbMsg msg = TbMsg.newMsg(RPC_DELETED, rpc.getDeviceId(), TbMsgMetaData.EMPTY, JacksonUtil.toString(rpc));
+                tbClusterService.pushMsgToRuleEngine(getTenantId(), rpc.getDeviceId(), msg, null);
+            }
         } catch (Exception e) {
             throw handleException(e);
         }
--- a/application/src/main/java/org/thingsboard/server/service/rpc/DefaultTbCoreDeviceRpcService.java
View file @d44adcf
+++ b/application/src/main/java/org/thingsboard/server/service/rpc/DefaultTbCoreDeviceRpcService.java
View file @d44adcf
@@ -139,6 +139,12 @@ public class DefaultTbCoreDeviceRpcService implements TbCoreDeviceRpcService {
         }
     }
+    @Override
+    public void processRemoveRpc(RemoveRpcActorMsg removeRpcMsg) {
+        log.trace("[{}][{}] Processing remove RPC [{}]", removeRpcMsg.getTenantId(), removeRpcMsg.getRequestId(), removeRpcMsg.getDeviceId());
+        actorContext.tellWithHighPriority(removeRpcMsg);
+    }
+
     private void sendRpcResponseToTbRuleEngine(String originServiceId, FromDeviceRpcResponse response) {
         if (serviceId.equals(originServiceId)) {
             if (tbRuleEngineRpcService.isPresent()) {
--- a/application/src/main/java/org/thingsboard/server/service/rpc/RemoveRpcActorMsg.java 0 → 100644
View file @d44adcf
+++ b/application/src/main/java/org/thingsboard/server/service/rpc/RemoveRpcActorMsg.java 0 → 100644
View file @d44adcf
+/**
+ * Copyright © 2016-2021 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.rpc;
+
+import lombok.Getter;
+import lombok.RequiredArgsConstructor;
+import lombok.ToString;
+import org.thingsboard.rule.engine.api.msg.ToDeviceActorNotificationMsg;
+import org.thingsboard.server.common.data.id.DeviceId;
+import org.thingsboard.server.common.data.id.TenantId;
+import org.thingsboard.server.common.msg.MsgType;
+
+import java.util.UUID;
+
+@ToString
+@RequiredArgsConstructor
+public class RemoveRpcActorMsg implements ToDeviceActorNotificationMsg {
+
+    @Getter
+    private final TenantId tenantId;
+    @Getter
+    private final DeviceId deviceId;
+
+    @Getter
+    private final UUID requestId;
+
+    @Override
+    public MsgType getMsgType() {
+        return MsgType.REMOVE_RPC_TO_DEVICE_ACTOR_MSG;
+    }
+}
--- a/application/src/main/java/org/thingsboard/server/service/rpc/TbCoreDeviceRpcService.java
View file @d44adcf
+++ b/application/src/main/java/org/thingsboard/server/service/rpc/TbCoreDeviceRpcService.java
View file @d44adcf
@@ -56,4 +56,6 @@ public interface TbCoreDeviceRpcService {
      */
     void processRpcResponseFromDeviceActor(FromDeviceRpcResponse response);
+    void processRemoveRpc(RemoveRpcActorMsg removeRpcMsg);
+
 }
--- a/common/data/src/main/java/org/thingsboard/server/common/data/DataConstants.java
View file @d44adcf
+++ b/common/data/src/main/java/org/thingsboard/server/common/data/DataConstants.java
View file @d44adcf
@@ -89,6 +89,7 @@ public class DataConstants {
     public static final String RPC_SUCCESSFUL = "RPC_SUCCESSFUL";
     public static final String RPC_TIMEOUT = "RPC_TIMEOUT";
     public static final String RPC_FAILED = "RPC_FAILED";
+    public static final String RPC_DELETED = "RPC_DELETED";
     public static final String DEFAULT_SECRET_KEY = "";
     public static final String SECRET_KEY_FIELD_NAME = "secretKey";
--- a/common/message/src/main/java/org/thingsboard/server/common/msg/MsgType.java
View file @d44adcf
+++ b/common/message/src/main/java/org/thingsboard/server/common/msg/MsgType.java
View file @d44adcf
@@ -92,6 +92,8 @@ public enum MsgType {
     DEVICE_ACTOR_SERVER_SIDE_RPC_TIMEOUT_MSG,
+    REMOVE_RPC_TO_DEVICE_ACTOR_MSG,
+
     /**
      * Message that is sent from the Device Actor to Rule Engine. Requires acknowledgement
      */
--- a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/filter/TbMsgTypeSwitchNode.java
View file @d44adcf
+++ b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/filter/TbMsgTypeSwitchNode.java
View file @d44adcf
@@ -33,7 +33,7 @@ import org.thingsboard.server.common.msg.session.SessionMsgType;
         type = ComponentType.FILTER,
         name = "message type switch",
         configClazz = EmptyNodeConfiguration.class,
-        relationTypes = {"Post attributes", "Post telemetry", "RPC Request from Device", "RPC Request to Device", "RPC Queued", "RPC Delivered", "RPC Successful", "RPC Timeout", "RPC Failed",
+        relationTypes = {"Post attributes", "Post telemetry", "RPC Request from Device", "RPC Request to Device", "RPC Queued", "RPC Delivered", "RPC Successful", "RPC Timeout", "RPC Failed", "RPC Deleted",
                 "Activity Event", "Inactivity Event", "Connect Event", "Disconnect Event", "Entity Created", "Entity Updated", "Entity Deleted", "Entity Assigned",
                 "Entity Unassigned", "Attributes Updated", "Attributes Deleted", "Alarm Acknowledged", "Alarm Cleared", "Other", "Entity Assigned From Tenant", "Entity Assigned To Tenant",
                 "Timeseries Updated", "Timeseries Deleted"},
@@ -105,6 +105,8 @@ public class TbMsgTypeSwitchNode implements TbNode {
             relationType = "RPC Timeout";
         } else if (msg.getType().equals(DataConstants.RPC_FAILED)) {
             relationType = "RPC Failed";
+        } else if (msg.getType().equals(DataConstants.RPC_DELETED)) {
+            relationType = "RPC Deleted";
         } else {
             relationType = "Other";
         }
--- a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/rpc/TbSendRPCRequestNode.java
View file @d44adcf
+++ b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/rpc/TbSendRPCRequestNode.java
View file @d44adcf
@@ -118,7 +118,7 @@ public class TbSendRPCRequestNode implements TbNode {
                     .build();
             ctx.getRpcService().sendRpcRequestToDevice(request, ruleEngineDeviceRpcResponse -> {
-                if (!ruleEngineDeviceRpcResponse.getError().isPresent()) {
+                if (ruleEngineDeviceRpcResponse.getError().isEmpty()) {
                     TbMsg next = ctx.newMsg(msg.getQueueName(), msg.getType(), msg.getOriginator(), msg.getCustomerId(), msg.getMetaData(), ruleEngineDeviceRpcResponse.getResponse().orElse("{}"));
                     ctx.enqueueForTellNext(next, TbRelationTypes.SUCCESS);
                 } else {