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