From 5d6ec0dd0e67b1de1e4793ef706a7d753ef51bdd Mon Sep 17 00:00:00 2001 From: YevhenBondarenko Date: Wed, 18 Aug 2021 10:48:04 +0300 Subject: [PATCH] refactoring --- .../server/actors/device/DeviceActorMessageProcessor.java | 4 ++-- .../transport/mqtt/session/GatewayDeviceSessionCtx.java | 7 ++++--- 2 files changed, 6 insertions(+), 5 deletions(-) diff --git a/application/src/main/java/org/thingsboard/server/actors/device/DeviceActorMessageProcessor.java b/application/src/main/java/org/thingsboard/server/actors/device/DeviceActorMessageProcessor.java index 24f99d625c..5042c5b2d8 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 @@ -387,7 +387,7 @@ class DeviceActorMessageProcessor extends AbstractContextAwareMsgProcessor { handleClaimDeviceMsg(context, sessionInfo, msg.getClaimDevice()); } if (msg.hasRpcResponseStatusMsg()) { - processPersistedRpcResponses(context, sessionInfo, msg.getRpcResponseStatusMsg()); + processRpcResponseStatus(context, sessionInfo, msg.getRpcResponseStatusMsg()); } if (msg.hasUplinkNotificationMsg()) { processUplinkNotificationMsg(context, sessionInfo, msg.getUplinkNotificationMsg()); @@ -572,7 +572,7 @@ class DeviceActorMessageProcessor extends AbstractContextAwareMsgProcessor { } } - private void processPersistedRpcResponses(TbActorCtx context, SessionInfoProto sessionInfo, ToDeviceRpcResponseStatusMsg 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()); 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 2713a36bb8..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 @@ -106,10 +106,11 @@ public class GatewayDeviceSessionCtx extends MqttDeviceAwareSessionContext imple if (request.getPersisted()) { channelFuture.addListener(result -> { if (result.cause() == null) { - if (isAckExpected(payload)) { - transportService.process(getSessionInfo(), request, RpcStatus.SENT, TransportServiceCallback.EMPTY); - } else { + if (!isAckExpected(payload)) { transportService.process(getSessionInfo(), request, RpcStatus.DELIVERED, TransportServiceCallback.EMPTY); + } else if (request.getPersisted()) { + transportService.process(getSessionInfo(), request, RpcStatus.SENT, TransportServiceCallback.EMPTY); + } } });