diff --git a/application/src/main/java/org/thingsboard/server/actors/ActorSystemContext.java b/application/src/main/java/org/thingsboard/server/actors/ActorSystemContext.java index 5516762783..07ce35d116 100644 --- a/application/src/main/java/org/thingsboard/server/actors/ActorSystemContext.java +++ b/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; 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 6e3f7b8233..d16c1f10fd 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 @@ -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 attributeSubscriptions; private final Map rpcSubscriptions; private final Map 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 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; - for (Map.Entry entry : toDeviceRpcPendingMap.entrySet()) { - if (entry.getValue().getMsg().getMsg().getId().equals(msg.getRequestId())) { - requestId = entry.getKey(); + Map.Entry entry = null; + for (Map.Entry e : toDeviceRpcPendingMap.entrySet()) { + if (e.getValue().getMsg().getMsg().getId().equals(msg.getRequestId())) { + entry = e; break; } } - if (requestId != null) { - toDeviceRpcPendingMap.remove(requestId); + if (entry != null) { + if (entry.getValue().isDelivered()) { + toDeviceRpcPendingMap.remove(entry.getKey()); + } else { + Optional> 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 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> 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> processPendingRpc(TbActorCtx context, UUID sessionId, String nodeId, Set 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()) { - processPersistedRpcResponses(context, sessionInfo, msg.getPersistedRpcResponseMsg()); + if (msg.hasRpcResponseStatusMsg()) { + 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(); - 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 (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; - if (RpcStatus.DELIVERED.equals(status)) { - md = toDeviceRpcPendingMap.get(responseMsg.getRequestId()); - } else { - md = toDeviceRpcPendingMap.remove(responseMsg.getRequestId()); - } + ToDeviceRpcRequestMetadata md = toDeviceRpcPendingMap.get(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 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); diff --git a/application/src/main/java/org/thingsboard/server/actors/device/ToDeviceRpcRequestMetadata.java b/application/src/main/java/org/thingsboard/server/actors/device/ToDeviceRpcRequestMetadata.java index 44a2e0f3de..2b10b0cba0 100644 --- a/application/src/main/java/org/thingsboard/server/actors/device/ToDeviceRpcRequestMetadata.java +++ b/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; } diff --git a/application/src/main/java/org/thingsboard/server/controller/AbstractRpcController.java b/application/src/main/java/org/thingsboard/server/controller/AbstractRpcController.java index 43b6af1626..98294b241a 100644 --- a/application/src/main/java/org/thingsboard/server/controller/AbstractRpcController.java +++ b/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 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)); + 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 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); 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 46e49fcdc8..2b2ade02ee 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 @@ -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()); diff --git a/application/src/main/java/org/thingsboard/server/service/rpc/DefaultTbRuleEngineRpcService.java b/application/src/main/java/org/thingsboard/server/service/rpc/DefaultTbRuleEngineRpcService.java index 230f6e759e..eca8864fda 100644 --- a/application/src/main/java/org/thingsboard/server/service/rpc/DefaultTbRuleEngineRpcService.java +++ b/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 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); diff --git a/application/src/main/resources/thingsboard.yml b/application/src/main/resources/thingsboard.yml index ced335aea4..7fc70e9fb1 100644 --- a/application/src/main/resources/thingsboard.yml +++ b/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}" diff --git a/application/src/test/java/org/thingsboard/server/transport/mqtt/rpc/AbstractMqttServerSideRpcDefaultIntegrationTest.java b/application/src/test/java/org/thingsboard/server/transport/mqtt/rpc/AbstractMqttServerSideRpcDefaultIntegrationTest.java index b5f005cd00..ea19ac6835 100644 --- a/application/src/test/java/org/thingsboard/server/transport/mqtt/rpc/AbstractMqttServerSideRpcDefaultIntegrationTest.java +++ b/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"); diff --git a/application/src/test/java/org/thingsboard/server/transport/mqtt/rpc/AbstractMqttServerSideRpcIntegrationTest.java b/application/src/test/java/org/thingsboard/server/transport/mqtt/rpc/AbstractMqttServerSideRpcIntegrationTest.java index 9f83f24bcb..e0ad7f79fc 100644 --- a/application/src/test/java/org/thingsboard/server/transport/mqtt/rpc/AbstractMqttServerSideRpcIntegrationTest.java +++ b/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 expected = new ArrayList<>(); + List 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 expected; + + TestSequenceMqttCallback(MqttAsyncClient client, CountDownLatch latch, List 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) { + + } + } } diff --git a/application/src/test/resources/application-test.properties b/application/src/test/resources/application-test.properties index cb412f77ac..f65bce749e 100644 --- a/application/src/test/resources/application-test.properties +++ b/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 \ No newline at end of file +transport.lwm2m.bootstrap.security.key_password=server +actors.rpc.sequential=true \ No newline at end of file diff --git a/common/cluster-api/src/main/proto/queue.proto b/common/cluster-api/src/main/proto/queue.proto index b496eaaf22..5b220974ef 100644 --- a/common/cluster-api/src/main/proto/queue.proto +++ b/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; } 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 ae51749056..707a50dfb3 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 @@ -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"; diff --git a/common/data/src/main/java/org/thingsboard/server/common/data/rpc/RpcStatus.java b/common/data/src/main/java/org/thingsboard/server/common/data/rpc/RpcStatus.java index c80d0c5993..43592fde0c 100644 --- a/common/data/src/main/java/org/thingsboard/server/common/data/rpc/RpcStatus.java +++ b/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 } diff --git a/common/message/src/main/java/org/thingsboard/server/common/msg/rpc/ToDeviceRpcRequest.java b/common/message/src/main/java/org/thingsboard/server/common/msg/rpc/ToDeviceRpcRequest.java index f9bb2b3810..f68f2cfdb8 100644 --- a/common/message/src/main/java/org/thingsboard/server/common/msg/rpc/ToDeviceRpcRequest.java +++ b/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; } diff --git a/common/transport/coap/src/main/java/org/thingsboard/server/transport/coap/client/DefaultCoapClientContext.java b/common/transport/coap/src/main/java/org/thingsboard/server/transport/coap/client/DefaultCoapClientContext.java index 4ea4a167aa..082a94dbe6 100644 --- a/common/transport/coap/src/main/java/org/thingsboard/server/transport/coap/client/DefaultCoapClientContext.java +++ b/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; - 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; - } + long timeout = getTimeout(client, powerMode, profileSettings); Future 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); - }, 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.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) { - transportService.process(state.getSession(), msg, TransportServiceCallback.EMPTY); + } 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); + } } } } @@ -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(), diff --git a/common/transport/http/src/main/java/org/thingsboard/server/transport/http/DeviceApiController.java b/common/transport/http/src/main/java/org/thingsboard/server/transport/http/DeviceApiController.java index cfb48656ed..dc1c49a3c0 100644 --- a/common/transport/http/src/main/java/org/thingsboard/server/transport/http/DeviceApiController.java +++ b/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 diff --git a/common/transport/lwm2m/src/main/java/org/thingsboard/server/transport/lwm2m/server/rpc/DefaultLwM2MRpcRequestHandler.java b/common/transport/lwm2m/src/main/java/org/thingsboard/server/transport/lwm2m/server/rpc/DefaultLwM2MRpcRequestHandler.java index 7ac8026a4c..8135b5aae0 100644 --- a/common/transport/lwm2m/src/main/java/org/thingsboard/server/transport/lwm2m/server/rpc/DefaultLwM2MRpcRequestHandler.java +++ b/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()); } diff --git a/common/transport/lwm2m/src/main/java/org/thingsboard/server/transport/lwm2m/server/rpc/RpcDownlinkRequestCallbackProxy.java b/common/transport/lwm2m/src/main/java/org/thingsboard/server/transport/lwm2m/server/rpc/RpcDownlinkRequestCallbackProxy.java index c94c9c5805..a65ac4dfaf 100644 --- a/common/transport/lwm2m/src/main/java/org/thingsboard/server/transport/lwm2m/server/rpc/RpcDownlinkRequestCallbackProxy.java +++ b/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 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 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) { diff --git a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/MqttTransportContext.java b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/MqttTransportContext.java index 11b46696da..1335b6105b 100644 --- a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/MqttTransportContext.java +++ b/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; } diff --git a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/MqttTransportHandler.java b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/MqttTransportHandler.java index fa8c2599eb..d8d06a3173 100644 --- a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/MqttTransportHandler.java +++ b/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); - }, 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(deviceSessionCtx.getContext().getTimeout(), rpcRequest.getExpirationTime() - System.currentTimeMillis())), TimeUnit.MILLISECONDS); } var cf = publish(payload, deviceSessionCtx); - if (rpcRequest.getPersisted() && !isAckExpected(payload)) { - cf.addListener(result -> { - if (result.cause() == null) { - transportService.process(deviceSessionCtx.getSessionInfo(), rpcRequest, 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(), diff --git a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/GatewayDeviceSessionCtx.java b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/GatewayDeviceSessionCtx.java index 21086d75d8..f41c5668cd 100644 --- a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/GatewayDeviceSessionCtx.java +++ b/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 -> { - if (future.cause() == null) { - transportService.process(getSessionInfo(), request, TransportServiceCallback.EMPTY); + channelFuture.addListener(result -> { + if (result.cause() == null) { + 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; + } + } diff --git a/common/transport/snmp/src/main/java/org/thingsboard/server/transport/snmp/session/DeviceSessionContext.java b/common/transport/snmp/src/main/java/org/thingsboard/server/transport/snmp/session/DeviceSessionContext.java index 1927aadc56..6d9238d6bc 100644 --- a/common/transport/snmp/src/main/java/org/thingsboard/server/transport/snmp/session/DeviceSessionContext.java +++ b/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 diff --git a/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/TransportService.java b/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/TransportService.java index 91f3c2ef02..28ad7ddb9b 100644 --- a/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/TransportService.java +++ b/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 callback); - void process(SessionInfoProto sessionInfo, ToDeviceRpcRequestMsg msg, TransportServiceCallback callback); + void process(SessionInfoProto sessionInfo, ToDeviceRpcRequestMsg msg, RpcStatus rpcStatus, TransportServiceCallback callback); void process(SessionInfoProto sessionInfo, SubscriptionInfoProto msg, TransportServiceCallback callback); diff --git a/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/service/DefaultTransportService.java b/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/service/DefaultTransportService.java index 16317b51ea..bdc03d767c 100644 --- a/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/service/DefaultTransportService.java +++ b/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 callback) { - if (msg.getPersisted()) { - RpcStatus status = msg.getOneway() ? RpcStatus.SUCCESSFUL : RpcStatus.DELIVERED; - - TransportProtos.ToDevicePersistedRpcResponseMsg responseMsg = TransportProtos.ToDevicePersistedRpcResponseMsg.newBuilder() - .setRequestId(msg.getRequestId()) - .setRequestIdLSB(msg.getRequestIdLSB()) - .setRequestIdMSB(msg.getRequestIdMSB()) - .setStatus(status.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)); - } + public void process(TransportProtos.SessionInfoProto sessionInfo, TransportProtos.ToDeviceRpcRequestMsg msg, RpcStatus rpcStatus, TransportServiceCallback callback) { + 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)); } } diff --git a/rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/RuleEngineDeviceRpcRequest.java b/rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/RuleEngineDeviceRpcRequest.java index afdee27b74..903cb291c2 100644 --- a/rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/RuleEngineDeviceRpcRequest.java +++ b/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; } diff --git a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/filter/TbMsgTypeSwitchNode.java b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/filter/TbMsgTypeSwitchNode.java index cda3269791..6303a888df 100644 --- a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/filter/TbMsgTypeSwitchNode.java +++ b/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)) { 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 a6d0a23ee0..1d0b727e5e 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 @@ -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)