added sequence for the all RPC

YevhenBondarenko
1 parent 8869dc0c
Showing 12 changed files with 105 additions and 85 deletions
application/src/main/java/org/thingsboard/server/actors/device/DeviceActorMessageProcessor.java
application/src/main/java/org/thingsboard/server/controller/AbstractRpcController.java
application/src/main/java/org/thingsboard/server/service/rpc/DefaultTbRuleEngineRpcService.java
application/src/test/java/org/thingsboard/server/transport/mqtt/rpc/AbstractMqttServerSideRpcIntegrationTest.java
common/cluster-api/src/main/proto/queue.proto
common/data/src/main/java/org/thingsboard/server/common/data/DataConstants.java
common/message/src/main/java/org/thingsboard/server/common/msg/rpc/ToDeviceRpcRequest.java
common/transport/coap/src/main/java/org/thingsboard/server/transport/coap/client/DefaultCoapClientContext.java
common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/MqttTransportHandler.java
common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/service/DefaultTransportService.java
rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/RuleEngineDeviceRpcRequest.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/device/DeviceActorMessageProcessor.java
View file @8513c99
+++ b/application/src/main/java/org/thingsboard/server/actors/device/DeviceActorMessageProcessor.java
View file @8513c99
@@ -80,9 +80,9 @@ import org.thingsboard.server.gen.transport.TransportProtos.SessionType;
 import org.thingsboard.server.gen.transport.TransportProtos.SubscribeToAttributeUpdatesMsg;
 import org.thingsboard.server.gen.transport.TransportProtos.SubscribeToRPCMsg;
 import org.thingsboard.server.gen.transport.TransportProtos.SubscriptionInfoProto;
-import org.thingsboard.server.gen.transport.TransportProtos.ToDevicePersistedRpcResponseMsg;
 import org.thingsboard.server.gen.transport.TransportProtos.ToDeviceRpcRequestMsg;
 import org.thingsboard.server.gen.transport.TransportProtos.ToDeviceRpcResponseMsg;
+import org.thingsboard.server.gen.transport.TransportProtos.ToDeviceRpcResponseStatusMsg;
 import org.thingsboard.server.gen.transport.TransportProtos.ToServerRpcResponseMsg;
 import org.thingsboard.server.gen.transport.TransportProtos.ToTransportMsg;
 import org.thingsboard.server.gen.transport.TransportProtos.ToTransportUpdateCredentialsProto;
@@ -298,7 +298,9 @@ class DeviceActorMessageProcessor extends AbstractContextAwareMsgProcessor {
             }
             systemContext.getTbCoreDeviceRpcService().processRpcResponseFromDeviceActor(new FromDeviceRpcResponse(requestMd.getMsg().getMsg().getId(),
                     null, requestMd.isSent() ? RpcError.TIMEOUT : RpcError.NO_ACTIVE_CONNECTION));
-            sendNextPendingRequest(context);
+            if (!requestMd.isDelivered()) {
+                sendNextPendingRequest(context);
+            }
         }
     }
@@ -315,22 +317,10 @@ class DeviceActorMessageProcessor extends AbstractContextAwareMsgProcessor {
         }
         Set<Integer> sentOneWayIds = new HashSet<>();
