From 65032092e17a96f3a096359cab930ec41799bf02 Mon Sep 17 00:00:00 2001 From: Andrii Shvaika Date: Fri, 9 Jul 2021 17:07:31 +0300 Subject: [PATCH] Persistent RPC calls review --- .../actors/device/DeviceActorMessageProcessor.java | 9 +++------ .../org/thingsboard/server/controller/RpcController.java | 9 +++++---- .../service/rpc/DefaultTbCoreDeviceRpcService.java | 2 +- .../thingsboard/server/common/data/DataConstants.java | 1 + .../rule/engine/rpc/TbSendRPCRequestNode.java | 2 +- 5 files changed, 11 insertions(+), 12 deletions(-) diff --git a/application/src/main/java/org/thingsboard/server/actors/device/DeviceActorMessageProcessor.java b/application/src/main/java/org/thingsboard/server/actors/device/DeviceActorMessageProcessor.java index ec4b79408b..2c30ae863a 100644 --- a/application/src/main/java/org/thingsboard/server/actors/device/DeviceActorMessageProcessor.java +++ b/application/src/main/java/org/thingsboard/server/actors/device/DeviceActorMessageProcessor.java @@ -206,13 +206,13 @@ class DeviceActorMessageProcessor extends AbstractContextAwareMsgProcessor { syncSessionSet.forEach(rpcSubscriptions::remove); } - if (persisted && !(sent || request.isOneway())) { + if (persisted) { ObjectNode response = JacksonUtil.newObjectNode(); response.put("rpcId", request.getId().toString()); systemContext.getTbCoreDeviceRpcService().processRpcResponseFromDeviceActor(new FromDeviceRpcResponse(msg.getMsg().getId(), JacksonUtil.toString(response), null)); } - if (request.isOneway() && sent) { + if (!persisted && request.isOneway() && sent) { log.debug("[{}] Rpc command response sent [{}]!", deviceId, request.getId()); systemContext.getTbCoreDeviceRpcService().processRpcResponseFromDeviceActor(new FromDeviceRpcResponse(msg.getMsg().getId(), null, null)); } else { @@ -298,7 +298,7 @@ class DeviceActorMessageProcessor extends AbstractContextAwareMsgProcessor { toDeviceRpcPendingMap.entrySet().stream().findFirst().ifPresent(processPendingRpc(context, sessionId, sessionInfo.getNodeId(), sentOneWayIds)); } - sentOneWayIds.forEach(toDeviceRpcPendingMap::remove); + sentOneWayIds.stream().filter(id -> !toDeviceRpcPendingMap.get(id).getMsg().getMsg().isPersisted()).forEach(toDeviceRpcPendingMap::remove); } private Consumer> processPendingRpc(TbActorCtx context, UUID sessionId, String nodeId, Set sentOneWayIds) { @@ -503,9 +503,6 @@ class DeviceActorMessageProcessor extends AbstractContextAwareMsgProcessor { } } else { log.debug("[{}] Rpc command response [{}] is stale!", deviceId, responseMsg.getRequestId()); - if (requestMd.getMsg().getMsg().isPersisted()) { - systemContext.getTbRpcService().save(tenantId, new RpcId(requestMd.getMsg().getMsg().getId()), RpcStatus.FAILED, JacksonUtil.toJsonNode(responseMsg.getPayload())); - } } } diff --git a/application/src/main/java/org/thingsboard/server/controller/RpcController.java b/application/src/main/java/org/thingsboard/server/controller/RpcController.java index eae26542fa..50ab5c7863 100644 --- a/application/src/main/java/org/thingsboard/server/controller/RpcController.java +++ b/application/src/main/java/org/thingsboard/server/controller/RpcController.java @@ -34,6 +34,7 @@ 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.rule.engine.api.RpcError; +import org.thingsboard.server.common.data.DataConstants; import org.thingsboard.server.common.data.audit.ActionType; import org.thingsboard.server.common.data.exception.ThingsboardErrorCode; import org.thingsboard.server.common.data.exception.ThingsboardException; @@ -100,7 +101,7 @@ public class RpcController extends BaseController { } @PreAuthorize("hasAnyAuthority('TENANT_ADMIN', 'CUSTOMER_USER')") - @RequestMapping(value = "/persisted/{rpcId}", method = RequestMethod.GET) + @RequestMapping(value = "/persistent/{rpcId}", method = RequestMethod.GET) @ResponseBody public Rpc getPersistedRpc(@PathVariable("rpcId") String strRpc) throws ThingsboardException { checkParameter("RpcId", strRpc); @@ -113,7 +114,7 @@ public class RpcController extends BaseController { } @PreAuthorize("hasAnyAuthority('TENANT_ADMIN', 'CUSTOMER_USER')") - @RequestMapping(value = "/persisted/{deviceId}", method = RequestMethod.GET) + @RequestMapping(value = "/persistent/device/{deviceId}", method = RequestMethod.GET) @ResponseBody public PageData getPersistedRpcByDevice(@PathVariable("deviceId") String strDeviceId, @RequestParam int pageSize, @@ -134,7 +135,7 @@ public class RpcController extends BaseController { } @PreAuthorize("hasAnyAuthority('TENANT_ADMIN')") - @RequestMapping(value = "/persisted/{rpcId}", method = RequestMethod.DELETE) + @RequestMapping(value = "/persistent/{rpcId}", method = RequestMethod.DELETE) @ResponseBody public void deleteResource(@PathVariable("rpcId") String strRpc) throws ThingsboardException { checkParameter("RpcId", strRpc); @@ -155,7 +156,7 @@ public class RpcController extends BaseController { long timeout = rpcRequestBody.has("timeout") ? rpcRequestBody.get("timeout").asLong() : defaultTimeout; long expTime = System.currentTimeMillis() + Math.max(minTimeout, timeout); UUID rpcRequestUUID = rpcRequestBody.has("requestUUID") ? UUID.fromString(rpcRequestBody.get("requestUUID").asText()) : UUID.randomUUID(); - boolean persisted = rpcRequestBody.has("persisted") && rpcRequestBody.get("persisted").asBoolean(); + boolean persisted = rpcRequestBody.has(DataConstants.PERSISTENT) && rpcRequestBody.get(DataConstants.PERSISTENT).asBoolean(); accessValidator.validate(currentUser, Operation.RPC_CALL, deviceId, new HttpValidationCallback(response, new FutureCallback>() { @Override public void onSuccess(@Nullable DeferredResult result) { diff --git a/application/src/main/java/org/thingsboard/server/service/rpc/DefaultTbCoreDeviceRpcService.java b/application/src/main/java/org/thingsboard/server/service/rpc/DefaultTbCoreDeviceRpcService.java index a69402201a..1ef8d9361e 100644 --- a/application/src/main/java/org/thingsboard/server/service/rpc/DefaultTbCoreDeviceRpcService.java +++ b/application/src/main/java/org/thingsboard/server/service/rpc/DefaultTbCoreDeviceRpcService.java @@ -157,7 +157,7 @@ public class DefaultTbCoreDeviceRpcService implements TbCoreDeviceRpcService { metaData.putValue("originServiceId", serviceId); metaData.putValue("expirationTime", Long.toString(msg.getExpirationTime())); metaData.putValue("oneway", Boolean.toString(msg.isOneway())); - metaData.putValue("persisted", Boolean.toString(msg.isPersisted())); + metaData.putValue(DataConstants.PERSISTENT, Boolean.toString(msg.isPersisted())); Device device = deviceService.findDeviceById(msg.getTenantId(), msg.getDeviceId()); if (device != null) { diff --git a/common/data/src/main/java/org/thingsboard/server/common/data/DataConstants.java b/common/data/src/main/java/org/thingsboard/server/common/data/DataConstants.java index efc81846d5..3b25bd1290 100644 --- a/common/data/src/main/java/org/thingsboard/server/common/data/DataConstants.java +++ b/common/data/src/main/java/org/thingsboard/server/common/data/DataConstants.java @@ -35,6 +35,7 @@ public class DataConstants { public static final String IS_CLEARED_ALARM = "isClearedAlarm"; public static final String ALARM_CONDITION_REPEATS = "alarmConditionRepeats"; public static final String ALARM_CONDITION_DURATION = "alarmConditionDuration"; + public static final String PERSISTENT = "persistent"; public static final String[] allScopes() { return new String[]{CLIENT_SCOPE, SHARED_SCOPE, SERVER_SCOPE}; diff --git a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/rpc/TbSendRPCRequestNode.java b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/rpc/TbSendRPCRequestNode.java index 1867ae4805..3d66c569d8 100644 --- a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/rpc/TbSendRPCRequestNode.java +++ b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/rpc/TbSendRPCRequestNode.java @@ -81,7 +81,7 @@ public class TbSendRPCRequestNode implements TbNode { tmp = msg.getMetaData().getValue("oneway"); boolean oneway = !StringUtils.isEmpty(tmp) && Boolean.parseBoolean(tmp); - tmp = msg.getMetaData().getValue("persisted"); + tmp = msg.getMetaData().getValue(DataConstants.PERSISTENT); boolean persisted = !StringUtils.isEmpty(tmp) && Boolean.parseBoolean(tmp); tmp = msg.getMetaData().getValue("requestUUID");