Merge pull request #5114 from thingsboard/feature/sequential-rpc

[3.3.1] Sequential processing of RPC calls

application/src/main/java/org/thingsboard/server/actors/ActorSystemContext.java

@@ -400,6 +400,14 @@ public class ActorSystemContext {
     @Getter
     private String debugPerTenantLimitsConfiguration;
 
+    @Value("${actors.rpc.sequential:false}")
+    @Getter
+    private boolean rpcSequential;
+
+    @Value("${actors.rpc.max_retries:5}")
+    @Getter
+    private int maxRpcRetries;
+
     @Getter
     @Setter
     private TbActorSystem actorSystem;

application/src/main/java/org/thingsboard/server/actors/device/DeviceActorMessageProcessor.java

@@ -48,6 +48,7 @@ import org.thingsboard.server.common.data.kv.AttributeKvEntry;
 import org.thingsboard.server.common.data.kv.KvEntry;
 import org.thingsboard.server.common.data.page.PageData;
 import org.thingsboard.server.common.data.page.PageLink;
+import org.thingsboard.server.common.data.page.SortOrder;
 import org.thingsboard.server.common.data.relation.EntityRelation;
 import org.thingsboard.server.common.data.relation.RelationTypeGroup;
 import org.thingsboard.server.common.data.rpc.Rpc;
@@ -79,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;
@@ -98,9 +99,11 @@ import java.util.Arrays;
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.HashSet;
+import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
+import java.util.Optional;
 import java.util.Set;
 import java.util.UUID;
 import java.util.function.Consumer;
@@ -119,6 +122,7 @@ class DeviceActorMessageProcessor extends AbstractContextAwareMsgProcessor {
     private final Map<UUID, SessionInfo> attributeSubscriptions;
     private final Map<UUID, SessionInfo> rpcSubscriptions;
     private final Map<Integer, ToDeviceRpcRequestMetadata> toDeviceRpcPendingMap;
+    private final boolean rpcSequential;
 
     private int rpcSeq = 0;
     private String deviceName;
@@ -130,9 +134,10 @@ class DeviceActorMessageProcessor extends AbstractContextAwareMsgProcessor {
         super(systemContext);
         this.tenantId = tenantId;
         this.deviceId = deviceId;
+        this.rpcSequential = systemContext.isRpcSequential();
         this.attributeSubscriptions = new HashMap<>();
         this.rpcSubscriptions = new HashMap<>();
-        this.toDeviceRpcPendingMap = new HashMap<>();
+        this.toDeviceRpcPendingMap = new LinkedHashMap<>();
         this.sessions = new LinkedHashMapRemoveEldest<>(systemContext.getMaxConcurrentSessionsPerDevice(), this::notifyTransportAboutClosedSessionMaxSessionsLimit);
         if (initAttributes()) {
             restoreSessions();
@@ -183,19 +188,19 @@ class DeviceActorMessageProcessor extends AbstractContextAwareMsgProcessor {
         if (timeout <= 0) {
             log.debug("[{}][{}] Ignoring message due to exp time reached, {}", deviceId, request.getId(), request.getExpirationTime());
             if (persisted) {
-                createRpc(request, RpcStatus.TIMEOUT);
+                createRpc(request, RpcStatus.EXPIRED);
             }
             return;
         } else if (persisted) {
             createRpc(request, RpcStatus.QUEUED);
         }
 
-        boolean sent;
+        boolean sent = false;
         if (systemContext.isEdgesEnabled() && edgeId != null) {
             log.debug("[{}][{}] device is related to edge [{}]. Saving RPC request to edge queue", tenantId, deviceId, edgeId.getId());
             saveRpcRequestToEdgeQueue(request, rpcRequest.getRequestId());
             sent = true;
-        } else {
+        } else if (isSendNewRpcAvailable()) {
             sent = rpcSubscriptions.size() > 0;
             Set<UUID> syncSessionSet = new HashSet<>();
             rpcSubscriptions.forEach((key, value) -> {
@@ -227,6 +232,10 @@ class DeviceActorMessageProcessor extends AbstractContextAwareMsgProcessor {
         }
     }
 
+    private boolean isSendNewRpcAvailable() {
+        return !rpcSequential || toDeviceRpcPendingMap.values().stream().filter(md -> !md.isDelivered()).findAny().isEmpty();
+    }
+
     private Rpc createRpc(ToDeviceRpcRequest request, RpcStatus status) {
         Rpc rpc = new Rpc(new RpcId(request.getId()));
         rpc.setCreatedTime(System.currentTimeMillis());
@@ -266,16 +275,26 @@ class DeviceActorMessageProcessor extends AbstractContextAwareMsgProcessor {
 
     void processRemoveRpc(TbActorCtx context, RemoveRpcActorMsg msg) {
         log.debug("[{}] Processing remove rpc command", msg.getRequestId());
-        Integer requestId = null;
+        Map.Entry<Integer, ToDeviceRpcRequestMetadata> entry = null;
-        for (Map.Entry<Integer, ToDeviceRpcRequestMetadata> entry : toDeviceRpcPendingMap.entrySet()) {
+        for (Map.Entry<Integer, ToDeviceRpcRequestMetadata> e : toDeviceRpcPendingMap.entrySet()) {
-            if (entry.getValue().getMsg().getMsg().getId().equals(msg.getRequestId())) {
+            if (e.getValue().getMsg().getMsg().getId().equals(msg.getRequestId())) {
-                requestId = entry.getKey();
+                entry = e;
                 break;
             }
         }
 
-        if (requestId != null) {
+        if (entry != null) {
-            toDeviceRpcPendingMap.remove(requestId);
+            if (entry.getValue().isDelivered()) {
+                toDeviceRpcPendingMap.remove(entry.getKey());
+            } else {
+                Optional<Map.Entry<Integer, ToDeviceRpcRequestMetadata>> firstRpc = getFirstRpc();
+                if (firstRpc.isPresent() && entry.getKey().equals(firstRpc.get().getKey())) {
+                    toDeviceRpcPendingMap.remove(entry.getKey());
+                    sendNextPendingRequest(context);
+                } else {
+                    toDeviceRpcPendingMap.remove(entry.getKey());
+                }
+            }
         }
     }
 
@@ -290,14 +309,17 @@ class DeviceActorMessageProcessor extends AbstractContextAwareMsgProcessor {
         if (requestMd != null) {
             log.debug("[{}] RPC request [{}] timeout detected!", deviceId, msg.getId());
             if (requestMd.getMsg().getMsg().isPersisted()) {
-                systemContext.getTbRpcService().save(tenantId, new RpcId(requestMd.getMsg().getMsg().getId()), RpcStatus.TIMEOUT, null);
+                systemContext.getTbRpcService().save(tenantId, new RpcId(requestMd.getMsg().getMsg().getId()), RpcStatus.EXPIRED, null);
             }
             systemContext.getTbCoreDeviceRpcService().processRpcResponseFromDeviceActor(new FromDeviceRpcResponse(requestMd.getMsg().getMsg().getId(),
                     null, requestMd.isSent() ? RpcError.TIMEOUT : RpcError.NO_ACTIVE_CONNECTION));
+            if (!requestMd.isDelivered()) {
+                sendNextPendingRequest(context);
+            }
         }
     }
 
-    private void sendPendingRequests(TbActorCtx context, UUID sessionId, SessionInfoProto sessionInfo) {
+    private void sendPendingRequests(TbActorCtx context, UUID sessionId, String nodeId) {
         SessionType sessionType = getSessionType(sessionId);
         if (!toDeviceRpcPendingMap.isEmpty()) {
             log.debug("[{}] Pushing {} pending RPC messages to new async session [{}]", deviceId, toDeviceRpcPendingMap.size(), sessionId);
@@ -309,20 +331,33 @@ class DeviceActorMessageProcessor extends AbstractContextAwareMsgProcessor {
             log.debug("[{}] No pending RPC messages for new async session [{}]", deviceId, sessionId);
         }
         Set<Integer> sentOneWayIds = new HashSet<>();
-        if (sessionType == SessionType.ASYNC) {
+
-            toDeviceRpcPendingMap.entrySet().forEach(processPendingRpc(context, sessionId, sessionInfo.getNodeId(), sentOneWayIds));
+        if (rpcSequential) {
+            getFirstRpc().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, sessionInfo.getNodeId(), sentOneWayIds));
+            toDeviceRpcPendingMap.entrySet().stream().findFirst().ifPresent(processPendingRpc(context, sessionId, nodeId, sentOneWayIds));
         }
 
         sentOneWayIds.stream().filter(id -> !toDeviceRpcPendingMap.get(id).getMsg().getMsg().isPersisted()).forEach(toDeviceRpcPendingMap::remove);
     }
 
+    private Optional<Map.Entry<Integer, ToDeviceRpcRequestMetadata>> getFirstRpc() {
+        return toDeviceRpcPendingMap.entrySet().stream().filter(e -> !e.getValue().isDelivered()).findFirst();
+    }
+
+    private void sendNextPendingRequest(TbActorCtx context) {
+        if (rpcSequential) {
+            rpcSubscriptions.forEach((id, s) -> sendPendingRequests(context, id, s.getNodeId()));
+        }
+    }
+
     private Consumer<Map.Entry<Integer, ToDeviceRpcRequestMetadata>> processPendingRpc(TbActorCtx context, UUID sessionId, String nodeId, Set<Integer> sentOneWayIds) {
         return entry -> {
             ToDeviceRpcRequest request = entry.getValue().getMsg().getMsg();
             ToDeviceRpcRequestBody body = request.getBody();
-            if (request.isOneway()) {
+            if (request.isOneway() && !rpcSequential) {
                 sentOneWayIds.add(entry.getKey());
                 systemContext.getTbCoreDeviceRpcService().processRpcResponseFromDeviceActor(new FromDeviceRpcResponse(request.getId(), null, null));
             }
@@ -355,7 +390,7 @@ class DeviceActorMessageProcessor extends AbstractContextAwareMsgProcessor {
             processSubscriptionCommands(context, sessionInfo, msg.getSubscribeToRPC());
         }
         if (msg.hasSendPendingRPC()) {
-            sendPendingRequests(context, getSessionId(sessionInfo), sessionInfo);
+            sendPendingRequests(context, getSessionId(sessionInfo), sessionInfo.getNodeId());
         }
         if (msg.hasGetAttributes()) {
             handleGetAttributesRequest(context, sessionInfo, msg.getGetAttributes());
@@ -369,8 +404,8 @@ class DeviceActorMessageProcessor extends AbstractContextAwareMsgProcessor {
         if (msg.hasClaimDevice()) {
             handleClaimDeviceMsg(context, sessionInfo, msg.getClaimDevice());
         }
-        if (msg.hasPersistedRpcResponseMsg()) {
+        if (msg.hasRpcResponseStatusMsg()) {
-            processPersistedRpcResponses(context, sessionInfo, msg.getPersistedRpcResponseMsg());
+            processRpcResponseStatus(context, sessionInfo, msg.getRpcResponseStatusMsg());
         }
         if (msg.hasUplinkNotificationMsg()) {
             processUplinkNotificationMsg(context, sessionInfo, msg.getUplinkNotificationMsg());
@@ -530,38 +565,63 @@ class DeviceActorMessageProcessor extends AbstractContextAwareMsgProcessor {
         boolean success = requestMd != null;
         if (success) {
             boolean hasError = StringUtils.isNotEmpty(responseMsg.getError());
-            String payload = hasError ? responseMsg.getError() : responseMsg.getPayload();
+            try {
-            systemContext.getTbCoreDeviceRpcService().processRpcResponseFromDeviceActor(
+                String payload = hasError ? responseMsg.getError() : responseMsg.getPayload();
-                    new FromDeviceRpcResponse(requestMd.getMsg().getMsg().getId(),
+                systemContext.getTbCoreDeviceRpcService().processRpcResponseFromDeviceActor(
-                            payload, null));
+                        new FromDeviceRpcResponse(requestMd.getMsg().getMsg().getId(),
-            if (requestMd.getMsg().getMsg().isPersisted()) {
+                                payload, null));
-                RpcStatus status = hasError ? RpcStatus.FAILED : RpcStatus.SUCCESSFUL;
+                if (requestMd.getMsg().getMsg().isPersisted()) {
-                JsonNode response;
+                    RpcStatus status = hasError ? RpcStatus.FAILED : RpcStatus.SUCCESSFUL;
-                try {
+                    JsonNode response;
-                    response = JacksonUtil.toJsonNode(payload);
+                    try {
-                } catch (IllegalArgumentException e) {
+                        response = JacksonUtil.toJsonNode(payload);
-                    response = JacksonUtil.newObjectNode().put("error", payload);
+                    } catch (IllegalArgumentException e) {
+                        response = JacksonUtil.newObjectNode().put("error", payload);
+                    }
+                    systemContext.getTbRpcService().save(tenantId, new RpcId(requestMd.getMsg().getMsg().getId()), status, response);
+                }
+            } finally {
+                if (hasError) {
+                    sendNextPendingRequest(context);
                 }
-                systemContext.getTbRpcService().save(tenantId, new RpcId(requestMd.getMsg().getMsg().getId()), status, response);
             }
         } else {
             log.debug("[{}] Rpc command response [{}] is stale!", deviceId, responseMsg.getRequestId());
         }
     }
 
-    private void processPersistedRpcResponses(TbActorCtx context, SessionInfoProto sessionInfo, ToDevicePersistedRpcResponseMsg responseMsg) {
+    private void processRpcResponseStatus(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());
-        ToDeviceRpcRequestMetadata md;
-        if (RpcStatus.DELIVERED.equals(status)) {
-            md = toDeviceRpcPendingMap.get(responseMsg.getRequestId());
-        } else {
-            md = toDeviceRpcPendingMap.remove(responseMsg.getRequestId());
-        }
 
         if (md != null) {
-            systemContext.getTbRpcService().save(tenantId, new RpcId(rpcId), status, null);
+            if (status.equals(RpcStatus.DELIVERED)) {
+                if (md.getMsg().getMsg().isOneway()) {
+                    toDeviceRpcPendingMap.remove(responseMsg.getRequestId());
+                    if (rpcSequential) {
+                        systemContext.getTbCoreDeviceRpcService().processRpcResponseFromDeviceActor(new FromDeviceRpcResponse(rpcId, null, null));
+                    }
+                } else {
+                    md.setDelivered(true);
+                }
+            } else if (status.equals(RpcStatus.TIMEOUT)) {
+                Integer maxRpcRetries = md.getMsg().getMsg().getRetries();
+                maxRpcRetries = maxRpcRetries == null ? systemContext.getMaxRpcRetries() : Math.min(maxRpcRetries, systemContext.getMaxRpcRetries());
+                if (maxRpcRetries <= md.getRetries()) {
+                    toDeviceRpcPendingMap.remove(responseMsg.getRequestId());
+                    status = RpcStatus.FAILED;
+                } else {
+                    md.setRetries(md.getRetries() + 1);
+                }
+            }
+
+            if (md.getMsg().getMsg().isPersisted()) {
+                systemContext.getTbRpcService().save(tenantId, new RpcId(rpcId), status, null);
+            }
+            if (status != RpcStatus.SENT) {
+                sendNextPendingRequest(context);
+            }
         } else {
             log.info("[{}][{}] Rpc has already removed from pending map.", deviceId, rpcId);
         }
@@ -601,7 +661,7 @@ class DeviceActorMessageProcessor extends AbstractContextAwareMsgProcessor {
             sessionMD.setSubscribedToRPC(true);
             log.debug("[{}] Registering rpc subscription for session [{}]", deviceId, sessionId);
             rpcSubscriptions.put(sessionId, sessionMD.getSessionInfo());
-            sendPendingRequests(context, sessionId, sessionInfo);
+            sendPendingRequests(context, sessionId, sessionInfo.getNodeId());
             dumpSessions();
         }
     }
@@ -871,7 +931,7 @@ class DeviceActorMessageProcessor extends AbstractContextAwareMsgProcessor {
 
     void init(TbActorCtx ctx) {
         schedulePeriodicMsgWithDelay(ctx, SessionTimeoutCheckMsg.instance(), systemContext.getSessionReportTimeout(), systemContext.getSessionReportTimeout());
-        PageLink pageLink = new PageLink(1024);
+        PageLink pageLink = new PageLink(1024, 0, null, new SortOrder("createdTime"));
         PageData<Rpc> pageData;
         do {
             pageData = systemContext.getTbRpcService().findAllByDeviceIdAndStatus(tenantId, deviceId, RpcStatus.QUEUED, pageLink);
@@ -879,7 +939,7 @@ class DeviceActorMessageProcessor extends AbstractContextAwareMsgProcessor {
                 ToDeviceRpcRequest msg = JacksonUtil.convertValue(rpc.getRequest(), ToDeviceRpcRequest.class);
                 long timeout = rpc.getExpirationTime() - System.currentTimeMillis();
                 if (timeout <= 0) {
-                    rpc.setStatus(RpcStatus.TIMEOUT);
+                    rpc.setStatus(RpcStatus.EXPIRED);
                     systemContext.getTbRpcService().save(tenantId, rpc);
                 } else {
                     registerPendingRpcRequest(ctx, new ToDeviceRpcRequestActorMsg(systemContext.getServiceId(), msg), false, creteToDeviceRpcRequestMsg(msg), timeout);

application/src/main/java/org/thingsboard/server/actors/device/ToDeviceRpcRequestMetadata.java

@@ -25,4 +25,6 @@ import org.thingsboard.server.service.rpc.ToDeviceRpcRequestActorMsg;
 public class ToDeviceRpcRequestMetadata {
     private final ToDeviceRpcRequestActorMsg msg;
     private final boolean sent;
+    private int retries;
+    private boolean delivered;
 }

application/src/main/java/org/thingsboard/server/controller/AbstractRpcController.java

@@ -75,11 +75,12 @@ 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 timeout = rpcRequestBody.has(DataConstants.TIMEOUT) ? rpcRequestBody.get(DataConstants.TIMEOUT).asLong() : defaultTimeout;
-            long expTime = System.currentTimeMillis() + Math.max(minTimeout, timeout);
+            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, deviceId, new HttpValidationCallback(response, new FutureCallback<>() {
                 @Override
                 public void onSuccess(@Nullable DeferredResult<ResponseEntity> result) {
@@ -90,6 +91,7 @@ public abstract class AbstractRpcController extends BaseController {
                             expTime,
                             body,
                             persisted,
+                            retries,
                             additionalInfo
                     );
                     deviceRpcService.processRestApiRpcRequest(rpcRequest, fromDeviceRpcResponse -> reply(new LocalRequestMetaData(rpcRequest, currentUser, result), fromDeviceRpcResponse, timeoutStatus, noActiveConnectionStatus), currentUser);

application/src/main/java/org/thingsboard/server/service/rpc/DefaultTbCoreDeviceRpcService.java

@@ -166,6 +166,11 @@ public class DefaultTbCoreDeviceRpcService implements TbCoreDeviceRpcService {
         metaData.putValue("oneway", Boolean.toString(msg.isOneway()));
         metaData.putValue(DataConstants.PERSISTENT, Boolean.toString(msg.isPersisted()));
 
+        if (msg.getRetries() != null) {
+            metaData.putValue(DataConstants.RETRIES, msg.getRetries().toString());
+        }
+
+
         Device device = deviceService.findDeviceById(msg.getTenantId(), msg.getDeviceId());
         if (device != null) {
             metaData.putValue("deviceName", device.getName());

application/src/main/java/org/thingsboard/server/service/rpc/DefaultTbRuleEngineRpcService.java

@@ -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(), new ToDeviceRpcRequestBody(src.getMethod(), src.getBody()), src.isPersisted(), src.getRetries(), src.getAdditionalInfo());
         forwardRpcRequestToDeviceActor(request, response -> {
             if (src.isRestApiCall()) {
                 sendRpcResponseToTbCore(src.getOriginServiceId(), response);

application/src/main/resources/thingsboard.yml

@@ -326,6 +326,9 @@ actors:
       queue_size: "${ACTORS_RULE_TRANSACTION_QUEUE_SIZE:15000}"
       # Time in milliseconds for transaction to complete
       duration: "${ACTORS_RULE_TRANSACTION_DURATION:60000}"
+  rpc:
+    max_retries: "${ACTORS_RPC_MAX_RETRIES:5}"
+    sequential: "${ACTORS_RPC_SEQUENTIAL:false}"
   statistics:
     # Enable/disable actor statistics
     enabled: "${ACTORS_STATISTICS_ENABLED:true}"
@@ -699,7 +702,7 @@ transport:
 
 # Edges parameters
 edges:
-  enabled: "${EDGES_ENABLED:false}"
+  enabled: "${EDGES_ENABLED:true}"
   rpc:
     port: "${EDGES_RPC_PORT:7070}"
     client_max_keep_alive_time_sec: "${EDGES_RPC_CLIENT_MAX_KEEP_ALIVE_TIME_SEC:300}"

application/src/test/java/org/thingsboard/server/transport/mqtt/rpc/AbstractMqttServerSideRpcDefaultIntegrationTest.java

@@ -89,6 +89,11 @@ public abstract class AbstractMqttServerSideRpcDefaultIntegrationTest extends Ab
         processTwoWayRpcTest();
     }
 
+    @Test
+    public void testSequenceServerMqttTwoWayRpc() throws Exception {
+        processSequenceTwoWayRpcTest();
+    }
+
     @Test
     public void testGatewayServerMqttOneWayRpc() throws Exception {
         processOneWayRpcTestGateway("Gateway Device OneWay RPC");

application/src/test/java/org/thingsboard/server/transport/mqtt/rpc/AbstractMqttServerSideRpcIntegrationTest.java

@@ -16,6 +16,7 @@
 package org.thingsboard.server.transport.mqtt.rpc;
 
 import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.node.ObjectNode;
 import com.google.protobuf.InvalidProtocolBufferException;
 import com.nimbusds.jose.util.StandardCharset;
 import io.netty.handler.codec.mqtt.MqttQoS;
@@ -27,13 +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;
 
@@ -101,6 +105,32 @@ public abstract class AbstractMqttServerSideRpcIntegrationTest extends AbstractM
         Assert.assertEquals(expected, result);
     }
 
+    protected void processSequenceTwoWayRpcTest() throws Exception {
+        List<String> expected = new ArrayList<>();
+        List<String> result = new ArrayList<>();
+
+        String deviceId = savedDevice.getId().getId().toString();
+
+        for (int i = 0; i < 10; i++) {
+            ObjectNode request = JacksonUtil.newObjectNode();
+            request.put("method", "test");
+            request.put("params", i);
+            expected.add(JacksonUtil.toString(request));
+            request.put("persistent", true);
+            doPostAsync("/api/rpc/twoway/" + deviceId, JacksonUtil.toString(request), String.class, status().isOk());
+        }
+
+        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(10, TimeUnit.SECONDS);
+        Assert.assertEquals(expected, result);
+    }
+
     protected void processTwoWayRpcTestGateway(String deviceName) throws Exception {
         MqttAsyncClient client = getMqttAsyncClient(gatewayAccessToken);
 
@@ -213,4 +243,38 @@ public abstract class AbstractMqttServerSideRpcIntegrationTest extends AbstractM
 
         }
     }
+
+    protected class TestSequenceMqttCallback implements MqttCallback {
+
+        private final MqttAsyncClient client;
+        private final CountDownLatch latch;
+        private final List<String> expected;
+
+        TestSequenceMqttCallback(MqttAsyncClient client, CountDownLatch latch, List<String> expected) {
+            this.client = client;
+            this.latch = latch;
+            this.expected = expected;
+        }
+
+        @Override
+        public void connectionLost(Throwable throwable) {
+        }
+
+        @Override
+        public void messageArrived(String requestTopic, MqttMessage mqttMessage) throws Exception {
+            log.info("Message Arrived: " + Arrays.toString(mqttMessage.getPayload()));
+            expected.add(new String(mqttMessage.getPayload()));
+            String responseTopic = requestTopic.replace("request", "response");
+            var qoS = mqttMessage.getQos();
+
+            client.messageArrivedComplete(mqttMessage.getId(), qoS);
+            client.publish(responseTopic, processMessageArrived(requestTopic, mqttMessage));
+            latch.countDown();
+        }
+
+        @Override
+        public void deliveryComplete(IMqttDeliveryToken iMqttDeliveryToken) {
+
+        }
+    }
 }

application/src/test/resources/application-test.properties

@@ -6,4 +6,5 @@ edges.storage.sleep_between_batches=500
 transport.lwm2m.server.security.key_alias=server
 transport.lwm2m.server.security.key_password=server
 transport.lwm2m.bootstrap.security.key_alias=server
-transport.lwm2m.bootstrap.security.key_password=server
+transport.lwm2m.bootstrap.security.key_password=server
+actors.rpc.sequential=true

common/cluster-api/src/main/proto/queue.proto

@@ -346,7 +346,7 @@ message UplinkNotificationMsg {
   int64 uplinkTs = 1;
 }
 
-message ToDevicePersistedRpcResponseMsg {
+message ToDeviceRpcResponseStatusMsg {
   int32 requestId = 1;
   int64 requestIdMSB = 2;
   int64 requestIdLSB = 3;
@@ -456,7 +456,7 @@ message TransportToDeviceActorMsg {
   SubscriptionInfoProto subscriptionInfo = 7;
   ClaimDeviceMsg claimDevice = 8;
   ProvisionDeviceRequestMsg provisionDevice = 9;
-  ToDevicePersistedRpcResponseMsg persistedRpcResponseMsg = 10;
+  ToDeviceRpcResponseStatusMsg rpcResponseStatusMsg = 10;
   SendPendingRPCMsg sendPendingRPC = 11;
   UplinkNotificationMsg uplinkNotificationMsg = 12;
 }

common/data/src/main/java/org/thingsboard/server/common/data/DataConstants.java

@@ -36,7 +36,10 @@ 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 RETRIES = "retries";
     public static final String COAP_TRANSPORT_NAME = "COAP";
     public static final String LWM2M_TRANSPORT_NAME = "LWM2M";
     public static final String MQTT_TRANSPORT_NAME = "MQTT";
@@ -85,9 +88,11 @@ public class DataConstants {
     public static final String RPC_CALL_FROM_SERVER_TO_DEVICE = "RPC_CALL_FROM_SERVER_TO_DEVICE";
 
     public static final String RPC_QUEUED = "RPC_QUEUED";
+    public static final String RPC_SENT = "RPC_SENT";
     public static final String RPC_DELIVERED = "RPC_DELIVERED";
     public static final String RPC_SUCCESSFUL = "RPC_SUCCESSFUL";
     public static final String RPC_TIMEOUT = "RPC_TIMEOUT";
+    public static final String RPC_EXPIRED = "RPC_EXPIRED";
     public static final String RPC_FAILED = "RPC_FAILED";
     public static final String RPC_DELETED = "RPC_DELETED";
 

common/data/src/main/java/org/thingsboard/server/common/data/rpc/RpcStatus.java

@@ -16,5 +16,5 @@
 package org.thingsboard.server.common.data.rpc;
 
 public enum RpcStatus {
-    QUEUED, DELIVERED, SUCCESSFUL, TIMEOUT, FAILED
+    QUEUED, SENT, DELIVERED, SUCCESSFUL, TIMEOUT, EXPIRED, FAILED
 }

common/message/src/main/java/org/thingsboard/server/common/msg/rpc/ToDeviceRpcRequest.java

@@ -36,6 +36,7 @@ public class ToDeviceRpcRequest implements Serializable {
     private final long expirationTime;
     private final ToDeviceRpcRequestBody body;
     private final boolean persisted;
+    private final Integer retries;
     @JsonIgnore
     private final String additionalInfo;
 }

common/transport/coap/src/main/java/org/thingsboard/server/transport/coap/client/DefaultCoapClientContext.java

@@ -42,6 +42,7 @@ import org.thingsboard.server.common.data.device.profile.ProtoTransportPayloadCo
 import org.thingsboard.server.common.data.device.profile.TransportPayloadTypeConfiguration;
 import org.thingsboard.server.common.data.id.DeviceId;
 import org.thingsboard.server.common.data.id.DeviceProfileId;
+import org.thingsboard.server.common.data.rpc.RpcStatus;
 import org.thingsboard.server.common.msg.session.FeatureType;
 import org.thingsboard.server.common.msg.session.SessionMsgType;
 import org.thingsboard.server.common.transport.SessionMsgListener;
@@ -192,29 +193,7 @@ public class DefaultCoapClientContext implements CoapClientContext {
         client.lock();
         try {
             long uplinkTime = client.updateLastUplinkTime(uplinkTs);
-            long timeout;
+            long timeout = getTimeout(client, powerMode, profileSettings);
-            if (PowerMode.PSM.equals(powerMode)) {
-                Long psmActivityTimer = client.getPsmActivityTimer();
-                if (psmActivityTimer == null && profileSettings != null) {
-                    psmActivityTimer = profileSettings.getPsmActivityTimer();
-
-                }
-                if (psmActivityTimer == null || psmActivityTimer == 0L) {
-                    psmActivityTimer = config.getPsmActivityTimer();
-                }
-
-                timeout = psmActivityTimer;
-            } else {
-                Long pagingTransmissionWindow = client.getPagingTransmissionWindow();
-                if (pagingTransmissionWindow == null && profileSettings != null) {
-                    pagingTransmissionWindow = profileSettings.getPagingTransmissionWindow();
-
-                }
-                if (pagingTransmissionWindow == null || pagingTransmissionWindow == 0L) {
-                    pagingTransmissionWindow = config.getPagingTransmissionWindow();
-                }
-                timeout = pagingTransmissionWindow;
-            }
             Future<Void> sleepTask = client.getSleepTask();
             if (sleepTask != null) {
                 sleepTask.cancel(false);
@@ -234,6 +213,33 @@ public class DefaultCoapClientContext implements CoapClientContext {
         }
     }
 
+    private long getTimeout(TbCoapClientState client, PowerMode powerMode, PowerSavingConfiguration profileSettings) {
+        long timeout;
+        if (PowerMode.PSM.equals(powerMode)) {
+            Long psmActivityTimer = client.getPsmActivityTimer();
+            if (psmActivityTimer == null && profileSettings != null) {
+                psmActivityTimer = profileSettings.getPsmActivityTimer();
+
+            }
+            if (psmActivityTimer == null || psmActivityTimer == 0L) {
+                psmActivityTimer = config.getPsmActivityTimer();
+            }
+
+            timeout = psmActivityTimer;
+        } else {
+            Long pagingTransmissionWindow = client.getPagingTransmissionWindow();
+            if (pagingTransmissionWindow == null && profileSettings != null) {
+                pagingTransmissionWindow = profileSettings.getPagingTransmissionWindow();
+
+            }
+            if (pagingTransmissionWindow == null || pagingTransmissionWindow == 0L) {
+                pagingTransmissionWindow = config.getPagingTransmissionWindow();
+            }
+            timeout = pagingTransmissionWindow;
+        }
+        return timeout;
+    }
+
     private boolean registerFeatureObservation(TbCoapClientState state, String token, CoapExchange exchange, FeatureType featureType) {
         state.lock();
         try {
@@ -524,17 +530,38 @@ 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) {
+                    PowerMode powerMode = state.getPowerMode();
+                    PowerSavingConfiguration profileSettings = null;
+                    if (powerMode == null) {
+                        var clientProfile = getProfile(state.getProfileId());
+                        if (clientProfile.isPresent()) {
+                            profileSettings = clientProfile.get().getClientSettings();
+                            if (profileSettings != null) {
+                                powerMode = profileSettings.getPowerMode();
+                            }
+                        }
+                    }
+
                     transportContext.getRpcAwaitingAck().put(requestId, msg);
                     transportContext.getScheduler().schedule(() -> {
-                        transportContext.getRpcAwaitingAck().remove(requestId);
+                        TransportProtos.ToDeviceRpcRequestMsg rpcRequestMsg = transportContext.getRpcAwaitingAck().remove(requestId);
-                    }, Math.max(0, msg.getExpirationTime() - System.currentTimeMillis()), TimeUnit.MILLISECONDS);
+                        if (rpcRequestMsg != null) {
+                            transportService.process(state.getSession(), msg, RpcStatus.TIMEOUT, TransportServiceCallback.EMPTY);
+                        }
+                    }, Math.min(getTimeout(state, powerMode, profileSettings), 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, TransportServiceCallback.EMPTY);
+                            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)));
@@ -553,8 +580,12 @@ public class DefaultCoapClientContext implements CoapClientContext {
                     transportService.process(state.getSession(),
                             TransportProtos.ToDeviceRpcResponseMsg.newBuilder()
                                     .setRequestId(msg.getRequestId()).setError(error).build(), TransportServiceCallback.EMPTY);
-                } else if (msg.getPersisted() && !conRequest && sent) {
+                } else if (sent) {
-                    transportService.process(state.getSession(), msg, TransportServiceCallback.EMPTY);
+                    if (!conRequest) {
+                        transportService.process(state.getSession(), msg, RpcStatus.DELIVERED, TransportServiceCallback.EMPTY);
+                    } else if (msg.getPersisted()) {
+                        transportService.process(state.getSession(), msg, RpcStatus.SENT, TransportServiceCallback.EMPTY);
+                    }
                 }
             }
         }
@@ -723,7 +754,7 @@ public class DefaultCoapClientContext implements CoapClientContext {
     private void cancelRpcSubscription(TbCoapClientState state) {
         if (state.getRpc() != null) {
             clientsByToken.remove(state.getRpc().getToken());
-            CoapExchange exchange = state.getAttrs().getExchange();
+            CoapExchange exchange = state.getRpc().getExchange();
             state.setRpc(null);
             transportService.process(state.getSession(),
                     TransportProtos.SubscribeToRPCMsg.newBuilder().setUnsubscribe(true).build(),

common/transport/http/src/main/java/org/thingsboard/server/transport/http/DeviceApiController.java

@@ -407,7 +407,7 @@ public class DeviceApiController implements TbTransportService {
         public void onToDeviceRpcRequest(UUID sessionId, ToDeviceRpcRequestMsg msg) {
             log.trace("[{}] Received RPC command to device", sessionId);
             responseWriter.setResult(new ResponseEntity<>(JsonConverter.toJson(msg, true).toString(), HttpStatus.OK));
-            transportService.process(sessionInfo, msg, TransportServiceCallback.EMPTY);
+            transportService.process(sessionInfo, msg, RpcStatus.DELIVERED, TransportServiceCallback.EMPTY);
         }
 
         @Override

common/transport/lwm2m/src/main/java/org/thingsboard/server/transport/lwm2m/server/rpc/DefaultLwM2MRpcRequestHandler.java

@@ -21,7 +21,9 @@ import org.eclipse.leshan.core.ResponseCode;
 import org.springframework.stereotype.Service;
 import org.thingsboard.common.util.JacksonUtil;
 import org.thingsboard.server.common.data.StringUtils;
+import org.thingsboard.server.common.data.rpc.RpcStatus;
 import org.thingsboard.server.common.transport.TransportService;
+import org.thingsboard.server.common.transport.TransportServiceCallback;
 import org.thingsboard.server.gen.transport.TransportProtos;
 import org.thingsboard.server.queue.util.TbLwM2mTransportComponent;
 import org.thingsboard.server.transport.lwm2m.config.LwM2MTransportServerConfig;
@@ -158,6 +160,7 @@ public class DefaultLwM2MRpcRequestHandler implements LwM2MRpcRequestHandler {
                         throw new IllegalArgumentException("Unsupported operation: " + operationType.name());
                 }
             }
+            transportService.process(client.getSession(), rpcRequest, RpcStatus.SENT, TransportServiceCallback.EMPTY);
         } catch (IllegalArgumentException e) {
             this.sendErrorRpcResponse(sessionInfo, rpcRequest.getRequestId(), ResponseCode.BAD_REQUEST, e.getMessage());
         }

common/transport/lwm2m/src/main/java/org/thingsboard/server/transport/lwm2m/server/rpc/RpcDownlinkRequestCallbackProxy.java

@@ -19,6 +19,7 @@ import org.eclipse.leshan.core.ResponseCode;
 import org.eclipse.leshan.core.request.exception.ClientSleepingException;
 import org.thingsboard.common.util.JacksonUtil;
 import org.thingsboard.server.common.data.StringUtils;
+import org.thingsboard.server.common.data.rpc.RpcStatus;
 import org.thingsboard.server.common.transport.TransportService;
 import org.thingsboard.server.common.transport.TransportServiceCallback;
 import org.thingsboard.server.gen.transport.TransportProtos;
@@ -44,7 +45,7 @@ public abstract class RpcDownlinkRequestCallbackProxy<R, T> implements DownlinkR
 
     @Override
     public void onSuccess(R request, T response) {
-        transportService.process(client.getSession(), this.request, TransportServiceCallback.EMPTY);
+        transportService.process(client.getSession(), this.request, RpcStatus.DELIVERED, TransportServiceCallback.EMPTY);
         sendRpcReplyOnSuccess(response);
         if (callback != null) {
             callback.onSuccess(request, response);
@@ -61,7 +62,9 @@ public abstract class RpcDownlinkRequestCallbackProxy<R, T> implements DownlinkR
 
     @Override
     public void onError(String params, Exception e) {
-        if (!(e instanceof TimeoutException || e instanceof ClientSleepingException)) {
+        if (e instanceof TimeoutException) {
+            transportService.process(client.getSession(), this.request, RpcStatus.TIMEOUT, TransportServiceCallback.EMPTY);
+        } else if (!(e instanceof ClientSleepingException)) {
             sendRpcReplyOnError(e);
         }
         if (callback != null) {

common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/MqttTransportContext.java

@@ -68,4 +68,7 @@ public class MqttTransportContext extends TransportContext {
     @Value("${transport.mqtt.msg_queue_size_per_device_limit:100}")
     private int messageQueueSizePerDeviceLimit;
 
+    @Getter
+    @Value("${transport.mqtt.timeout:10000}")
+    private long timeout;
 }

common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/MqttTransportHandler.java

@@ -50,6 +50,7 @@ import org.thingsboard.server.common.data.TransportPayloadType;
 import org.thingsboard.server.common.data.device.profile.MqttTopics;
 import org.thingsboard.server.common.data.id.OtaPackageId;
 import org.thingsboard.server.common.data.ota.OtaPackageType;
+import org.thingsboard.server.common.data.rpc.RpcStatus;
 import org.thingsboard.server.common.msg.EncryptionUtil;
 import org.thingsboard.server.common.msg.tools.TbRateLimitsException;
 import org.thingsboard.server.common.transport.SessionMsgListener;
@@ -272,7 +273,7 @@ public class MqttTransportHandler extends ChannelInboundHandlerAdapter implement
                 int msgId = ((MqttPubAckMessage) msg).variableHeader().messageId();
                 TransportProtos.ToDeviceRpcRequestMsg rpcRequest = rpcAwaitingAck.remove(msgId);
                 if (rpcRequest != null) {
-                    transportService.process(deviceSessionCtx.getSessionInfo(), rpcRequest, TransportServiceCallback.EMPTY);
+                    transportService.process(deviceSessionCtx.getSessionInfo(), rpcRequest, RpcStatus.DELIVERED, TransportServiceCallback.EMPTY);
                 }
                 break;
             default:
@@ -849,20 +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);
+                        TransportProtos.ToDeviceRpcRequestMsg msg = rpcAwaitingAck.remove(msgId);
-                    }, Math.max(0, rpcRequest.getExpirationTime() - System.currentTimeMillis()), TimeUnit.MILLISECONDS);
+                        if (msg != null) {
+                            transportService.process(deviceSessionCtx.getSessionInfo(), rpcRequest, RpcStatus.TIMEOUT, TransportServiceCallback.EMPTY);
+                        }
+                    }, Math.max(0, Math.min(deviceSessionCtx.getContext().getTimeout(), rpcRequest.getExpirationTime() - System.currentTimeMillis())), TimeUnit.MILLISECONDS);
                 }
                 var cf = publish(payload, deviceSessionCtx);
-                if (rpcRequest.getPersisted() && !isAckExpected(payload)) {
+                cf.addListener(result -> {
-                    cf.addListener(result -> {
+                    if (result.cause() == null) {
-                        if (result.cause() == null) {
+                        if (!isAckExpected(payload)) {
-                            transportService.process(deviceSessionCtx.getSessionInfo(), rpcRequest, TransportServiceCallback.EMPTY);
+                            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(),

common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/GatewayDeviceSessionCtx.java

@@ -16,8 +16,10 @@
 package org.thingsboard.server.transport.mqtt.session;
 
 import io.netty.channel.ChannelFuture;
+import io.netty.handler.codec.mqtt.MqttMessage;
 import lombok.extern.slf4j.Slf4j;
 import org.thingsboard.server.common.data.DeviceProfile;
+import org.thingsboard.server.common.data.rpc.RpcStatus;
 import org.thingsboard.server.common.transport.SessionMsgListener;
 import org.thingsboard.server.common.transport.TransportService;
 import org.thingsboard.server.common.transport.TransportServiceCallback;
@@ -102,9 +104,14 @@ public class GatewayDeviceSessionCtx extends MqttDeviceAwareSessionContext imple
                     payload -> {
                         ChannelFuture channelFuture = parent.writeAndFlush(payload);
                         if (request.getPersisted()) {
-                            channelFuture.addListener(future -> {
+                            channelFuture.addListener(result -> {
-                                if (future.cause() == null) {
+                                if (result.cause() == null) {
-                                    transportService.process(getSessionInfo(), request, TransportServiceCallback.EMPTY);
+                                    if (!isAckExpected(payload)) {
+                                        transportService.process(getSessionInfo(), request, RpcStatus.DELIVERED, TransportServiceCallback.EMPTY);
+                                    } else if (request.getPersisted()) {
+                                        transportService.process(getSessionInfo(), request, RpcStatus.SENT, TransportServiceCallback.EMPTY);
+
+                                    }
                                 }
                             });
                         }
@@ -129,4 +136,8 @@ public class GatewayDeviceSessionCtx extends MqttDeviceAwareSessionContext imple
         // This feature is not supported in the TB IoT Gateway yet.
     }
 
+    private boolean isAckExpected(MqttMessage message) {
+        return message.fixedHeader().qosLevel().value() > 0;
+    }
+
 }

common/transport/snmp/src/main/java/org/thingsboard/server/transport/snmp/session/DeviceSessionContext.java

@@ -26,6 +26,7 @@ import org.thingsboard.server.common.data.DeviceProfile;
 import org.thingsboard.server.common.data.device.data.SnmpDeviceTransportConfiguration;
 import org.thingsboard.server.common.data.device.profile.SnmpDeviceProfileTransportConfiguration;
 import org.thingsboard.server.common.data.id.DeviceId;
+import org.thingsboard.server.common.data.rpc.RpcStatus;
 import org.thingsboard.server.common.transport.SessionMsgListener;
 import org.thingsboard.server.common.transport.TransportServiceCallback;
 import org.thingsboard.server.common.transport.session.DeviceAwareSessionContext;
@@ -142,7 +143,7 @@ public class DeviceSessionContext extends DeviceAwareSessionContext implements S
     public void onToDeviceRpcRequest(UUID sessionId, ToDeviceRpcRequestMsg toDeviceRequest) {
         log.trace("[{}] Received RPC command to device", sessionId);
         snmpTransportContext.getSnmpTransportService().onToDeviceRpcRequest(this, toDeviceRequest);
-        snmpTransportContext.getTransportService().process(getSessionInfo(), toDeviceRequest, TransportServiceCallback.EMPTY);
+        snmpTransportContext.getTransportService().process(getSessionInfo(), toDeviceRequest, RpcStatus.DELIVERED, TransportServiceCallback.EMPTY);
     }
 
     @Override

common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/TransportService.java

@@ -17,6 +17,7 @@ package org.thingsboard.server.common.transport;
 
 import org.thingsboard.server.common.data.DeviceProfile;
 import org.thingsboard.server.common.data.DeviceTransportType;
+import org.thingsboard.server.common.data.rpc.RpcStatus;
 import org.thingsboard.server.common.transport.auth.GetOrCreateDeviceFromGatewayResponse;
 import org.thingsboard.server.common.transport.auth.ValidateDeviceCredentialsResponse;
 import org.thingsboard.server.common.transport.service.SessionMetaData;
@@ -112,7 +113,7 @@ public interface TransportService {
 
     void process(SessionInfoProto sessionInfo, ToServerRpcRequestMsg msg, TransportServiceCallback<Void> callback);
 
-    void process(SessionInfoProto sessionInfo, ToDeviceRpcRequestMsg msg, TransportServiceCallback<Void> callback);
+    void process(SessionInfoProto sessionInfo, ToDeviceRpcRequestMsg msg, RpcStatus rpcStatus, TransportServiceCallback<Void> callback);
 
     void process(SessionInfoProto sessionInfo, SubscriptionInfoProto msg, TransportServiceCallback<Void> callback);
 

common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/service/DefaultTransportService.java

@@ -589,22 +589,18 @@ public class DefaultTransportService implements TransportService {
     }
 
     @Override
-    public void process(TransportProtos.SessionInfoProto sessionInfo, TransportProtos.ToDeviceRpcRequestMsg msg, TransportServiceCallback<Void> callback) {
+    public void process(TransportProtos.SessionInfoProto sessionInfo, TransportProtos.ToDeviceRpcRequestMsg msg, RpcStatus rpcStatus, TransportServiceCallback<Void> callback) {
-        if (msg.getPersisted()) {
+        TransportProtos.ToDeviceRpcResponseStatusMsg responseMsg = TransportProtos.ToDeviceRpcResponseStatusMsg.newBuilder()
-            RpcStatus status = msg.getOneway() ? RpcStatus.SUCCESSFUL : RpcStatus.DELIVERED;
+                .setRequestId(msg.getRequestId())
-
+                .setRequestIdLSB(msg.getRequestIdLSB())
-            TransportProtos.ToDevicePersistedRpcResponseMsg responseMsg = TransportProtos.ToDevicePersistedRpcResponseMsg.newBuilder()
+                .setRequestIdMSB(msg.getRequestIdMSB())
-                    .setRequestId(msg.getRequestId())
+                .setStatus(rpcStatus.name())
-                    .setRequestIdLSB(msg.getRequestIdLSB())
+                .build();
-                    .setRequestIdMSB(msg.getRequestIdMSB())
+
-                    .setStatus(status.name())
+        if (checkLimits(sessionInfo, responseMsg, callback)) {
-                    .build();
+            reportActivityInternal(sessionInfo);
-
+            sendToDeviceActor(sessionInfo, TransportToDeviceActorMsg.newBuilder().setSessionInfo(sessionInfo).setRpcResponseStatusMsg(responseMsg).build(),
-            if (checkLimits(sessionInfo, responseMsg, callback)) {
+                    new ApiStatsProxyCallback<>(getTenantId(sessionInfo), getCustomerId(sessionInfo), 1, TransportServiceCallback.EMPTY));
-                reportActivityInternal(sessionInfo);
-                sendToDeviceActor(sessionInfo, TransportToDeviceActorMsg.newBuilder().setSessionInfo(sessionInfo).setPersistedRpcResponseMsg(responseMsg).build(),
-                        new ApiStatsProxyCallback<>(getTenantId(sessionInfo), getCustomerId(sessionInfo), 1, TransportServiceCallback.EMPTY));
-            }
         }
     }
 

rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/RuleEngineDeviceRpcRequest.java

@@ -41,5 +41,5 @@ public final class RuleEngineDeviceRpcRequest {
     private final long expirationTime;
     private final boolean restApiCall;
     private final String additionalInfo;
-
+    private final Integer retries;
 }

rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/filter/TbMsgTypeSwitchNode.java

@@ -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", "RPC Deleted",
+        relationTypes = {"Post attributes", "Post telemetry", "RPC Request from Device", "RPC Request to Device", "RPC Queued", "RPC Sent", "RPC Delivered", "RPC Successful", "RPC Timeout", "RPC Expired", "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"},
@@ -97,12 +97,16 @@ public class TbMsgTypeSwitchNode implements TbNode {
             relationType = "Timeseries Deleted";
         } else if (msg.getType().equals(DataConstants.RPC_QUEUED)) {
             relationType = "RPC Queued";
+        } else if (msg.getType().equals(DataConstants.RPC_SENT)) {
+            relationType = "RPC Sent";
         } else if (msg.getType().equals(DataConstants.RPC_DELIVERED)) {
             relationType = "RPC Delivered";
         } else if (msg.getType().equals(DataConstants.RPC_SUCCESSFUL)) {
             relationType = "RPC Successful";
         } else if (msg.getType().equals(DataConstants.RPC_TIMEOUT)) {
             relationType = "RPC Timeout";
+        } else if (msg.getType().equals(DataConstants.RPC_EXPIRED)) {
+            relationType = "RPC Expired";
         } else if (msg.getType().equals(DataConstants.RPC_FAILED)) {
             relationType = "RPC Failed";
         } else if (msg.getType().equals(DataConstants.RPC_DELETED)) {

rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/rpc/TbSendRPCRequestNode.java

@@ -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.RETRIES);
+            Integer retries = !StringUtils.isEmpty(tmp) ? Integer.parseInt(tmp) : null;
+
             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)
+                    .retries(retries)
                     .restApiCall(restApiCall)
                     .persisted(persisted)
                     .additionalInfo(additionalInfo)