Browse Source

refactoring

pull/5091/head
YevhenBondarenko 5 years ago
parent
commit
5d6ec0dd0e
  1. 4
      application/src/main/java/org/thingsboard/server/actors/device/DeviceActorMessageProcessor.java
  2. 7
      common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/GatewayDeviceSessionCtx.java

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

@ -387,7 +387,7 @@ class DeviceActorMessageProcessor extends AbstractContextAwareMsgProcessor {
handleClaimDeviceMsg(context, sessionInfo, msg.getClaimDevice()); handleClaimDeviceMsg(context, sessionInfo, msg.getClaimDevice());
} }
if (msg.hasRpcResponseStatusMsg()) { if (msg.hasRpcResponseStatusMsg()) {
processPersistedRpcResponses(context, sessionInfo, msg.getRpcResponseStatusMsg()); processRpcResponseStatus(context, sessionInfo, msg.getRpcResponseStatusMsg());
} }
if (msg.hasUplinkNotificationMsg()) { if (msg.hasUplinkNotificationMsg()) {
processUplinkNotificationMsg(context, sessionInfo, msg.getUplinkNotificationMsg()); 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()); UUID rpcId = new UUID(responseMsg.getRequestIdMSB(), responseMsg.getRequestIdLSB());
RpcStatus status = RpcStatus.valueOf(responseMsg.getStatus()); RpcStatus status = RpcStatus.valueOf(responseMsg.getStatus());
ToDeviceRpcRequestMetadata md = toDeviceRpcPendingMap.get(responseMsg.getRequestId()); ToDeviceRpcRequestMetadata md = toDeviceRpcPendingMap.get(responseMsg.getRequestId());

7
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()) { if (request.getPersisted()) {
channelFuture.addListener(result -> { channelFuture.addListener(result -> {
if (result.cause() == null) { if (result.cause() == null) {
if (isAckExpected(payload)) { if (!isAckExpected(payload)) {
transportService.process(getSessionInfo(), request, RpcStatus.SENT, TransportServiceCallback.EMPTY);
} else {
transportService.process(getSessionInfo(), request, RpcStatus.DELIVERED, TransportServiceCallback.EMPTY); transportService.process(getSessionInfo(), request, RpcStatus.DELIVERED, TransportServiceCallback.EMPTY);
} else if (request.getPersisted()) {
transportService.process(getSessionInfo(), request, RpcStatus.SENT, TransportServiceCallback.EMPTY);
} }
} }
}); });

Loading…
Cancel
Save