-        if (sessionType == SessionType.ASYNC) {
-            if (rpcSequenceEnabled) {
-                List<Map.Entry<Integer, ToDeviceRpcRequestMetadata>> entries = new ArrayList<>();
-                for (Map.Entry<Integer, ToDeviceRpcRequestMetadata> entry : toDeviceRpcPendingMap.entrySet()) {
-                    if (entry.getValue().isDelivered()) {
-                        continue;
-                    }
-                    entries.add(entry);
-                    if (entry.getValue().getMsg().getMsg().isPersisted() || entry.getValue().getMsg().getMsg().isOneway()) {
-                        break;
-                    }
-                }
-                entries.forEach(processPendingRpc(context, sessionId, nodeId, sentOneWayIds));
-            } else {
-                toDeviceRpcPendingMap.entrySet().forEach(processPendingRpc(context, sessionId, nodeId, sentOneWayIds));
-            }
+        if (rpcSequenceEnabled) {
+            toDeviceRpcPendingMap.entrySet().stream().filter(e -> !e.getValue().isDelivered()).findFirst().ifPresent(processPendingRpc(context, sessionId, nodeId, sentOneWayIds));
+        } else if (sessionType == SessionType.ASYNC) {
+            toDeviceRpcPendingMap.entrySet().forEach(processPendingRpc(context, sessionId, nodeId, sentOneWayIds));
         } else {
             toDeviceRpcPendingMap.entrySet().stream().findFirst().ifPresent(processPendingRpc(context, sessionId, nodeId, sentOneWayIds));
         }
@@ -348,7 +338,7 @@ class DeviceActorMessageProcessor extends AbstractContextAwareMsgProcessor {
         return entry -> {
             ToDeviceRpcRequest request = entry.getValue().getMsg().getMsg();
             ToDeviceRpcRequestBody body = request.getBody();
-            if (request.isOneway()) {
+            if (request.isOneway() && !rpcSequenceEnabled) {
                 sentOneWayIds.add(entry.getKey());
                 systemContext.getTbCoreDeviceRpcService().processRpcResponseFromDeviceActor(new FromDeviceRpcResponse(request.getId(), null, null));
             }
@@ -357,6 +347,7 @@ class DeviceActorMessageProcessor extends AbstractContextAwareMsgProcessor {
                     .setMethodName(body.getMethod())
                     .setParams(body.getParams())
                     .setExpirationTime(request.getExpirationTime())
+                    .setTimeout(request.getTimeout())
                     .setRequestIdMSB(request.getId().getMostSignificantBits())
                     .setRequestIdLSB(request.getId().getLeastSignificantBits())
                     .setOneway(request.isOneway())
@@ -395,8 +386,8 @@ class DeviceActorMessageProcessor extends AbstractContextAwareMsgProcessor {
         if (msg.hasClaimDevice()) {
             handleClaimDeviceMsg(context, sessionInfo, msg.getClaimDevice());
         }
-        if (msg.hasPersistedRpcResponseMsg()) {
-            processPersistedRpcResponses(context, sessionInfo, msg.getPersistedRpcResponseMsg());
+        if (msg.hasRpcResponseStatusMsg()) {
+            processPersistedRpcResponses(context, sessionInfo, msg.getRpcResponseStatusMsg());
         }
         if (msg.hasUplinkNotificationMsg()) {
             processUplinkNotificationMsg(context, sessionInfo, msg.getUplinkNotificationMsg());
@@ -556,27 +547,32 @@ class DeviceActorMessageProcessor extends AbstractContextAwareMsgProcessor {
         boolean success = requestMd != null;
         if (success) {
             boolean hasError = StringUtils.isNotEmpty(responseMsg.getError());
-            String payload = hasError ? responseMsg.getError() : responseMsg.getPayload();
-            systemContext.getTbCoreDeviceRpcService().processRpcResponseFromDeviceActor(
-                    new FromDeviceRpcResponse(requestMd.getMsg().getMsg().getId(),
-                            payload, null));
-            if (requestMd.getMsg().getMsg().isPersisted()) {
-                RpcStatus status = hasError ? RpcStatus.FAILED : RpcStatus.SUCCESSFUL;
-                JsonNode response;
-                try {
-                    response = JacksonUtil.toJsonNode(payload);
-                } catch (IllegalArgumentException e) {
-                    response = JacksonUtil.newObjectNode().put("error", payload);
+            try {
+                String payload = hasError ? responseMsg.getError() : responseMsg.getPayload();
+                systemContext.getTbCoreDeviceRpcService().processRpcResponseFromDeviceActor(
+                        new FromDeviceRpcResponse(requestMd.getMsg().getMsg().getId(),
+                                payload, null));
+                if (requestMd.getMsg().getMsg().isPersisted()) {
+                    RpcStatus status = hasError ? RpcStatus.FAILED : RpcStatus.SUCCESSFUL;
+                    JsonNode response;
+                    try {
+                        response = JacksonUtil.toJsonNode(payload);
+                    } catch (IllegalArgumentException e) {
+                        response = JacksonUtil.newObjectNode().put("error", payload);
+                    }
+                    systemContext.getTbRpcService().save(tenantId, new RpcId(requestMd.getMsg().getMsg().getId()), status, response);
+                }
+            } finally {
+                if (!requestMd.isDelivered() && hasError) {
+                    sendNextPendingRequest(context);
                 }
-                systemContext.getTbRpcService().save(tenantId, new RpcId(requestMd.getMsg().getMsg().getId()), status, response);
             }
-            sendNextPendingRequest(context);
         } else {
             log.debug("[{}] Rpc command response [{}] is stale!", deviceId, responseMsg.getRequestId());
         }
     }
-    private void processPersistedRpcResponses(TbActorCtx context, SessionInfoProto sessionInfo, ToDevicePersistedRpcResponseMsg responseMsg) {
+    private void processPersistedRpcResponses(TbActorCtx context, SessionInfoProto sessionInfo, ToDeviceRpcResponseStatusMsg responseMsg) {
         UUID rpcId = new UUID(responseMsg.getRequestIdMSB(), responseMsg.getRequestIdLSB());
         RpcStatus status = RpcStatus.valueOf(responseMsg.getStatus());
         ToDeviceRpcRequestMetadata md = toDeviceRpcPendingMap.get(responseMsg.getRequestId());
@@ -585,6 +581,9 @@ class DeviceActorMessageProcessor extends AbstractContextAwareMsgProcessor {
             if (status.equals(RpcStatus.DELIVERED)) {
                 if (md.getMsg().getMsg().isOneway()) {
                     toDeviceRpcPendingMap.remove(responseMsg.getRequestId());
+                    if (rpcSequenceEnabled) {
+                        systemContext.getTbCoreDeviceRpcService().processRpcResponseFromDeviceActor(new FromDeviceRpcResponse(rpcId, null, null));
+                    }
                 } else {
                     md.setDelivered(true);
                 }
@@ -597,7 +596,9 @@ class DeviceActorMessageProcessor extends AbstractContextAwareMsgProcessor {
                 }
             }
-            systemContext.getTbRpcService().save(tenantId, new RpcId(rpcId), status, null);
+            if (md.getMsg().getMsg().isPersisted()) {
+                systemContext.getTbRpcService().save(tenantId, new RpcId(rpcId), status, null);
+            }
             if (status != RpcStatus.SENT) {
                 sendNextPendingRequest(context);
             }
--- a/application/src/main/java/org/thingsboard/server/controller/AbstractRpcController.java
View file @8513c99
+++ b/application/src/main/java/org/thingsboard/server/controller/AbstractRpcController.java
View file @8513c99
@@ -75,8 +75,8 @@ public abstract class AbstractRpcController extends BaseController {
             SecurityUser currentUser = getCurrentUser();
             TenantId tenantId = currentUser.getTenantId();
             final DeferredResult<ResponseEntity> response = new DeferredResult<>();
-            long timeout = rpcRequestBody.has("timeout") ? rpcRequestBody.get("timeout").asLong() : defaultTimeout;
-            long expTime = System.currentTimeMillis() + Math.max(minTimeout, timeout);
+            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));
@@ -88,6 +88,7 @@ public abstract class AbstractRpcController extends BaseController {
                             deviceId,
                             oneWay,
                             expTime,
+                            timeout,
                             body,
                             persisted,
                             additionalInfo
--- a/application/src/main/java/org/thingsboard/server/service/rpc/DefaultTbRuleEngineRpcService.java
View file @8513c99
+++ b/application/src/main/java/org/thingsboard/server/service/rpc/DefaultTbRuleEngineRpcService.java
View file @8513c99
@@ -101,7 +101,7 @@ public class DefaultTbRuleEngineRpcService implements TbRuleEngineDeviceRpcServi
     @Override
     public void sendRpcRequestToDevice(RuleEngineDeviceRpcRequest src, Consumer<RuleEngineDeviceRpcResponse> consumer) {
         ToDeviceRpcRequest request = new ToDeviceRpcRequest(src.getRequestUUID(), src.getTenantId(), src.getDeviceId(),
-                src.isOneway(), src.getExpirationTime(), new ToDeviceRpcRequestBody(src.getMethod(), src.getBody()), src.isPersisted(), src.getAdditionalInfo());
+                src.isOneway(), src.getExpirationTime(), src.getTimeout(), new ToDeviceRpcRequestBody(src.getMethod(), src.getBody()), src.isPersisted(), src.getAdditionalInfo());
         forwardRpcRequestToDeviceActor(request, response -> {
             if (src.isRestApiCall()) {
                 sendRpcResponseToTbCore(src.getOriginServiceId(), response);
--- a/application/src/test/java/org/thingsboard/server/transport/mqtt/rpc/AbstractMqttServerSideRpcIntegrationTest.java
View file @8513c99
+++ b/application/src/test/java/org/thingsboard/server/transport/mqtt/rpc/AbstractMqttServerSideRpcIntegrationTest.java
View file @8513c99
@@ -28,15 +28,16 @@ import org.eclipse.paho.client.mqttv3.MqttCallback;
 import org.eclipse.paho.client.mqttv3.MqttException;
 import org.eclipse.paho.client.mqttv3.MqttMessage;
 import org.junit.Assert;
+import org.thingsboard.common.util.JacksonUtil;
 import org.thingsboard.server.common.data.Device;
 import org.thingsboard.server.common.data.TransportPayloadType;
 import org.thingsboard.server.common.data.device.profile.MqttTopics;
-import org.thingsboard.common.util.JacksonUtil;
 import org.thingsboard.server.transport.mqtt.AbstractMqttIntegrationTest;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.List;
+import java.util.concurrent.CopyOnWriteArrayList;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
@@ -120,12 +121,13 @@ public abstract class AbstractMqttServerSideRpcIntegrationTest extends AbstractM
         }
         MqttAsyncClient client = getMqttAsyncClient(accessToken);
+        client.setManualAcks(true);
         CountDownLatch latch = new CountDownLatch(10);
         TestSequenceMqttCallback callback = new TestSequenceMqttCallback(client, latch, result);
         client.setCallback(callback);
         client.subscribe(MqttTopics.DEVICE_RPC_REQUESTS_SUB_TOPIC, 1);
-        latch.await(30, TimeUnit.SECONDS);
+        latch.await(10, TimeUnit.SECONDS);
         Assert.assertEquals(expected, result);
     }
@@ -246,8 +248,7 @@ public abstract class AbstractMqttServerSideRpcIntegrationTest extends AbstractM
         private final MqttAsyncClient client;
         private final CountDownLatch latch;
-        private final  List<String> expected;
-        private Integer qoS;
+        private final List<String> expected;
         TestSequenceMqttCallback(MqttAsyncClient client, CountDownLatch latch, List<String> expected) {
             this.client = client;
@@ -255,10 +256,6 @@ public abstract class AbstractMqttServerSideRpcIntegrationTest extends AbstractM
             this.expected = expected;
         }
-        int getQoS() {
-            return qoS;
-        }
-
         @Override
         public void connectionLost(Throwable throwable) {
         }
@@ -268,7 +265,9 @@ public abstract class AbstractMqttServerSideRpcIntegrationTest extends AbstractM
             log.info("Message Arrived: " + Arrays.toString(mqttMessage.getPayload()));
             expected.add(new String(mqttMessage.getPayload()));
             String responseTopic = requestTopic.replace("request", "response");
-            qoS = mqttMessage.getQos();
+            var qoS = mqttMessage.getQos();
+
+            client.messageArrivedComplete(mqttMessage.getId(), qoS);
             client.publish(responseTopic, processMessageArrived(requestTopic, mqttMessage));
             latch.countDown();
         }
--- a/common/cluster-api/src/main/proto/queue.proto
View file @8513c99
+++ b/common/cluster-api/src/main/proto/queue.proto
View file @8513c99
@@ -334,6 +334,7 @@ message ToDeviceRpcRequestMsg {
   int64 requestIdLSB = 6;
   bool oneway = 7;
   bool persisted = 8;
+  int64 timeout = 9;
 }
 message ToDeviceRpcResponseMsg {
@@ -346,7 +347,7 @@ message UplinkNotificationMsg {
   int64 uplinkTs = 1;
 }
-message ToDevicePersistedRpcResponseMsg {
+message ToDeviceRpcResponseStatusMsg {
   int32 requestId = 1;
   int64 requestIdMSB = 2;
   int64 requestIdLSB = 3;
@@ -456,7 +457,7 @@ message TransportToDeviceActorMsg {
   SubscriptionInfoProto subscriptionInfo = 7;
   ClaimDeviceMsg claimDevice = 8;
   ProvisionDeviceRequestMsg provisionDevice = 9;
-  ToDevicePersistedRpcResponseMsg persistedRpcResponseMsg = 10;
+  ToDeviceRpcResponseStatusMsg rpcResponseStatusMsg = 10;
   SendPendingRPCMsg sendPendingRPC = 11;
   UplinkNotificationMsg uplinkNotificationMsg = 12;
 }
--- a/common/data/src/main/java/org/thingsboard/server/common/data/DataConstants.java
View file @8513c99
+++ b/common/data/src/main/java/org/thingsboard/server/common/data/DataConstants.java
View file @8513c99
@@ -36,6 +36,8 @@ public class DataConstants {
     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 TIMEOUT = "timeout";
+    public static final String EXPIRATION_TIME = "expirationTime";
     public static final String ADDITIONAL_INFO = "additionalInfo";
     public static final String COAP_TRANSPORT_NAME = "COAP";
     public static final String LWM2M_TRANSPORT_NAME = "LWM2M";
--- a/common/message/src/main/java/org/thingsboard/server/common/msg/rpc/ToDeviceRpcRequest.java
View file @8513c99
+++ b/common/message/src/main/java/org/thingsboard/server/common/msg/rpc/ToDeviceRpcRequest.java
View file @8513c99
@@ -34,6 +34,7 @@ public class ToDeviceRpcRequest implements Serializable {
     private final DeviceId deviceId;
     private final boolean oneway;
     private final long expirationTime;
+    private final long timeout;
     private final ToDeviceRpcRequestBody body;
     private final boolean persisted;
     @JsonIgnore
--- a/common/transport/coap/src/main/java/org/thingsboard/server/transport/coap/client/DefaultCoapClientContext.java
View file @8513c99
+++ b/common/transport/coap/src/main/java/org/thingsboard/server/transport/coap/client/DefaultCoapClientContext.java
View file @8513c99
@@ -525,17 +525,26 @@ public class DefaultCoapClientContext implements CoapClientContext {
                 Response response = state.getAdaptor().convertToPublish(conRequest, msg, state.getConfiguration().getRpcRequestDynamicMessageBuilder());
                 int requestId = getNextMsgId();
                 response.setMID(requestId);
-                if (msg.getPersisted() && conRequest) {
+                if (conRequest) {
                     transportContext.getRpcAwaitingAck().put(requestId, msg);
                     transportContext.getScheduler().schedule(() -> {
-                        transportContext.getRpcAwaitingAck().remove(requestId);
-                    }, Math.max(0, msg.getExpirationTime() - System.currentTimeMillis()), TimeUnit.MILLISECONDS);
+                        TransportProtos.ToDeviceRpcRequestMsg rpcRequestMsg = transportContext.getRpcAwaitingAck().remove(requestId);
+                        if (rpcRequestMsg != null) {
+                            transportService.process(state.getSession(), msg, RpcStatus.TIMEOUT, TransportServiceCallback.EMPTY);
+                        }
+                    }, Math.max(0, Math.min(msg.getTimeout(), msg.getExpirationTime() - System.currentTimeMillis())), TimeUnit.MILLISECONDS);
+
                     response.addMessageObserver(new TbCoapMessageObserver(requestId, id -> {
                         TransportProtos.ToDeviceRpcRequestMsg rpcRequestMsg = transportContext.getRpcAwaitingAck().remove(id);
                         if (rpcRequestMsg != null) {
                             transportService.process(state.getSession(), rpcRequestMsg, RpcStatus.DELIVERED, TransportServiceCallback.EMPTY);
                         }
-                    }, null));
+                    }, id -> {
+                        TransportProtos.ToDeviceRpcRequestMsg rpcRequestMsg = transportContext.getRpcAwaitingAck().remove(id);
+                        if (rpcRequestMsg != null) {
+                            transportService.process(state.getSession(), msg, RpcStatus.TIMEOUT, TransportServiceCallback.EMPTY);
+                        }
+                    }));
                 }
                 if (conRequest) {
                     response.addMessageObserver(new TbCoapMessageObserver(requestId, id -> awake(state), id -> asleep(state)));
@@ -554,11 +563,11 @@ public class DefaultCoapClientContext implements CoapClientContext {
                     transportService.process(state.getSession(),
                             TransportProtos.ToDeviceRpcResponseMsg.newBuilder()
                                     .setRequestId(msg.getRequestId()).setError(error).build(), TransportServiceCallback.EMPTY);
-                } else if (msg.getPersisted() && sent) {
-                    if (conRequest) {
-                        transportService.process(state.getSession(), msg, RpcStatus.SENT, TransportServiceCallback.EMPTY);
-                    } else {
+                } else if (sent) {
+                    if (!conRequest) {
                         transportService.process(state.getSession(), msg, RpcStatus.DELIVERED, TransportServiceCallback.EMPTY);
+                    } else if (msg.getPersisted()) {
+                        transportService.process(state.getSession(), msg, RpcStatus.SENT, TransportServiceCallback.EMPTY);
                     }
                 }
             }
--- a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/MqttTransportHandler.java
View file @8513c99
+++ b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/MqttTransportHandler.java
View file @8513c99
@@ -850,24 +850,27 @@ public class MqttTransportHandler extends ChannelInboundHandlerAdapter implement
         try {
             deviceSessionCtx.getPayloadAdaptor().convertToPublish(deviceSessionCtx, rpcRequest).ifPresent(payload -> {
                 int msgId = ((MqttPublishMessage) payload).variableHeader().packetId();
-                if (rpcRequest.getPersisted() && isAckExpected(payload)) {
+                if (isAckExpected(payload)) {
                     rpcAwaitingAck.put(msgId, rpcRequest);
                     context.getScheduler().schedule(() -> {
-                        rpcAwaitingAck.remove(msgId);
-                    }, Math.max(0, rpcRequest.getExpirationTime() - System.currentTimeMillis()), TimeUnit.MILLISECONDS);
+                        TransportProtos.ToDeviceRpcRequestMsg msg = rpcAwaitingAck.remove(msgId);
+                        if (msg != null) {
+                            transportService.process(deviceSessionCtx.getSessionInfo(), rpcRequest, RpcStatus.TIMEOUT, TransportServiceCallback.EMPTY);
+                        }
+                    }, Math.max(0, Math.min(rpcRequest.getTimeout(), rpcRequest.getExpirationTime() - System.currentTimeMillis())), TimeUnit.MILLISECONDS);
                 }
                 var cf = publish(payload, deviceSessionCtx);
-                if (rpcRequest.getPersisted()) {
-                    cf.addListener(result -> {
-                        if (result.cause() == null) {
-                            if (isAckExpected(payload)) {
-                                transportService.process(deviceSessionCtx.getSessionInfo(), rpcRequest, RpcStatus.SENT, TransportServiceCallback.EMPTY);
-                            } else {
-                                transportService.process(deviceSessionCtx.getSessionInfo(), rpcRequest, RpcStatus.DELIVERED, TransportServiceCallback.EMPTY);
-                            }
+                cf.addListener(result -> {
+                    if (result.cause() == null) {
+                        if (!isAckExpected(payload)) {
+                            transportService.process(deviceSessionCtx.getSessionInfo(), rpcRequest, RpcStatus.DELIVERED, TransportServiceCallback.EMPTY);
+                        } else if (rpcRequest.getPersisted()) {
+                            transportService.process(deviceSessionCtx.getSessionInfo(), rpcRequest, RpcStatus.SENT, TransportServiceCallback.EMPTY);
                         }
-                    });
-                }
+                    } else {
+                        // TODO: send error
+                    }
+                });
             });
         } catch (Exception e) {
             transportService.process(deviceSessionCtx.getSessionInfo(),
--- a/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/service/DefaultTransportService.java
View file @8513c99
+++ b/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/service/DefaultTransportService.java
View file @8513c99
@@ -581,19 +581,17 @@ public class DefaultTransportService implements TransportService {
     @Override
     public void process(TransportProtos.SessionInfoProto sessionInfo, TransportProtos.ToDeviceRpcRequestMsg msg, RpcStatus rpcStatus, TransportServiceCallback<Void> callback) {
-        if (msg.getPersisted()) {
-            TransportProtos.ToDevicePersistedRpcResponseMsg responseMsg = TransportProtos.ToDevicePersistedRpcResponseMsg.newBuilder()
-                    .setRequestId(msg.getRequestId())
-                    .setRequestIdLSB(msg.getRequestIdLSB())
-                    .setRequestIdMSB(msg.getRequestIdMSB())
-                    .setStatus(rpcStatus.name())
-                    .build();
-
-            if (checkLimits(sessionInfo, responseMsg, callback)) {
-                reportActivityInternal(sessionInfo);
-                sendToDeviceActor(sessionInfo, TransportToDeviceActorMsg.newBuilder().setSessionInfo(sessionInfo).setPersistedRpcResponseMsg(responseMsg).build(),
-                        new ApiStatsProxyCallback<>(getTenantId(sessionInfo), getCustomerId(sessionInfo), 1, TransportServiceCallback.EMPTY));
-            }
+        TransportProtos.ToDeviceRpcResponseStatusMsg responseMsg = TransportProtos.ToDeviceRpcResponseStatusMsg.newBuilder()
+                .setRequestId(msg.getRequestId())
+                .setRequestIdLSB(msg.getRequestIdLSB())
+                .setRequestIdMSB(msg.getRequestIdMSB())
+                .setStatus(rpcStatus.name())
+                .build();
+
+        if (checkLimits(sessionInfo, responseMsg, callback)) {
+            reportActivityInternal(sessionInfo);
+            sendToDeviceActor(sessionInfo, TransportToDeviceActorMsg.newBuilder().setSessionInfo(sessionInfo).setRpcResponseStatusMsg(responseMsg).build(),
+                    new ApiStatsProxyCallback<>(getTenantId(sessionInfo), getCustomerId(sessionInfo), 1, TransportServiceCallback.EMPTY));
         }
     }
--- a/rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/RuleEngineDeviceRpcRequest.java
View file @8513c99
+++ b/rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/RuleEngineDeviceRpcRequest.java
View file @8513c99
@@ -39,6 +39,7 @@ public final class RuleEngineDeviceRpcRequest {
     private final String method;
     private final String body;
     private final long expirationTime;
+    private final long timeout;
     private final boolean restApiCall;
     private final String additionalInfo;
--- a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/rpc/TbSendRPCRequestNode.java
View file @8513c99
+++ b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/rpc/TbSendRPCRequestNode.java
View file @8513c99
@@ -89,9 +89,12 @@ public class TbSendRPCRequestNode implements TbNode {
             tmp = msg.getMetaData().getValue("originServiceId");
             String originServiceId = !StringUtils.isEmpty(tmp) ? tmp : null;
-            tmp = msg.getMetaData().getValue("expirationTime");
+            tmp = msg.getMetaData().getValue(DataConstants.EXPIRATION_TIME);
             long expirationTime = !StringUtils.isEmpty(tmp) ? Long.parseLong(tmp) : (System.currentTimeMillis() + TimeUnit.SECONDS.toMillis(config.getTimeoutInSeconds()));
+            tmp = msg.getMetaData().getValue(DataConstants.TIMEOUT);
+            long timeout = !StringUtils.isEmpty(tmp) ? Long.parseLong(tmp) : TimeUnit.SECONDS.toMillis(config.getTimeoutInSeconds());
+
             String params;
             JsonElement paramsEl = json.get("params");
             if (paramsEl.isJsonPrimitive()) {
@@ -112,6 +115,7 @@ public class TbSendRPCRequestNode implements TbNode {
                     .requestUUID(requestUUID)
                     .originServiceId(originServiceId)
                     .expirationTime(expirationTime)
+                    .timeout(timeout)
                     .restApiCall(restApiCall)
                     .persisted(persisted)
                     .additionalInfo(additionalInfo)