From 7fa6cbd53f7f03305190562f7c1c88aebe6b3f82 Mon Sep 17 00:00:00 2001 From: Volodymyr Babak Date: Tue, 1 Jun 2021 16:16:51 +0300 Subject: [PATCH] Resend only failed messages and not the entire pack --- .../service/edge/rpc/EdgeGrpcSession.java | 28 +++++++++++++++---- .../processor/AdminSettingsEdgeProcessor.java | 2 ++ .../rpc/processor/AlarmEdgeProcessor.java | 1 + .../rpc/processor/AssetEdgeProcessor.java | 3 ++ .../rpc/processor/CustomerEdgeProcessor.java | 3 ++ .../rpc/processor/DashboardEdgeProcessor.java | 3 ++ .../rpc/processor/DeviceEdgeProcessor.java | 7 ++++- .../processor/DeviceProfileEdgeProcessor.java | 3 ++ .../rpc/processor/EntityEdgeProcessor.java | 3 ++ .../processor/EntityViewEdgeProcessor.java | 3 ++ .../rpc/processor/RelationEdgeProcessor.java | 2 ++ .../rpc/processor/RuleChainEdgeProcessor.java | 4 +++ .../rpc/processor/TelemetryEdgeProcessor.java | 2 ++ .../edge/rpc/processor/UserEdgeProcessor.java | 4 +++ .../processor/WidgetBundleEdgeProcessor.java | 3 ++ .../processor/WidgetTypeEdgeProcessor.java | 3 ++ .../server/common/data/EdgeUtils.java | 6 ++++ common/edge-api/src/main/proto/edge.proto | 2 ++ 18 files changed, 75 insertions(+), 7 deletions(-) diff --git a/application/src/main/java/org/thingsboard/server/service/edge/rpc/EdgeGrpcSession.java b/application/src/main/java/org/thingsboard/server/service/edge/rpc/EdgeGrpcSession.java index b7127d4ee2..f3626c0650 100644 --- a/application/src/main/java/org/thingsboard/server/service/edge/rpc/EdgeGrpcSession.java +++ b/application/src/main/java/org/thingsboard/server/service/edge/rpc/EdgeGrpcSession.java @@ -27,6 +27,7 @@ import lombok.extern.slf4j.Slf4j; import org.checkerframework.checker.nullness.qual.Nullable; import org.thingsboard.common.util.JacksonUtil; import org.thingsboard.server.common.data.DataConstants; +import org.thingsboard.server.common.data.EdgeUtils; import org.thingsboard.server.common.data.edge.Edge; import org.thingsboard.server.common.data.edge.EdgeEvent; import org.thingsboard.server.common.data.edge.EdgeEventActionType; @@ -83,7 +84,9 @@ import org.thingsboard.server.service.edge.rpc.fetch.TenantWidgetsBundlesEdgeEve import java.io.Closeable; import java.util.ArrayList; import java.util.Collections; +import java.util.HashMap; import java.util.List; +import java.util.Map; import java.util.Objects; import java.util.Optional; import java.util.UUID; @@ -108,6 +111,8 @@ public final class EdgeGrpcSession implements Closeable { private final Consumer sessionCloseListener; private final ObjectMapper mapper; + private final Map pendingMsgsMap; + private EdgeContextComponent ctx; private Edge edge; private StreamObserver inputStream; @@ -128,6 +133,7 @@ public final class EdgeGrpcSession implements Closeable { this.sessionCloseListener = sessionCloseListener; this.mapper = mapper; this.syncExecutorService = syncExecutorService; + this.pendingMsgsMap = new HashMap<>(); initInputStream(); } @@ -214,6 +220,7 @@ public final class EdgeGrpcSession implements Closeable { startProcessingEdgeEvents(new DashboardsEdgeEventFetcher(ctx.getDashboardService())); DownlinkMsg syncCompleteDownlinkMsg = DownlinkMsg.newBuilder() + .setDownlinkMsgId(EdgeUtils.nextPositiveInt()) .setSyncCompletedMsg(SyncCompletedMsg.newBuilder().build()) .build(); sendDownlinkMsgsPack(Collections.singletonList(syncCompleteDownlinkMsg)); @@ -228,7 +235,9 @@ public final class EdgeGrpcSession implements Closeable { Futures.addCallback(future, new FutureCallback<>() { @Override public void onSuccess(@Nullable List result) { - UplinkResponseMsg uplinkResponseMsg = UplinkResponseMsg.newBuilder().setSuccess(true).build(); + UplinkResponseMsg uplinkResponseMsg = UplinkResponseMsg.newBuilder() + .setUplinkMsgId(uplinkMsg.getUplinkMsgId()) + .setSuccess(true).build(); sendDownlinkMsg(ResponseMsg.newBuilder() .setUplinkResponseMsg(uplinkResponseMsg) .build()); @@ -236,7 +245,9 @@ public final class EdgeGrpcSession implements Closeable { @Override public void onFailure(Throwable t) { - UplinkResponseMsg uplinkResponseMsg = UplinkResponseMsg.newBuilder().setSuccess(false).setErrorMsg(t.getMessage()).build(); + UplinkResponseMsg uplinkResponseMsg = UplinkResponseMsg.newBuilder() + .setUplinkMsgId(uplinkMsg.getUplinkMsgId()) + .setSuccess(false).setErrorMsg(t.getMessage()).build(); sendDownlinkMsg(ResponseMsg.newBuilder() .setUplinkResponseMsg(uplinkResponseMsg) .build()); @@ -247,6 +258,7 @@ public final class EdgeGrpcSession implements Closeable { private void onDownlinkResponse(DownlinkResponseMsg msg) { try { if (msg.getSuccess()) { + pendingMsgsMap.remove(msg.getDownlinkMsgId()); log.debug("[{}] Msg has been processed successfully! {}", edge.getRoutingKey(), msg); } else { log.error("[{}] Msg processing failed! Error msg: {}", edge.getRoutingKey(), msg.getErrorMsg()); @@ -325,17 +337,19 @@ public final class EdgeGrpcSession implements Closeable { private boolean sendDownlinkMsgsPack(List downlinkMsgsPack) throws InterruptedException { boolean success; + pendingMsgsMap.clear(); + downlinkMsgsPack.forEach(msg -> pendingMsgsMap.put(msg.getDownlinkMsgId(), msg)); do { - log.trace("[{}] [{}] downlink msg(s) are going to be send.", this.sessionId, downlinkMsgsPack.size()); - latch = new CountDownLatch(downlinkMsgsPack.size()); - for (DownlinkMsg downlinkMsg : downlinkMsgsPack) { + log.trace("[{}] [{}] downlink msg(s) are going to be send.", this.sessionId, pendingMsgsMap.values().size()); + latch = new CountDownLatch(pendingMsgsMap.values().size()); + for (DownlinkMsg downlinkMsg : pendingMsgsMap.values()) { sendDownlinkMsg(ResponseMsg.newBuilder() .setDownlinkMsg(downlinkMsg) .build()); } success = latch.await(10, TimeUnit.SECONDS); if (!success) { - log.warn("[{}] Failed to deliver the batch: {}", this.sessionId, downlinkMsgsPack); + log.warn("[{}] Failed to deliver the batch: {}", this.sessionId, pendingMsgsMap.values()); } if (isConnected() && !success) { try { @@ -383,6 +397,8 @@ public final class EdgeGrpcSession implements Closeable { case RPC_CALL: downlinkMsg = ctx.getDeviceProcessor().processRpcCallMsgToEdge(edgeEvent); break; + default: + log.warn("[{}][{}] Unsupported action type [{}]", edge.getTenantId(), this.sessionId, edgeEvent.getAction()); } } catch (Exception e) { log.error("[{}][{}] Exception during converting edge event to downlink msg", edge.getTenantId(), this.sessionId, e); diff --git a/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/AdminSettingsEdgeProcessor.java b/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/AdminSettingsEdgeProcessor.java index e39b3eda4c..9eda3e3f81 100644 --- a/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/AdminSettingsEdgeProcessor.java +++ b/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/AdminSettingsEdgeProcessor.java @@ -18,6 +18,7 @@ package org.thingsboard.server.service.edge.rpc.processor; import lombok.extern.slf4j.Slf4j; import org.springframework.stereotype.Component; import org.thingsboard.server.common.data.AdminSettings; +import org.thingsboard.server.common.data.EdgeUtils; import org.thingsboard.server.common.data.edge.EdgeEvent; import org.thingsboard.server.gen.edge.AdminSettingsUpdateMsg; import org.thingsboard.server.gen.edge.DownlinkMsg; @@ -34,6 +35,7 @@ public class AdminSettingsEdgeProcessor extends BaseEdgeProcessor { AdminSettings adminSettings = mapper.convertValue(edgeEvent.getBody(), AdminSettings.class); AdminSettingsUpdateMsg t = adminSettingsMsgConstructor.constructAdminSettingsUpdateMsg(adminSettings); return DownlinkMsg.newBuilder() + .setDownlinkMsgId(EdgeUtils.nextPositiveInt()) .addAllAdminSettingsUpdateMsg(Collections.singletonList(t)) .build(); } diff --git a/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/AlarmEdgeProcessor.java b/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/AlarmEdgeProcessor.java index 5893840728..9113c39ff0 100644 --- a/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/AlarmEdgeProcessor.java +++ b/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/AlarmEdgeProcessor.java @@ -121,6 +121,7 @@ public class AlarmEdgeProcessor extends BaseEdgeProcessor { Alarm alarm = alarmService.findAlarmByIdAsync(edgeEvent.getTenantId(), alarmId).get(); if (alarm != null) { downlinkMsg = DownlinkMsg.newBuilder() + .setDownlinkMsgId(EdgeUtils.nextPositiveInt()) .addAllAlarmUpdateMsg(Collections.singletonList(alarmMsgConstructor.constructAlarmUpdatedMsg(edge.getTenantId(), msgType, alarm))) .build(); } diff --git a/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/AssetEdgeProcessor.java b/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/AssetEdgeProcessor.java index cf0180c85f..bd108e04e6 100644 --- a/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/AssetEdgeProcessor.java +++ b/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/AssetEdgeProcessor.java @@ -17,6 +17,7 @@ package org.thingsboard.server.service.edge.rpc.processor; import lombok.extern.slf4j.Slf4j; import org.springframework.stereotype.Component; +import org.thingsboard.server.common.data.EdgeUtils; import org.thingsboard.server.common.data.asset.Asset; import org.thingsboard.server.common.data.edge.Edge; import org.thingsboard.server.common.data.edge.EdgeEvent; @@ -50,6 +51,7 @@ public class AssetEdgeProcessor extends BaseEdgeProcessor { AssetUpdateMsg assetUpdateMsg = assetMsgConstructor.constructAssetUpdatedMsg(msgType, asset, customerId); downlinkMsg = DownlinkMsg.newBuilder() + .setDownlinkMsgId(EdgeUtils.nextPositiveInt()) .addAllAssetUpdateMsg(Collections.singletonList(assetUpdateMsg)) .build(); } @@ -59,6 +61,7 @@ public class AssetEdgeProcessor extends BaseEdgeProcessor { AssetUpdateMsg assetUpdateMsg = assetMsgConstructor.constructAssetDeleteMsg(assetId); downlinkMsg = DownlinkMsg.newBuilder() + .setDownlinkMsgId(EdgeUtils.nextPositiveInt()) .addAllAssetUpdateMsg(Collections.singletonList(assetUpdateMsg)) .build(); break; diff --git a/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/CustomerEdgeProcessor.java b/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/CustomerEdgeProcessor.java index 00de4eb576..d8c1c00c2e 100644 --- a/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/CustomerEdgeProcessor.java +++ b/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/CustomerEdgeProcessor.java @@ -18,6 +18,7 @@ package org.thingsboard.server.service.edge.rpc.processor; import lombok.extern.slf4j.Slf4j; import org.springframework.stereotype.Component; import org.thingsboard.server.common.data.Customer; +import org.thingsboard.server.common.data.EdgeUtils; import org.thingsboard.server.common.data.edge.Edge; import org.thingsboard.server.common.data.edge.EdgeEvent; import org.thingsboard.server.common.data.edge.EdgeEventActionType; @@ -53,6 +54,7 @@ public class CustomerEdgeProcessor extends BaseEdgeProcessor { CustomerUpdateMsg customerUpdateMsg = customerMsgConstructor.constructCustomerUpdatedMsg(msgType, customer); downlinkMsg = DownlinkMsg.newBuilder() + .setDownlinkMsgId(EdgeUtils.nextPositiveInt()) .addAllCustomerUpdateMsg(Collections.singletonList(customerUpdateMsg)) .build(); } @@ -61,6 +63,7 @@ public class CustomerEdgeProcessor extends BaseEdgeProcessor { CustomerUpdateMsg customerUpdateMsg = customerMsgConstructor.constructCustomerDeleteMsg(customerId); downlinkMsg = DownlinkMsg.newBuilder() + .setDownlinkMsgId(EdgeUtils.nextPositiveInt()) .addAllCustomerUpdateMsg(Collections.singletonList(customerUpdateMsg)) .build(); break; diff --git a/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/DashboardEdgeProcessor.java b/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/DashboardEdgeProcessor.java index 730e928a2d..585298af03 100644 --- a/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/DashboardEdgeProcessor.java +++ b/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/DashboardEdgeProcessor.java @@ -18,6 +18,7 @@ package org.thingsboard.server.service.edge.rpc.processor; import lombok.extern.slf4j.Slf4j; import org.springframework.stereotype.Component; import org.thingsboard.server.common.data.Dashboard; +import org.thingsboard.server.common.data.EdgeUtils; import org.thingsboard.server.common.data.edge.Edge; import org.thingsboard.server.common.data.edge.EdgeEvent; import org.thingsboard.server.common.data.edge.EdgeEventActionType; @@ -53,6 +54,7 @@ public class DashboardEdgeProcessor extends BaseEdgeProcessor { DashboardUpdateMsg dashboardUpdateMsg = dashboardMsgConstructor.constructDashboardUpdatedMsg(msgType, dashboard, customerId); downlinkMsg = DownlinkMsg.newBuilder() + .setDownlinkMsgId(EdgeUtils.nextPositiveInt()) .addAllDashboardUpdateMsg(Collections.singletonList(dashboardUpdateMsg)) .build(); } @@ -62,6 +64,7 @@ public class DashboardEdgeProcessor extends BaseEdgeProcessor { DashboardUpdateMsg dashboardUpdateMsg = dashboardMsgConstructor.constructDashboardDeleteMsg(dashboardId); downlinkMsg = DownlinkMsg.newBuilder() + .setDownlinkMsgId(EdgeUtils.nextPositiveInt()) .addAllDashboardUpdateMsg(Collections.singletonList(dashboardUpdateMsg)) .build(); break; diff --git a/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/DeviceEdgeProcessor.java b/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/DeviceEdgeProcessor.java index f540186976..013e50482b 100644 --- a/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/DeviceEdgeProcessor.java +++ b/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/DeviceEdgeProcessor.java @@ -24,10 +24,12 @@ import lombok.extern.slf4j.Slf4j; import org.apache.commons.lang3.RandomStringUtils; import org.apache.commons.lang3.StringUtils; import org.springframework.stereotype.Component; +import org.thingsboard.common.util.JacksonUtil; import org.thingsboard.rule.engine.api.RpcError; import org.thingsboard.server.common.data.Customer; import org.thingsboard.server.common.data.DataConstants; import org.thingsboard.server.common.data.Device; +import org.thingsboard.server.common.data.EdgeUtils; import org.thingsboard.server.common.data.edge.Edge; import org.thingsboard.server.common.data.edge.EdgeEvent; import org.thingsboard.server.common.data.edge.EdgeEventActionType; @@ -48,7 +50,6 @@ import org.thingsboard.server.common.msg.TbMsg; import org.thingsboard.server.common.msg.TbMsgDataType; import org.thingsboard.server.common.msg.TbMsgMetaData; import org.thingsboard.server.dao.model.ModelConstants; -import org.thingsboard.common.util.JacksonUtil; import org.thingsboard.server.gen.edge.DeviceCredentialsUpdateMsg; import org.thingsboard.server.gen.edge.DeviceRpcCallMsg; import org.thingsboard.server.gen.edge.DeviceUpdateMsg; @@ -308,6 +309,7 @@ public class DeviceEdgeProcessor extends BaseEdgeProcessor { DeviceUpdateMsg deviceUpdateMsg = deviceMsgConstructor.constructDeviceUpdatedMsg(msgType, device, customerId, null); downlinkMsg = DownlinkMsg.newBuilder() + .setDownlinkMsgId(EdgeUtils.nextPositiveInt()) .addAllDeviceUpdateMsg(Collections.singletonList(deviceUpdateMsg)) .build(); } @@ -317,6 +319,7 @@ public class DeviceEdgeProcessor extends BaseEdgeProcessor { DeviceUpdateMsg deviceUpdateMsg = deviceMsgConstructor.constructDeviceDeleteMsg(deviceId); downlinkMsg = DownlinkMsg.newBuilder() + .setDownlinkMsgId(EdgeUtils.nextPositiveInt()) .addAllDeviceUpdateMsg(Collections.singletonList(deviceUpdateMsg)) .build(); break; @@ -326,6 +329,7 @@ public class DeviceEdgeProcessor extends BaseEdgeProcessor { DeviceCredentialsUpdateMsg deviceCredentialsUpdateMsg = deviceMsgConstructor.constructDeviceCredentialsUpdatedMsg(deviceCredentials); downlinkMsg = DownlinkMsg.newBuilder() + .setDownlinkMsgId(EdgeUtils.nextPositiveInt()) .addAllDeviceCredentialsUpdateMsg(Collections.singletonList(deviceCredentialsUpdateMsg)) .build(); } @@ -339,6 +343,7 @@ public class DeviceEdgeProcessor extends BaseEdgeProcessor { DeviceRpcCallMsg deviceRpcCallMsg = deviceMsgConstructor.constructDeviceRpcCallMsg(edgeEvent.getEntityId(), edgeEvent.getBody()); return DownlinkMsg.newBuilder() + .setDownlinkMsgId(EdgeUtils.nextPositiveInt()) .addAllDeviceRpcCallMsg(Collections.singletonList(deviceRpcCallMsg)) .build(); } diff --git a/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/DeviceProfileEdgeProcessor.java b/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/DeviceProfileEdgeProcessor.java index d66dc2b67c..a9bc4841df 100644 --- a/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/DeviceProfileEdgeProcessor.java +++ b/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/DeviceProfileEdgeProcessor.java @@ -18,6 +18,7 @@ package org.thingsboard.server.service.edge.rpc.processor; import lombok.extern.slf4j.Slf4j; import org.springframework.stereotype.Component; import org.thingsboard.server.common.data.DeviceProfile; +import org.thingsboard.server.common.data.EdgeUtils; import org.thingsboard.server.common.data.edge.EdgeEvent; import org.thingsboard.server.common.data.edge.EdgeEventActionType; import org.thingsboard.server.common.data.id.DeviceProfileId; @@ -44,6 +45,7 @@ public class DeviceProfileEdgeProcessor extends BaseEdgeProcessor { DeviceProfileUpdateMsg deviceProfileUpdateMsg = deviceProfileMsgConstructor.constructDeviceProfileUpdatedMsg(msgType, deviceProfile); downlinkMsg = DownlinkMsg.newBuilder() + .setDownlinkMsgId(EdgeUtils.nextPositiveInt()) .addAllDeviceProfileUpdateMsg(Collections.singletonList(deviceProfileUpdateMsg)) .build(); } @@ -52,6 +54,7 @@ public class DeviceProfileEdgeProcessor extends BaseEdgeProcessor { DeviceProfileUpdateMsg deviceProfileUpdateMsg = deviceProfileMsgConstructor.constructDeviceProfileDeleteMsg(deviceProfileId); downlinkMsg = DownlinkMsg.newBuilder() + .setDownlinkMsgId(EdgeUtils.nextPositiveInt()) .addAllDeviceProfileUpdateMsg(Collections.singletonList(deviceProfileUpdateMsg)) .build(); break; diff --git a/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/EntityEdgeProcessor.java b/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/EntityEdgeProcessor.java index feb50ce750..b4a591e4cc 100644 --- a/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/EntityEdgeProcessor.java +++ b/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/EntityEdgeProcessor.java @@ -22,6 +22,7 @@ import lombok.extern.slf4j.Slf4j; import org.checkerframework.checker.nullness.qual.Nullable; import org.springframework.stereotype.Component; import org.thingsboard.server.common.data.Device; +import org.thingsboard.server.common.data.EdgeUtils; import org.thingsboard.server.common.data.edge.Edge; import org.thingsboard.server.common.data.edge.EdgeEvent; import org.thingsboard.server.common.data.edge.EdgeEventActionType; @@ -66,6 +67,7 @@ public class EntityEdgeProcessor extends BaseEdgeProcessor { DeviceUpdateMsg d = deviceMsgConstructor .constructDeviceUpdatedMsg(UpdateMsgType.ENTITY_MERGE_RPC_MESSAGE, device, customerId, conflictName); downlinkMsg = DownlinkMsg.newBuilder() + .setDownlinkMsgId(EdgeUtils.nextPositiveInt()) .addAllDeviceUpdateMsg(Collections.singletonList(d)) .build(); } @@ -81,6 +83,7 @@ public class EntityEdgeProcessor extends BaseEdgeProcessor { .setDeviceIdLSB(deviceId.getId().getLeastSignificantBits()) .build(); DownlinkMsg.Builder builder = DownlinkMsg.newBuilder() + .setDownlinkMsgId(EdgeUtils.nextPositiveInt()) .addAllDeviceCredentialsRequestMsg(Collections.singletonList(deviceCredentialsRequestMsg)); downlinkMsg = builder.build(); } diff --git a/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/EntityViewEdgeProcessor.java b/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/EntityViewEdgeProcessor.java index ab0e2e868f..77eebccb8d 100644 --- a/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/EntityViewEdgeProcessor.java +++ b/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/EntityViewEdgeProcessor.java @@ -17,6 +17,7 @@ package org.thingsboard.server.service.edge.rpc.processor; import lombok.extern.slf4j.Slf4j; import org.springframework.stereotype.Component; +import org.thingsboard.server.common.data.EdgeUtils; import org.thingsboard.server.common.data.EntityView; import org.thingsboard.server.common.data.edge.Edge; import org.thingsboard.server.common.data.edge.EdgeEvent; @@ -50,6 +51,7 @@ public class EntityViewEdgeProcessor extends BaseEdgeProcessor { EntityViewUpdateMsg entityViewUpdateMsg = entityViewMsgConstructor.constructEntityViewUpdatedMsg(msgType, entityView, customerId); downlinkMsg = DownlinkMsg.newBuilder() + .setDownlinkMsgId(EdgeUtils.nextPositiveInt()) .addAllEntityViewUpdateMsg(Collections.singletonList(entityViewUpdateMsg)) .build(); } @@ -59,6 +61,7 @@ public class EntityViewEdgeProcessor extends BaseEdgeProcessor { EntityViewUpdateMsg entityViewUpdateMsg = entityViewMsgConstructor.constructEntityViewDeleteMsg(entityViewId); downlinkMsg = DownlinkMsg.newBuilder() + .setDownlinkMsgId(EdgeUtils.nextPositiveInt()) .addAllEntityViewUpdateMsg(Collections.singletonList(entityViewUpdateMsg)) .build(); break; diff --git a/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/RelationEdgeProcessor.java b/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/RelationEdgeProcessor.java index ddaf55a73b..94b1c97968 100644 --- a/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/RelationEdgeProcessor.java +++ b/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/RelationEdgeProcessor.java @@ -20,6 +20,7 @@ import com.google.common.util.concurrent.Futures; import com.google.common.util.concurrent.ListenableFuture; import lombok.extern.slf4j.Slf4j; import org.springframework.stereotype.Component; +import org.thingsboard.server.common.data.EdgeUtils; import org.thingsboard.server.common.data.EntityType; import org.thingsboard.server.common.data.edge.EdgeEvent; import org.thingsboard.server.common.data.edge.EdgeEventActionType; @@ -119,6 +120,7 @@ public class RelationEdgeProcessor extends BaseEdgeProcessor { EntityRelation entityRelation = mapper.convertValue(edgeEvent.getBody(), EntityRelation.class); RelationUpdateMsg r = relationMsgConstructor.constructRelationUpdatedMsg(msgType, entityRelation); return DownlinkMsg.newBuilder() + .setDownlinkMsgId(EdgeUtils.nextPositiveInt()) .addAllRelationUpdateMsg(Collections.singletonList(r)) .build(); } diff --git a/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/RuleChainEdgeProcessor.java b/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/RuleChainEdgeProcessor.java index 90fc8abe0e..6988c13892 100644 --- a/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/RuleChainEdgeProcessor.java +++ b/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/RuleChainEdgeProcessor.java @@ -17,6 +17,7 @@ package org.thingsboard.server.service.edge.rpc.processor; import lombok.extern.slf4j.Slf4j; import org.springframework.stereotype.Component; +import org.thingsboard.server.common.data.EdgeUtils; import org.thingsboard.server.common.data.edge.Edge; import org.thingsboard.server.common.data.edge.EdgeEvent; import org.thingsboard.server.common.data.edge.EdgeEventActionType; @@ -48,6 +49,7 @@ public class RuleChainEdgeProcessor extends BaseEdgeProcessor { RuleChainUpdateMsg ruleChainUpdateMsg = ruleChainMsgConstructor.constructRuleChainUpdatedMsg(edge.getRootRuleChainId(), msgType, ruleChain); downlinkMsg = DownlinkMsg.newBuilder() + .setDownlinkMsgId(EdgeUtils.nextPositiveInt()) .addAllRuleChainUpdateMsg(Collections.singletonList(ruleChainUpdateMsg)) .build(); } @@ -55,6 +57,7 @@ public class RuleChainEdgeProcessor extends BaseEdgeProcessor { case DELETED: case UNASSIGNED_FROM_EDGE: downlinkMsg = DownlinkMsg.newBuilder() + .setDownlinkMsgId(EdgeUtils.nextPositiveInt()) .addAllRuleChainUpdateMsg(Collections.singletonList(ruleChainMsgConstructor.constructRuleChainDeleteMsg(ruleChainId))) .build(); break; @@ -72,6 +75,7 @@ public class RuleChainEdgeProcessor extends BaseEdgeProcessor { ruleChainMsgConstructor.constructRuleChainMetadataUpdatedMsg(msgType, ruleChainMetaData); if (ruleChainMetadataUpdateMsg != null) { downlinkMsg = DownlinkMsg.newBuilder() + .setDownlinkMsgId(EdgeUtils.nextPositiveInt()) .addAllRuleChainMetadataUpdateMsg(Collections.singletonList(ruleChainMetadataUpdateMsg)) .build(); } diff --git a/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/TelemetryEdgeProcessor.java b/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/TelemetryEdgeProcessor.java index 419cd95179..008b20951c 100644 --- a/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/TelemetryEdgeProcessor.java +++ b/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/TelemetryEdgeProcessor.java @@ -31,6 +31,7 @@ import org.thingsboard.rule.engine.api.msg.DeviceAttributesEventNotificationMsg; import org.thingsboard.server.common.data.DataConstants; import org.thingsboard.server.common.data.Device; import org.thingsboard.server.common.data.DeviceProfile; +import org.thingsboard.server.common.data.EdgeUtils; import org.thingsboard.server.common.data.EntityType; import org.thingsboard.server.common.data.EntityView; import org.thingsboard.server.common.data.asset.Asset; @@ -321,6 +322,7 @@ public class TelemetryEdgeProcessor extends BaseEdgeProcessor { private DownlinkMsg constructEntityDataProtoMsg(EntityId entityId, EdgeEventActionType actionType, JsonElement entityData) { EntityDataProto entityDataProto = entityDataMsgConstructor.constructEntityDataMsg(entityId, actionType, entityData); return DownlinkMsg.newBuilder() + .setDownlinkMsgId(EdgeUtils.nextPositiveInt()) .addAllEntityData(Collections.singletonList(entityDataProto)) .build(); } diff --git a/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/UserEdgeProcessor.java b/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/UserEdgeProcessor.java index 747d2ab4fd..175b601905 100644 --- a/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/UserEdgeProcessor.java +++ b/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/UserEdgeProcessor.java @@ -17,6 +17,7 @@ package org.thingsboard.server.service.edge.rpc.processor; import lombok.extern.slf4j.Slf4j; import org.springframework.stereotype.Component; +import org.thingsboard.server.common.data.EdgeUtils; import org.thingsboard.server.common.data.User; import org.thingsboard.server.common.data.edge.Edge; import org.thingsboard.server.common.data.edge.EdgeEvent; @@ -46,12 +47,14 @@ public class UserEdgeProcessor extends BaseEdgeProcessor { if (user != null) { CustomerId customerId = getCustomerIdIfEdgeAssignedToCustomer(user, edge); downlinkMsg = DownlinkMsg.newBuilder() + .setDownlinkMsgId(EdgeUtils.nextPositiveInt()) .addAllUserUpdateMsg(Collections.singletonList(userMsgConstructor.constructUserUpdatedMsg(msgType, user, customerId))) .build(); } break; case DELETED: downlinkMsg = DownlinkMsg.newBuilder() + .setDownlinkMsgId(EdgeUtils.nextPositiveInt()) .addAllUserUpdateMsg(Collections.singletonList(userMsgConstructor.constructUserDeleteMsg(userId))) .build(); break; @@ -61,6 +64,7 @@ public class UserEdgeProcessor extends BaseEdgeProcessor { UserCredentialsUpdateMsg userCredentialsUpdateMsg = userMsgConstructor.constructUserCredentialsUpdatedMsg(userCredentialsByUserId); downlinkMsg = DownlinkMsg.newBuilder() + .setDownlinkMsgId(EdgeUtils.nextPositiveInt()) .addAllUserCredentialsUpdateMsg(Collections.singletonList(userCredentialsUpdateMsg)) .build(); } diff --git a/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/WidgetBundleEdgeProcessor.java b/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/WidgetBundleEdgeProcessor.java index 8c0d28226d..bc8f92b45b 100644 --- a/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/WidgetBundleEdgeProcessor.java +++ b/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/WidgetBundleEdgeProcessor.java @@ -17,6 +17,7 @@ package org.thingsboard.server.service.edge.rpc.processor; import lombok.extern.slf4j.Slf4j; import org.springframework.stereotype.Component; +import org.thingsboard.server.common.data.EdgeUtils; import org.thingsboard.server.common.data.edge.EdgeEvent; import org.thingsboard.server.common.data.edge.EdgeEventActionType; import org.thingsboard.server.common.data.id.WidgetsBundleId; @@ -44,6 +45,7 @@ public class WidgetBundleEdgeProcessor extends BaseEdgeProcessor { WidgetsBundleUpdateMsg widgetsBundleUpdateMsg = widgetsBundleMsgConstructor.constructWidgetsBundleUpdateMsg(msgType, widgetsBundle); downlinkMsg = DownlinkMsg.newBuilder() + .setDownlinkMsgId(EdgeUtils.nextPositiveInt()) .addAllWidgetsBundleUpdateMsg(Collections.singletonList(widgetsBundleUpdateMsg)) .build(); } @@ -52,6 +54,7 @@ public class WidgetBundleEdgeProcessor extends BaseEdgeProcessor { WidgetsBundleUpdateMsg widgetsBundleUpdateMsg = widgetsBundleMsgConstructor.constructWidgetsBundleDeleteMsg(widgetsBundleId); downlinkMsg = DownlinkMsg.newBuilder() + .setDownlinkMsgId(EdgeUtils.nextPositiveInt()) .addAllWidgetsBundleUpdateMsg(Collections.singletonList(widgetsBundleUpdateMsg)) .build(); break; diff --git a/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/WidgetTypeEdgeProcessor.java b/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/WidgetTypeEdgeProcessor.java index 891b224a6e..dcf2c447bb 100644 --- a/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/WidgetTypeEdgeProcessor.java +++ b/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/WidgetTypeEdgeProcessor.java @@ -17,6 +17,7 @@ package org.thingsboard.server.service.edge.rpc.processor; import lombok.extern.slf4j.Slf4j; import org.springframework.stereotype.Component; +import org.thingsboard.server.common.data.EdgeUtils; import org.thingsboard.server.common.data.edge.EdgeEvent; import org.thingsboard.server.common.data.edge.EdgeEventActionType; import org.thingsboard.server.common.data.id.WidgetTypeId; @@ -44,6 +45,7 @@ public class WidgetTypeEdgeProcessor extends BaseEdgeProcessor { WidgetTypeUpdateMsg widgetTypeUpdateMsg = widgetTypeMsgConstructor.constructWidgetTypeUpdateMsg(msgType, widgetType); downlinkMsg = DownlinkMsg.newBuilder() + .setDownlinkMsgId(EdgeUtils.nextPositiveInt()) .addAllWidgetTypeUpdateMsg(Collections.singletonList(widgetTypeUpdateMsg)) .build(); } @@ -52,6 +54,7 @@ public class WidgetTypeEdgeProcessor extends BaseEdgeProcessor { WidgetTypeUpdateMsg widgetTypeUpdateMsg = widgetTypeMsgConstructor.constructWidgetTypeDeleteMsg(widgetTypeId); downlinkMsg = DownlinkMsg.newBuilder() + .setDownlinkMsgId(EdgeUtils.nextPositiveInt()) .addAllWidgetTypeUpdateMsg(Collections.singletonList(widgetTypeUpdateMsg)) .build(); break; diff --git a/common/data/src/main/java/org/thingsboard/server/common/data/EdgeUtils.java b/common/data/src/main/java/org/thingsboard/server/common/data/EdgeUtils.java index 8b78bcab3a..de8e7e0308 100644 --- a/common/data/src/main/java/org/thingsboard/server/common/data/EdgeUtils.java +++ b/common/data/src/main/java/org/thingsboard/server/common/data/EdgeUtils.java @@ -18,6 +18,8 @@ package org.thingsboard.server.common.data; import lombok.extern.slf4j.Slf4j; import org.thingsboard.server.common.data.edge.EdgeEventType; +import java.util.concurrent.ThreadLocalRandom; + @Slf4j public final class EdgeUtils { @@ -57,4 +59,8 @@ public final class EdgeUtils { return null; } } + + public static int nextPositiveInt() { + return ThreadLocalRandom.current().nextInt(0, Integer.MAX_VALUE); + } } diff --git a/common/edge-api/src/main/proto/edge.proto b/common/edge-api/src/main/proto/edge.proto index a630e64066..ea6b47defb 100644 --- a/common/edge-api/src/main/proto/edge.proto +++ b/common/edge-api/src/main/proto/edge.proto @@ -434,11 +434,13 @@ message UplinkMsg { message UplinkResponseMsg { bool success = 1; string errorMsg = 2; + int32 uplinkMsgId = 3; } message DownlinkResponseMsg { bool success = 1; string errorMsg = 2; + int32 downlinkMsgId = 3; } message DownlinkMsg {