diff --git a/application/src/main/java/org/thingsboard/server/controller/AlarmController.java b/application/src/main/java/org/thingsboard/server/controller/AlarmController.java index e56b4b34ef..01c5ea28ae 100644 --- a/application/src/main/java/org/thingsboard/server/controller/AlarmController.java +++ b/application/src/main/java/org/thingsboard/server/controller/AlarmController.java @@ -91,6 +91,9 @@ public class AlarmController extends BaseController { logEntityAction(savedAlarm.getId(), savedAlarm, getCurrentUser().getCustomerId(), alarm.getId() == null ? ActionType.ADDED : ActionType.UPDATED, null); + + sendNotificationMsgToEdgeService(getTenantId(), savedAlarm.getId(), alarm.getId() == null ? ActionType.ADDED : ActionType.UPDATED); + return savedAlarm; } catch (Exception e) { logEntityAction(emptyId(EntityType.ALARM), alarm, @@ -107,8 +110,11 @@ public class AlarmController extends BaseController { try { AlarmId alarmId = new AlarmId(toUUID(strAlarmId)); checkAlarmId(alarmId, Operation.WRITE); + + sendNotificationMsgToEdgeService(getTenantId(), alarmId, ActionType.DELETED); + return alarmService.deleteAlarm(getTenantId(), alarmId); - } catch (Exception e) { + } catch (Exception e) { throw handleException(e); } } diff --git a/application/src/main/java/org/thingsboard/server/controller/AssetController.java b/application/src/main/java/org/thingsboard/server/controller/AssetController.java index 31f256c03a..20fae1e4c6 100644 --- a/application/src/main/java/org/thingsboard/server/controller/AssetController.java +++ b/application/src/main/java/org/thingsboard/server/controller/AssetController.java @@ -88,7 +88,8 @@ public class AssetController extends BaseController { Asset savedAsset = checkNotNull(assetService.saveAsset(asset)); - sendNotificationMsgToEdgeService(savedAsset.getTenantId(), savedAsset.getId(), EdgeEventType.ASSET, asset.getId() == null ? ActionType.ADDED : ActionType.UPDATED); + sendNotificationMsgToEdgeService(savedAsset.getTenantId(), null, + savedAsset.getId(), EdgeEventType.ASSET, asset.getId() == null ? ActionType.ADDED : ActionType.UPDATED); logEntityAction(savedAsset.getId(), savedAsset, savedAsset.getCustomerId(), @@ -116,7 +117,7 @@ public class AssetController extends BaseController { asset.getCustomerId(), ActionType.DELETED, null, strAssetId); - sendNotificationMsgToEdgeService(getTenantId(), assetId, EdgeEventType.ASSET, ActionType.DELETED); + sendNotificationMsgToEdgeService(getTenantId(), null, assetId, EdgeEventType.ASSET, ActionType.DELETED); } catch (Exception e) { logEntityAction(emptyId(EntityType.ASSET), null, @@ -359,7 +360,7 @@ public class AssetController extends BaseController { savedAsset.getCustomerId(), ActionType.ASSIGNED_TO_EDGE, null, strAssetId, strEdgeId, edge.getName()); - sendNotificationMsgToEdgeService(getTenantId(), savedAsset.getId(), EdgeEventType.ASSET, ActionType.ASSIGNED_TO_EDGE); + sendNotificationMsgToEdgeService(getTenantId(), edgeId, savedAsset.getId(), EdgeEventType.ASSET, ActionType.ASSIGNED_TO_EDGE); return savedAsset; } catch (Exception e) { @@ -392,7 +393,8 @@ public class AssetController extends BaseController { asset.getCustomerId(), ActionType.UNASSIGNED_FROM_EDGE, null, strAssetId, edge.getId().toString(), edge.getName()); - sendNotificationMsgToEdgeService(getTenantId(), savedAsset.getId(), EdgeEventType.ASSET, ActionType.UNASSIGNED_FROM_EDGE); + sendNotificationMsgToEdgeService(getTenantId(), edgeId, savedAsset.getId(), + EdgeEventType.ASSET, ActionType.UNASSIGNED_FROM_EDGE); return savedAsset; } catch (Exception e) { diff --git a/application/src/main/java/org/thingsboard/server/controller/BaseController.java b/application/src/main/java/org/thingsboard/server/controller/BaseController.java index 77f066795c..f71c98d63d 100644 --- a/application/src/main/java/org/thingsboard/server/controller/BaseController.java +++ b/application/src/main/java/org/thingsboard/server/controller/BaseController.java @@ -701,7 +701,7 @@ public abstract class BaseController { protected void sendNotificationMsgToEdgeService(TenantId tenantId, EntityRelation relation, ActionType edgeEventAction) { try { - sendNotificationMsgToEdgeService(tenantId, null, json.writeValueAsString(relation), EdgeEventType.RELATION, edgeEventAction); + sendNotificationMsgToEdgeService(tenantId, null, null, json.writeValueAsString(relation), EdgeEventType.RELATION, edgeEventAction); } catch (Exception e) { log.warn("Failed to push relation to core: {}", relation, e); } @@ -710,15 +710,15 @@ public abstract class BaseController { protected void sendNotificationMsgToEdgeService(TenantId tenantId, EntityId entityId, ActionType edgeEventAction) { EdgeEventType edgeEventType = edgeEventService.getEdgeEventTypeByEntityType(entityId.getEntityType()); if (edgeEventType != null) { - sendNotificationMsgToEdgeService(tenantId, entityId, null, edgeEventType, edgeEventAction); + sendNotificationMsgToEdgeService(tenantId, null, entityId, null, edgeEventType, edgeEventAction); } } - protected void sendNotificationMsgToEdgeService(TenantId tenantId, EntityId entityId, EdgeEventType edgeEventType, ActionType edgeEventAction) { - sendNotificationMsgToEdgeService(tenantId, entityId, null, edgeEventType, edgeEventAction); + protected void sendNotificationMsgToEdgeService(TenantId tenantId, EdgeId edgeId, EntityId entityId, EdgeEventType edgeEventType, ActionType edgeEventAction) { + sendNotificationMsgToEdgeService(tenantId, edgeId, entityId, null, edgeEventType, edgeEventAction); } - private void sendNotificationMsgToEdgeService(TenantId tenantId, EntityId entityId, String entityBody, EdgeEventType edgeEventType, ActionType edgeEventAction) { + private void sendNotificationMsgToEdgeService(TenantId tenantId, EdgeId edgeId, EntityId entityId, String entityBody, EdgeEventType edgeEventType, ActionType edgeEventAction) { TransportProtos.EdgeNotificationMsgProto.Builder builder = TransportProtos.EdgeNotificationMsgProto.newBuilder(); builder.setTenantIdMSB(tenantId.getId().getMostSignificantBits()); builder.setTenantIdLSB(tenantId.getId().getLeastSignificantBits()); @@ -729,6 +729,10 @@ public abstract class BaseController { builder.setEntityIdLSB(entityId.getId().getLeastSignificantBits()); builder.setEntityType(entityId.getEntityType().name()); } + if (edgeId != null) { + builder.setEdgeIdMSB(edgeId.getId().getMostSignificantBits()); + builder.setEdgeIdLSB(edgeId.getId().getLeastSignificantBits()); + } if (entityBody != null) { builder.setEntityBody(entityBody); } diff --git a/application/src/main/java/org/thingsboard/server/controller/DashboardController.java b/application/src/main/java/org/thingsboard/server/controller/DashboardController.java index 90d991a400..22674b6ced 100644 --- a/application/src/main/java/org/thingsboard/server/controller/DashboardController.java +++ b/application/src/main/java/org/thingsboard/server/controller/DashboardController.java @@ -117,7 +117,7 @@ public class DashboardController extends BaseController { null, dashboard.getId() == null ? ActionType.ADDED : ActionType.UPDATED, null); - sendNotificationMsgToEdgeService(savedDashboard.getTenantId(), savedDashboard.getId(), + sendNotificationMsgToEdgeService(savedDashboard.getTenantId(), null, savedDashboard.getId(), EdgeEventType.DASHBOARD, savedDashboard.getId() == null ? ActionType.ADDED : ActionType.UPDATED); return savedDashboard; @@ -143,7 +143,7 @@ public class DashboardController extends BaseController { null, ActionType.DELETED, null, strDashboardId); - sendNotificationMsgToEdgeService(getTenantId(), dashboardId, EdgeEventType.DASHBOARD, ActionType.DELETED); + sendNotificationMsgToEdgeService(getTenantId(), null, dashboardId, EdgeEventType.DASHBOARD, ActionType.DELETED); } catch (Exception e) { logEntityAction(emptyId(EntityType.DASHBOARD), @@ -500,7 +500,8 @@ public class DashboardController extends BaseController { null, ActionType.ASSIGNED_TO_EDGE, null, strDashboardId, strEdgeId, edge.getName()); - sendNotificationMsgToEdgeService(getTenantId(), savedDashboard.getId(), EdgeEventType.DASHBOARD, ActionType.ASSIGNED_TO_EDGE); + sendNotificationMsgToEdgeService(getTenantId(), edgeId, savedDashboard.getId(), + EdgeEventType.DASHBOARD, ActionType.ASSIGNED_TO_EDGE); return savedDashboard; } catch (Exception e) { @@ -532,7 +533,8 @@ public class DashboardController extends BaseController { null, ActionType.UNASSIGNED_FROM_EDGE, null, strDashboardId, edge.getId().toString(), edge.getName()); - sendNotificationMsgToEdgeService(getTenantId(), savedDashboard.getId(), EdgeEventType.DASHBOARD, ActionType.UNASSIGNED_FROM_EDGE); + sendNotificationMsgToEdgeService(getTenantId(), edgeId, savedDashboard.getId(), + EdgeEventType.DASHBOARD, ActionType.UNASSIGNED_FROM_EDGE); return savedDashboard; } catch (Exception e) { diff --git a/application/src/main/java/org/thingsboard/server/controller/DeviceController.java b/application/src/main/java/org/thingsboard/server/controller/DeviceController.java index 0f24507cbc..6c7391127d 100644 --- a/application/src/main/java/org/thingsboard/server/controller/DeviceController.java +++ b/application/src/main/java/org/thingsboard/server/controller/DeviceController.java @@ -108,7 +108,8 @@ public class DeviceController extends BaseController { tbClusterService.pushMsgToCore(new DeviceNameOrTypeUpdateMsg(savedDevice.getTenantId(), savedDevice.getId(), savedDevice.getName(), savedDevice.getType()), null); - sendNotificationMsgToEdgeService(savedDevice.getTenantId(), savedDevice.getId(), EdgeEventType.DEVICE, device.getId() == null ? ActionType.ADDED : ActionType.UPDATED); + sendNotificationMsgToEdgeService(savedDevice.getTenantId(), null, savedDevice.getId(), + EdgeEventType.DEVICE, device.getId() == null ? ActionType.ADDED : ActionType.UPDATED); logEntityAction(savedDevice.getId(), savedDevice, savedDevice.getCustomerId(), @@ -141,7 +142,7 @@ public class DeviceController extends BaseController { device.getCustomerId(), ActionType.DELETED, null, strDeviceId); - sendNotificationMsgToEdgeService(getTenantId(), deviceId, EdgeEventType.DEVICE, ActionType.DELETED); + sendNotificationMsgToEdgeService(getTenantId(), null, deviceId, EdgeEventType.DEVICE, ActionType.DELETED); deviceStateService.onDeviceDeleted(device); } catch (Exception e) { @@ -266,7 +267,7 @@ public class DeviceController extends BaseController { tbClusterService.pushMsgToCore(new DeviceCredentialsUpdateNotificationMsg(getCurrentUser().getTenantId(), deviceCredentials.getDeviceId()), null); - sendNotificationMsgToEdgeService(getTenantId(), device.getId(), EdgeEventType.DEVICE, ActionType.CREDENTIALS_UPDATED); + sendNotificationMsgToEdgeService(getTenantId(), null, device.getId(), EdgeEventType.DEVICE, ActionType.CREDENTIALS_UPDATED); logEntityAction(device.getId(), device, device.getCustomerId(), @@ -517,7 +518,7 @@ public class DeviceController extends BaseController { savedDevice.getCustomerId(), ActionType.ASSIGNED_TO_EDGE, null, strDeviceId, strEdgeId, edge.getName()); - sendNotificationMsgToEdgeService(getTenantId(), savedDevice.getId(), EdgeEventType.DEVICE, ActionType.ASSIGNED_TO_EDGE); + sendNotificationMsgToEdgeService(getTenantId(), edgeId, savedDevice.getId(), EdgeEventType.DEVICE, ActionType.ASSIGNED_TO_EDGE); return savedDevice; } catch (Exception e) { @@ -548,7 +549,7 @@ public class DeviceController extends BaseController { device.getCustomerId(), ActionType.UNASSIGNED_FROM_EDGE, null, strDeviceId, edge.getId().toString(), edge.getName()); - sendNotificationMsgToEdgeService(getTenantId(), savedDevice.getId(), EdgeEventType.DEVICE, ActionType.UNASSIGNED_FROM_EDGE); + sendNotificationMsgToEdgeService(getTenantId(), edgeId, savedDevice.getId(), EdgeEventType.DEVICE, ActionType.UNASSIGNED_FROM_EDGE); return savedDevice; } catch (Exception e) { diff --git a/application/src/main/java/org/thingsboard/server/controller/EntityViewController.java b/application/src/main/java/org/thingsboard/server/controller/EntityViewController.java index 970cd98db3..f2b95963f5 100644 --- a/application/src/main/java/org/thingsboard/server/controller/EntityViewController.java +++ b/application/src/main/java/org/thingsboard/server/controller/EntityViewController.java @@ -118,7 +118,8 @@ public class EntityViewController extends BaseController { logEntityAction(savedEntityView.getId(), savedEntityView, null, entityView.getId() == null ? ActionType.ADDED : ActionType.UPDATED, null); - sendNotificationMsgToEdgeService(getTenantId(), savedEntityView.getId(), EdgeEventType.ENTITY_VIEW, entityView.getId() == null ? ActionType.ADDED : ActionType.UPDATED); + sendNotificationMsgToEdgeService(getTenantId(), null, savedEntityView.getId(), + EdgeEventType.ENTITY_VIEW, entityView.getId() == null ? ActionType.ADDED : ActionType.UPDATED); return savedEntityView; } catch (Exception e) { logEntityAction(emptyId(EntityType.ENTITY_VIEW), entityView, null, @@ -189,7 +190,7 @@ public class EntityViewController extends BaseController { logEntityAction(entityViewId, entityView, entityView.getCustomerId(), ActionType.DELETED, null, strEntityViewId); - sendNotificationMsgToEdgeService(getTenantId(), entityViewId, EdgeEventType.ENTITY_VIEW, ActionType.DELETED); + sendNotificationMsgToEdgeService(getTenantId(), null, entityViewId, EdgeEventType.ENTITY_VIEW, ActionType.DELETED); } catch (Exception e) { logEntityAction(emptyId(EntityType.ENTITY_VIEW), null, @@ -395,7 +396,8 @@ public class EntityViewController extends BaseController { savedEntityView.getCustomerId(), ActionType.ASSIGNED_TO_EDGE, null, strEntityViewId, strEdgeId, edge.getName()); - sendNotificationMsgToEdgeService(getTenantId(), savedEntityView.getId(), EdgeEventType.ENTITY_VIEW, ActionType.ASSIGNED_TO_EDGE); + sendNotificationMsgToEdgeService(getTenantId(), edgeId, savedEntityView.getId(), + EdgeEventType.ENTITY_VIEW, ActionType.ASSIGNED_TO_EDGE); return savedEntityView; } catch (Exception e) { @@ -425,7 +427,8 @@ public class EntityViewController extends BaseController { entityView.getCustomerId(), ActionType.UNASSIGNED_FROM_EDGE, null, strEntityViewId, edge.getId().toString(), edge.getName()); - sendNotificationMsgToEdgeService(getTenantId(), savedEntityView.getId(), EdgeEventType.ENTITY_VIEW, ActionType.UNASSIGNED_FROM_EDGE); + sendNotificationMsgToEdgeService(getTenantId(), edgeId, savedEntityView.getId(), + EdgeEventType.ENTITY_VIEW, ActionType.UNASSIGNED_FROM_EDGE); return savedEntityView; } catch (Exception e) { diff --git a/application/src/main/java/org/thingsboard/server/controller/RuleChainController.java b/application/src/main/java/org/thingsboard/server/controller/RuleChainController.java index ce4a96c2c9..463934b4ab 100644 --- a/application/src/main/java/org/thingsboard/server/controller/RuleChainController.java +++ b/application/src/main/java/org/thingsboard/server/controller/RuleChainController.java @@ -145,7 +145,7 @@ public class RuleChainController extends BaseController { created ? ActionType.ADDED : ActionType.UPDATED, null); if (RuleChainType.EDGE.equals(savedRuleChain.getType())) { - sendNotificationMsgToEdgeService(savedRuleChain.getTenantId(), + sendNotificationMsgToEdgeService(savedRuleChain.getTenantId(), null, savedRuleChain.getId(), EdgeEventType.RULE_CHAIN, savedRuleChain.getId() == null ? ActionType.ADDED : ActionType.UPDATED); } @@ -226,6 +226,7 @@ public class RuleChainController extends BaseController { if (RuleChainType.EDGE.equals(ruleChain.getType())) { sendNotificationMsgToEdgeService(ruleChain.getTenantId(), + null, ruleChain.getId(), EdgeEventType.RULE_CHAIN, ActionType.UPDATED); } @@ -292,9 +293,8 @@ public class RuleChainController extends BaseController { ActionType.DELETED, null, strRuleChainId); if (RuleChainType.EDGE.equals(ruleChain.getType())) { - sendNotificationMsgToEdgeService(ruleChain.getTenantId(), - ruleChain.getId(), EdgeEventType.RULE_CHAIN, - ActionType.DELETED); + sendNotificationMsgToEdgeService(ruleChain.getTenantId(), null, + ruleChain.getId(), EdgeEventType.RULE_CHAIN, ActionType.DELETED); } } catch (Exception e) { @@ -426,7 +426,7 @@ public class RuleChainController extends BaseController { null, ActionType.ASSIGNED_TO_EDGE, null, strRuleChainId, strEdgeId, edge.getName()); - sendNotificationMsgToEdgeService(getTenantId(), savedRuleChain.getId(), + sendNotificationMsgToEdgeService(getTenantId(), edgeId, savedRuleChain.getId(), EdgeEventType.RULE_CHAIN, ActionType.ASSIGNED_TO_EDGE); return savedRuleChain; @@ -459,7 +459,7 @@ public class RuleChainController extends BaseController { null, ActionType.UNASSIGNED_FROM_EDGE, null, strRuleChainId, edge.getId().toString(), edge.getName()); - sendNotificationMsgToEdgeService(getTenantId(), savedRuleChain.getId(), + sendNotificationMsgToEdgeService(getTenantId(), edgeId, savedRuleChain.getId(), EdgeEventType.RULE_CHAIN, ActionType.UNASSIGNED_FROM_EDGE); return savedRuleChain; diff --git a/application/src/main/java/org/thingsboard/server/service/edge/DefaultEdgeNotificationService.java b/application/src/main/java/org/thingsboard/server/service/edge/DefaultEdgeNotificationService.java index 79d0569e2e..954cef733d 100644 --- a/application/src/main/java/org/thingsboard/server/service/edge/DefaultEdgeNotificationService.java +++ b/application/src/main/java/org/thingsboard/server/service/edge/DefaultEdgeNotificationService.java @@ -36,11 +36,12 @@ import org.thingsboard.server.common.data.id.EntityIdFactory; import org.thingsboard.server.common.data.id.IdBased; import org.thingsboard.server.common.data.id.RuleChainId; import org.thingsboard.server.common.data.id.TenantId; +import org.thingsboard.server.common.data.page.TextPageData; +import org.thingsboard.server.common.data.page.TextPageLink; import org.thingsboard.server.common.data.page.TimePageData; import org.thingsboard.server.common.data.page.TimePageLink; import org.thingsboard.server.common.data.relation.EntityRelation; import org.thingsboard.server.common.data.relation.RelationTypeGroup; -import org.thingsboard.server.common.data.rule.RuleChain; import org.thingsboard.server.common.data.rule.RuleChainMetaData; import org.thingsboard.server.common.msg.queue.TbCallback; import org.thingsboard.server.dao.alarm.AlarmService; @@ -141,9 +142,8 @@ public class DefaultEdgeNotificationService implements EdgeNotificationService { @Override public void pushNotificationToEdge(TransportProtos.EdgeNotificationMsgProto edgeNotificationMsg, TbCallback callback) { try { - EdgeEventType edgeEventType = EdgeEventType.valueOf(edgeNotificationMsg.getEdgeEventType()); - ActionType edgeEventAction = ActionType.valueOf(edgeNotificationMsg.getEdgeEventAction()); TenantId tenantId = new TenantId(new UUID(edgeNotificationMsg.getTenantIdMSB(), edgeNotificationMsg.getTenantIdLSB())); + EdgeEventType edgeEventType = EdgeEventType.valueOf(edgeNotificationMsg.getEdgeEventType()); switch (edgeEventType) { // TODO: voba - handle edge updates // case EDGE: @@ -152,34 +152,13 @@ public class DefaultEdgeNotificationService implements EdgeNotificationService { case ENTITY_VIEW: case DASHBOARD: case RULE_CHAIN: - EntityId entityId = EntityIdFactory.getByEdgeEventTypeAndUuid(edgeEventType, new UUID(edgeNotificationMsg.getEntityIdMSB(), edgeNotificationMsg.getEntityIdLSB())); - ListenableFuture> edgeIdsFuture = findRelatedEdgeIdsEntityId(tenantId, entityId); - Futures.transform(edgeIdsFuture, edgeIds -> { - if (edgeIds != null && !edgeIds.isEmpty()) { - for (EdgeId edgeId : edgeIds) { - try { - saveEdgeEvent(tenantId, edgeId, edgeEventType, edgeEventAction, entityId, null); - if (edgeEventType.equals(EdgeEventType.RULE_CHAIN) && - (ActionType.UPDATED.equals(edgeEventAction) || ActionType.ADDED.equals(edgeEventAction))) { - RuleChainMetaData ruleChainMetaData = ruleChainService.loadRuleChainMetaData(tenantId, new RuleChainId(entityId.getId())); - saveEdgeEvent(tenantId, edgeId, EdgeEventType.RULE_CHAIN_METADATA, edgeEventAction, ruleChainMetaData.getRuleChainId(), null); - } - } catch (Exception e) { - log.error("[{}] Failed to push event to edge, edgeId [{}], edgeEventType [{}], edgeEventAction [{}], entityId [{}]", - tenantId, edgeId, edgeEventType, edgeEventAction, entityId, e); - } - } - } - return null; - }, dbCallbackExecutorService); + processEntities(tenantId, edgeNotificationMsg); break; case ALARM: - EntityId alarmId = EntityIdFactory.getByEdgeEventTypeAndUuid(edgeEventType, new UUID(edgeNotificationMsg.getEntityIdMSB(), edgeNotificationMsg.getEntityIdLSB())); - processAlarm(tenantId, edgeEventAction, alarmId); + processAlarm(tenantId, edgeNotificationMsg); break; case RELATION: - EntityRelation entityRelation = mapper.convertValue(edgeNotificationMsg.getEntityBody(), EntityRelation.class); - processRelation(tenantId, edgeEventAction, entityRelation); + processRelation(tenantId, edgeNotificationMsg); break; default: log.debug("Edge event type [{}] is not designed to be pushed to edge", edgeEventType); @@ -192,17 +171,69 @@ public class DefaultEdgeNotificationService implements EdgeNotificationService { } } - private void processAlarm(TenantId tenantId, ActionType edgeActionType, EntityId alarmId) { - ListenableFuture alarmFuture = alarmService.findAlarmByIdAsync(tenantId, new AlarmId(alarmId.getId())); + private void processEntities(TenantId tenantId, TransportProtos.EdgeNotificationMsgProto edgeNotificationMsg) { + ActionType edgeEventActionType = ActionType.valueOf(edgeNotificationMsg.getEdgeEventAction()); + EdgeEventType edgeEventType = EdgeEventType.valueOf(edgeNotificationMsg.getEdgeEventType()); + EntityId entityId = EntityIdFactory.getByEdgeEventTypeAndUuid(edgeEventType, new UUID(edgeNotificationMsg.getEntityIdMSB(), edgeNotificationMsg.getEntityIdLSB())); + switch (edgeEventActionType) { + // TODO: voba - ADDED is not required for CE version ? + // case ADDED: + case UPDATED: + ListenableFuture> edgeIdsFuture = findRelatedEdgeIdsByEntityId(tenantId, entityId); + Futures.transform(edgeIdsFuture, edgeIds -> { + if (edgeIds != null && !edgeIds.isEmpty()) { + for (EdgeId edgeId : edgeIds) { + try { + saveEdgeEvent(tenantId, edgeId, edgeEventType, edgeEventActionType, entityId, null); + if (edgeEventType.equals(EdgeEventType.RULE_CHAIN)) { + RuleChainMetaData ruleChainMetaData = ruleChainService.loadRuleChainMetaData(tenantId, new RuleChainId(entityId.getId())); + saveEdgeEvent(tenantId, edgeId, EdgeEventType.RULE_CHAIN_METADATA, edgeEventActionType, ruleChainMetaData.getRuleChainId(), null); + } + } catch (Exception e) { + log.error("[{}] Failed to push event to edge, edgeId [{}], edgeEventType [{}], edgeEventActionType [{}], entityId [{}]", + tenantId, edgeId, edgeEventType, edgeEventActionType, entityId, e); + } + } + } + return null; + }, dbCallbackExecutorService); + break; + case DELETED: + TextPageData edgesByTenantId = edgeService.findEdgesByTenantId(tenantId, new TextPageLink(Integer.MAX_VALUE)); + if (edgesByTenantId != null && edgesByTenantId.getData() != null && !edgesByTenantId.getData().isEmpty()) { + for (Edge edge : edgesByTenantId.getData()) { + saveEdgeEvent(tenantId, edge.getId(), edgeEventType, edgeEventActionType, entityId, null); + } + } + break; + case ASSIGNED_TO_EDGE: + case UNASSIGNED_FROM_EDGE: + EdgeId edgeId = new EdgeId(new UUID(edgeNotificationMsg.getEdgeIdMSB(), edgeNotificationMsg.getEdgeIdLSB())); + saveEdgeEvent(tenantId, edgeId, edgeEventType, edgeEventActionType, entityId, null); + break; + case RELATIONS_DELETED: + // TODO: voba - add support for relations deleted + break; + } + } + + private void processAlarm(TenantId tenantId, TransportProtos.EdgeNotificationMsgProto edgeNotificationMsg) { + AlarmId alarmId = new AlarmId(new UUID(edgeNotificationMsg.getEntityIdMSB(), edgeNotificationMsg.getEntityIdLSB())); + ListenableFuture alarmFuture = alarmService.findAlarmByIdAsync(tenantId, alarmId); Futures.transform(alarmFuture, alarm -> { if (alarm != null) { EdgeEventType edgeEventType = getEdgeQueueTypeByEntityType(alarm.getOriginator().getEntityType()); if (edgeEventType != null) { - ListenableFuture> relatedEdgeIdsEntityIdFuture = findRelatedEdgeIdsEntityId(tenantId, alarm.getOriginator()); - Futures.transform(relatedEdgeIdsEntityIdFuture, relatedEdgeIdsEntityId -> { - if (relatedEdgeIdsEntityId != null) { - for (EdgeId edgeId : relatedEdgeIdsEntityId) { - saveEdgeEvent(tenantId, edgeId, EdgeEventType.ALARM, edgeActionType, alarmId, null); + ListenableFuture> relatedEdgeIdsByEntityIdFuture = findRelatedEdgeIdsByEntityId(tenantId, alarm.getOriginator()); + Futures.transform(relatedEdgeIdsByEntityIdFuture, relatedEdgeIdsByEntityId -> { + if (relatedEdgeIdsByEntityId != null) { + for (EdgeId edgeId : relatedEdgeIdsByEntityId) { + saveEdgeEvent(tenantId, + edgeId, + EdgeEventType.ALARM, + ActionType.valueOf(edgeNotificationMsg.getEdgeEventAction()), + alarmId, + null); } } return null; @@ -213,10 +244,11 @@ public class DefaultEdgeNotificationService implements EdgeNotificationService { }, dbCallbackExecutorService); } - private void processRelation(TenantId tenantId, ActionType edgeActionType, EntityRelation entityRelation) { + private void processRelation(TenantId tenantId, TransportProtos.EdgeNotificationMsgProto edgeNotificationMsg) { + EntityRelation entityRelation = mapper.convertValue(edgeNotificationMsg.getEntityBody(), EntityRelation.class); List>> futures = new ArrayList<>(); - futures.add(findRelatedEdgeIdsEntityId(tenantId, entityRelation.getTo())); - futures.add(findRelatedEdgeIdsEntityId(tenantId, entityRelation.getFrom())); + futures.add(findRelatedEdgeIdsByEntityId(tenantId, entityRelation.getTo())); + futures.add(findRelatedEdgeIdsByEntityId(tenantId, entityRelation.getFrom())); ListenableFuture>> combinedFuture = Futures.allAsList(futures); Futures.transform(combinedFuture, listOfListsEdgeIds -> { Set uniqueEdgeIds = new HashSet<>(); @@ -229,14 +261,19 @@ public class DefaultEdgeNotificationService implements EdgeNotificationService { } if (!uniqueEdgeIds.isEmpty()) { for (EdgeId edgeId : uniqueEdgeIds) { - saveEdgeEvent(tenantId, edgeId, EdgeEventType.RELATION, edgeActionType, null, mapper.valueToTree(entityRelation)); + saveEdgeEvent(tenantId, + edgeId, + EdgeEventType.RELATION, + ActionType.valueOf(edgeNotificationMsg.getEdgeEventAction()), + null, + mapper.valueToTree(entityRelation)); } } return null; }, dbCallbackExecutorService); } - private ListenableFuture> findRelatedEdgeIdsEntityId(TenantId tenantId, EntityId entityId) { + private ListenableFuture> findRelatedEdgeIdsByEntityId(TenantId tenantId, EntityId entityId) { switch (entityId.getEntityType()) { case DEVICE: case ASSET: 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 a3d9bb7593..341142e34d 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 @@ -360,122 +360,193 @@ public final class EdgeGrpcSession implements Closeable { private void processDeviceCRUD(EdgeEvent edgeEvent, UpdateMsgType msgType) { DeviceId deviceId = new DeviceId(edgeEvent.getEntityId()); - ListenableFuture deviceFuture = ctx.getDeviceService().findDeviceByIdAsync(edgeEvent.getTenantId(), deviceId); - Futures.addCallback(deviceFuture, - new FutureCallback() { - @Override - public void onSuccess(@Nullable Device device) { - if (device != null) { - EntityUpdateMsg entityUpdateMsg = EntityUpdateMsg.newBuilder() - .setDeviceUpdateMsg(ctx.getDeviceUpdateMsgConstructor().constructDeviceUpdatedMsg(msgType, device)) - .build(); - outputStream.onNext(ResponseMsg.newBuilder() - .setEntityUpdateMsg(entityUpdateMsg) - .build()); - } - } + switch (msgType) { + case ENTITY_CREATED_RPC_MESSAGE: + case ENTITY_UPDATED_RPC_MESSAGE: + case DEVICE_CONFLICT_RPC_MESSAGE: + ListenableFuture deviceFuture = ctx.getDeviceService().findDeviceByIdAsync(edgeEvent.getTenantId(), deviceId); + Futures.addCallback(deviceFuture, + new FutureCallback() { + @Override + public void onSuccess(@Nullable Device device) { + if (device != null) { + EntityUpdateMsg entityUpdateMsg = EntityUpdateMsg.newBuilder() + .setDeviceUpdateMsg(ctx.getDeviceUpdateMsgConstructor().constructDeviceUpdatedMsg(msgType, device)) + .build(); + outputStream.onNext(ResponseMsg.newBuilder() + .setEntityUpdateMsg(entityUpdateMsg) + .build()); + } + } + + @Override + public void onFailure(Throwable t) { + log.warn("Can't processDeviceCRUD, edgeEvent [{}]", edgeEvent, t); + } + }, ctx.getDbCallbackExecutor()); + break; + case ENTITY_DELETED_RPC_MESSAGE: + EntityUpdateMsg entityUpdateMsg = EntityUpdateMsg.newBuilder() + .setDeviceUpdateMsg(ctx.getDeviceUpdateMsgConstructor().constructDeviceDeleteMsg(deviceId)) + .build(); + outputStream.onNext(ResponseMsg.newBuilder() + .setEntityUpdateMsg(entityUpdateMsg) + .build()); + } + - @Override - public void onFailure(Throwable t) { - log.warn("Can't processDeviceCRUD, edgeEvent [{}]", edgeEvent, t); - } - }, ctx.getDbCallbackExecutor()); } private void processAssetCRUD(EdgeEvent edgeEvent, UpdateMsgType msgType) { AssetId assetId = new AssetId(edgeEvent.getEntityId()); - ListenableFuture assetFuture = ctx.getAssetService().findAssetByIdAsync(edgeEvent.getTenantId(), assetId); - Futures.addCallback(assetFuture, - new FutureCallback() { - @Override - public void onSuccess(@Nullable Asset asset) { - if (asset != null) { - EntityUpdateMsg entityUpdateMsg = EntityUpdateMsg.newBuilder() - .setAssetUpdateMsg(ctx.getAssetUpdateMsgConstructor().constructAssetUpdatedMsg(msgType, asset)) - .build(); - outputStream.onNext(ResponseMsg.newBuilder() - .setEntityUpdateMsg(entityUpdateMsg) - .build()); - } - } + switch (msgType) { + case ENTITY_CREATED_RPC_MESSAGE: + case ENTITY_UPDATED_RPC_MESSAGE: + case DEVICE_CONFLICT_RPC_MESSAGE: + ListenableFuture assetFuture = ctx.getAssetService().findAssetByIdAsync(edgeEvent.getTenantId(), assetId); + Futures.addCallback(assetFuture, + new FutureCallback() { + @Override + public void onSuccess(@Nullable Asset asset) { + if (asset != null) { + EntityUpdateMsg entityUpdateMsg = EntityUpdateMsg.newBuilder() + .setAssetUpdateMsg(ctx.getAssetUpdateMsgConstructor().constructAssetUpdatedMsg(msgType, asset)) + .build(); + outputStream.onNext(ResponseMsg.newBuilder() + .setEntityUpdateMsg(entityUpdateMsg) + .build()); + } + } - @Override - public void onFailure(Throwable t) { - log.warn("Can't processAssetCRUD, edgeEvent [{}]", edgeEvent, t); - } - }, ctx.getDbCallbackExecutor()); + @Override + public void onFailure(Throwable t) { + log.warn("Can't processAssetCRUD, edgeEvent [{}]", edgeEvent, t); + } + }, ctx.getDbCallbackExecutor()); + break; + case ENTITY_DELETED_RPC_MESSAGE: + EntityUpdateMsg entityUpdateMsg = EntityUpdateMsg.newBuilder() + .setAssetUpdateMsg(ctx.getAssetUpdateMsgConstructor().constructAssetDeleteMsg(assetId)) + .build(); + outputStream.onNext(ResponseMsg.newBuilder() + .setEntityUpdateMsg(entityUpdateMsg) + .build()); + break; + } } private void processEntityViewCRUD(EdgeEvent edgeEvent, UpdateMsgType msgType) { EntityViewId entityViewId = new EntityViewId(edgeEvent.getEntityId()); - ListenableFuture entityViewFuture = ctx.getEntityViewService().findEntityViewByIdAsync(edgeEvent.getTenantId(), entityViewId); - Futures.addCallback(entityViewFuture, - new FutureCallback() { - @Override - public void onSuccess(@Nullable EntityView entityView) { - if (entityView != null) { - EntityUpdateMsg entityUpdateMsg = EntityUpdateMsg.newBuilder() - .setEntityViewUpdateMsg(ctx.getEntityViewUpdateMsgConstructor().constructEntityViewUpdatedMsg(msgType, entityView)) - .build(); - outputStream.onNext(ResponseMsg.newBuilder() - .setEntityUpdateMsg(entityUpdateMsg) - .build()); - } - } + switch (msgType) { + case ENTITY_CREATED_RPC_MESSAGE: + case ENTITY_UPDATED_RPC_MESSAGE: + case DEVICE_CONFLICT_RPC_MESSAGE: + ListenableFuture entityViewFuture = ctx.getEntityViewService().findEntityViewByIdAsync(edgeEvent.getTenantId(), entityViewId); + Futures.addCallback(entityViewFuture, + new FutureCallback() { + @Override + public void onSuccess(@Nullable EntityView entityView) { + if (entityView != null) { + EntityUpdateMsg entityUpdateMsg = EntityUpdateMsg.newBuilder() + .setEntityViewUpdateMsg(ctx.getEntityViewUpdateMsgConstructor().constructEntityViewUpdatedMsg(msgType, entityView)) + .build(); + outputStream.onNext(ResponseMsg.newBuilder() + .setEntityUpdateMsg(entityUpdateMsg) + .build()); + } + } - @Override - public void onFailure(Throwable t) { - log.warn("Can't processEntityViewCRUD, edgeEvent [{}]", edgeEvent, t); - } - }, ctx.getDbCallbackExecutor()); + @Override + public void onFailure(Throwable t) { + log.warn("Can't processEntityViewCRUD, edgeEvent [{}]", edgeEvent, t); + } + }, ctx.getDbCallbackExecutor()); + break; + case ENTITY_DELETED_RPC_MESSAGE: + EntityUpdateMsg entityUpdateMsg = EntityUpdateMsg.newBuilder() + .setEntityViewUpdateMsg(ctx.getEntityViewUpdateMsgConstructor().constructEntityViewDeleteMsg(entityViewId)) + .build(); + outputStream.onNext(ResponseMsg.newBuilder() + .setEntityUpdateMsg(entityUpdateMsg) + .build()); + break; + } } private void processDashboardCRUD(EdgeEvent edgeEvent, UpdateMsgType msgType) { DashboardId dashboardId = new DashboardId(edgeEvent.getEntityId()); - ListenableFuture dashboardFuture = ctx.getDashboardService().findDashboardByIdAsync(edgeEvent.getTenantId(), dashboardId); - Futures.addCallback(dashboardFuture, - new FutureCallback() { - @Override - public void onSuccess(@Nullable Dashboard dashboard) { - if (dashboard != null) { - EntityUpdateMsg entityUpdateMsg = EntityUpdateMsg.newBuilder() - .setDashboardUpdateMsg(ctx.getDashboardUpdateMsgConstructor().constructDashboardUpdatedMsg(msgType, dashboard)) - .build(); - outputStream.onNext(ResponseMsg.newBuilder() - .setEntityUpdateMsg(entityUpdateMsg) - .build()); - } - } + switch (msgType) { + case ENTITY_CREATED_RPC_MESSAGE: + case ENTITY_UPDATED_RPC_MESSAGE: + case DEVICE_CONFLICT_RPC_MESSAGE: + ListenableFuture dashboardFuture = ctx.getDashboardService().findDashboardByIdAsync(edgeEvent.getTenantId(), dashboardId); + Futures.addCallback(dashboardFuture, + new FutureCallback() { + @Override + public void onSuccess(@Nullable Dashboard dashboard) { + if (dashboard != null) { + EntityUpdateMsg entityUpdateMsg = EntityUpdateMsg.newBuilder() + .setDashboardUpdateMsg(ctx.getDashboardUpdateMsgConstructor().constructDashboardUpdatedMsg(msgType, dashboard)) + .build(); + outputStream.onNext(ResponseMsg.newBuilder() + .setEntityUpdateMsg(entityUpdateMsg) + .build()); + } + } - @Override - public void onFailure(Throwable t) { - log.warn("Can't processDashboardCRUD, edgeEvent [{}]", edgeEvent, t); - } - }, ctx.getDbCallbackExecutor()); + @Override + public void onFailure(Throwable t) { + log.warn("Can't processDashboardCRUD, edgeEvent [{}]", edgeEvent, t); + } + }, ctx.getDbCallbackExecutor()); + break; + case ENTITY_DELETED_RPC_MESSAGE: + EntityUpdateMsg entityUpdateMsg = EntityUpdateMsg.newBuilder() + .setDashboardUpdateMsg(ctx.getDashboardUpdateMsgConstructor().constructDashboardDeleteMsg(dashboardId)) + .build(); + outputStream.onNext(ResponseMsg.newBuilder() + .setEntityUpdateMsg(entityUpdateMsg) + .build()); + break; + } } private void processRuleChainCRUD(EdgeEvent edgeEvent, UpdateMsgType msgType) { RuleChainId ruleChainId = new RuleChainId(edgeEvent.getEntityId()); - ListenableFuture ruleChainFuture = ctx.getRuleChainService().findRuleChainByIdAsync(edgeEvent.getTenantId(), ruleChainId); - Futures.addCallback(ruleChainFuture, - new FutureCallback() { - @Override - public void onSuccess(@Nullable RuleChain ruleChain) { - if (ruleChain != null) { - EntityUpdateMsg entityUpdateMsg = EntityUpdateMsg.newBuilder() - .setRuleChainUpdateMsg(ctx.getRuleChainUpdateMsgConstructor().constructRuleChainUpdatedMsg(edge.getRootRuleChainId(), msgType, ruleChain)) - .build(); - outputStream.onNext(ResponseMsg.newBuilder() - .setEntityUpdateMsg(entityUpdateMsg) - .build()); - } - } + switch (msgType) { + case ENTITY_CREATED_RPC_MESSAGE: + case ENTITY_UPDATED_RPC_MESSAGE: + case DEVICE_CONFLICT_RPC_MESSAGE: + ListenableFuture ruleChainFuture = ctx.getRuleChainService().findRuleChainByIdAsync(edgeEvent.getTenantId(), ruleChainId); + Futures.addCallback(ruleChainFuture, + new FutureCallback() { + @Override + public void onSuccess(@Nullable RuleChain ruleChain) { + if (ruleChain != null) { + EntityUpdateMsg entityUpdateMsg = EntityUpdateMsg.newBuilder() + .setRuleChainUpdateMsg(ctx.getRuleChainUpdateMsgConstructor().constructRuleChainUpdatedMsg(edge.getRootRuleChainId(), msgType, ruleChain)) + .build(); + outputStream.onNext(ResponseMsg.newBuilder() + .setEntityUpdateMsg(entityUpdateMsg) + .build()); + } + } - @Override - public void onFailure(Throwable t) { - log.warn("Can't processRuleChainCRUD, edgeEvent [{}]", edgeEvent, t); - } - }, ctx.getDbCallbackExecutor()); + @Override + public void onFailure(Throwable t) { + log.warn("Can't processRuleChainCRUD, edgeEvent [{}]", edgeEvent, t); + } + }, ctx.getDbCallbackExecutor()); + break; + case ENTITY_DELETED_RPC_MESSAGE: + EntityUpdateMsg entityUpdateMsg = EntityUpdateMsg.newBuilder() + .setRuleChainUpdateMsg(ctx.getRuleChainUpdateMsgConstructor().constructRuleChainDeleteMsg(ruleChainId)) + .build(); + outputStream.onNext(ResponseMsg.newBuilder() + .setEntityUpdateMsg(entityUpdateMsg) + .build()); + break; + } } private void processRuleChainMetadataCRUD(EdgeEvent edgeEvent, UpdateMsgType msgType) { @@ -509,26 +580,40 @@ public final class EdgeGrpcSession implements Closeable { private void processUserCRUD(EdgeEvent edgeEvent, UpdateMsgType msgType) { UserId userId = new UserId(edgeEvent.getEntityId()); - ListenableFuture userFuture = ctx.getUserService().findUserByIdAsync(edgeEvent.getTenantId(), userId); - Futures.addCallback(userFuture, - new FutureCallback() { - @Override - public void onSuccess(@Nullable User user) { - if (user != null) { - EntityUpdateMsg entityUpdateMsg = EntityUpdateMsg.newBuilder() - .setUserUpdateMsg(ctx.getUserUpdateMsgConstructor().constructUserUpdatedMsg(msgType, user)) - .build(); - outputStream.onNext(ResponseMsg.newBuilder() - .setEntityUpdateMsg(entityUpdateMsg) - .build()); - } - } + switch (msgType) { + case ENTITY_CREATED_RPC_MESSAGE: + case ENTITY_UPDATED_RPC_MESSAGE: + case DEVICE_CONFLICT_RPC_MESSAGE: + ListenableFuture userFuture = ctx.getUserService().findUserByIdAsync(edgeEvent.getTenantId(), userId); + Futures.addCallback(userFuture, + new FutureCallback() { + @Override + public void onSuccess(@Nullable User user) { + if (user != null) { + EntityUpdateMsg entityUpdateMsg = EntityUpdateMsg.newBuilder() + .setUserUpdateMsg(ctx.getUserUpdateMsgConstructor().constructUserUpdatedMsg(msgType, user)) + .build(); + outputStream.onNext(ResponseMsg.newBuilder() + .setEntityUpdateMsg(entityUpdateMsg) + .build()); + } + } - @Override - public void onFailure(Throwable t) { - log.warn("Can't processUserCRUD, edgeEvent [{}]", edgeEvent, t); - } - }, ctx.getDbCallbackExecutor()); + @Override + public void onFailure(Throwable t) { + log.warn("Can't processUserCRUD, edgeEvent [{}]", edgeEvent, t); + } + }, ctx.getDbCallbackExecutor()); + break; + case ENTITY_DELETED_RPC_MESSAGE: + EntityUpdateMsg entityUpdateMsg = EntityUpdateMsg.newBuilder() + .setUserUpdateMsg(ctx.getUserUpdateMsgConstructor().constructUserDeleteMsg(userId)) + .build(); + outputStream.onNext(ResponseMsg.newBuilder() + .setEntityUpdateMsg(entityUpdateMsg) + .build()); + break; + } } private void processRelationCRUD(EdgeEvent edgeEvent, UpdateMsgType msgType) { @@ -567,9 +652,9 @@ public final class EdgeGrpcSession implements Closeable { private UpdateMsgType getResponseMsgType(ActionType actionType) { switch (actionType) { - case ADDED: - return UpdateMsgType.ENTITY_UPDATED_RPC_MESSAGE; case UPDATED: + return UpdateMsgType.ENTITY_UPDATED_RPC_MESSAGE; + case ADDED: case ASSIGNED_TO_EDGE: return ENTITY_CREATED_RPC_MESSAGE; case DELETED: diff --git a/application/src/main/java/org/thingsboard/server/service/edge/rpc/constructor/AssetUpdateMsgConstructor.java b/application/src/main/java/org/thingsboard/server/service/edge/rpc/constructor/AssetUpdateMsgConstructor.java index 30af762756..c404fe5c41 100644 --- a/application/src/main/java/org/thingsboard/server/service/edge/rpc/constructor/AssetUpdateMsgConstructor.java +++ b/application/src/main/java/org/thingsboard/server/service/edge/rpc/constructor/AssetUpdateMsgConstructor.java @@ -18,6 +18,7 @@ package org.thingsboard.server.service.edge.rpc.constructor; import lombok.extern.slf4j.Slf4j; import org.springframework.stereotype.Component; import org.thingsboard.server.common.data.asset.Asset; +import org.thingsboard.server.common.data.id.AssetId; import org.thingsboard.server.gen.edge.AssetUpdateMsg; import org.thingsboard.server.gen.edge.UpdateMsgType; @@ -38,4 +39,10 @@ public class AssetUpdateMsgConstructor { return builder.build(); } + public AssetUpdateMsg constructAssetDeleteMsg(AssetId assetId) { + return AssetUpdateMsg.newBuilder() + .setMsgType(UpdateMsgType.ENTITY_DELETED_RPC_MESSAGE) + .setIdMSB(assetId.getId().getMostSignificantBits()) + .setIdLSB(assetId.getId().getLeastSignificantBits()).build(); + } } diff --git a/application/src/main/java/org/thingsboard/server/service/edge/rpc/constructor/DashboardUpdateMsgConstructor.java b/application/src/main/java/org/thingsboard/server/service/edge/rpc/constructor/DashboardUpdateMsgConstructor.java index 3826cc25f0..6bd24f123b 100644 --- a/application/src/main/java/org/thingsboard/server/service/edge/rpc/constructor/DashboardUpdateMsgConstructor.java +++ b/application/src/main/java/org/thingsboard/server/service/edge/rpc/constructor/DashboardUpdateMsgConstructor.java @@ -19,6 +19,7 @@ import lombok.extern.slf4j.Slf4j; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.stereotype.Component; import org.thingsboard.server.common.data.Dashboard; +import org.thingsboard.server.common.data.id.DashboardId; import org.thingsboard.server.dao.dashboard.DashboardService; import org.thingsboard.server.dao.util.mapping.JacksonUtil; import org.thingsboard.server.gen.edge.DashboardUpdateMsg; @@ -42,4 +43,11 @@ public class DashboardUpdateMsgConstructor { return builder.build(); } + public DashboardUpdateMsg constructDashboardDeleteMsg(DashboardId dashboardId) { + return DashboardUpdateMsg.newBuilder() + .setMsgType(UpdateMsgType.ENTITY_DELETED_RPC_MESSAGE) + .setIdMSB(dashboardId.getId().getMostSignificantBits()) + .setIdLSB(dashboardId.getId().getLeastSignificantBits()).build(); + } + } diff --git a/application/src/main/java/org/thingsboard/server/service/edge/rpc/constructor/DeviceUpdateMsgConstructor.java b/application/src/main/java/org/thingsboard/server/service/edge/rpc/constructor/DeviceUpdateMsgConstructor.java index 791abc8cf7..8d7fbc4c25 100644 --- a/application/src/main/java/org/thingsboard/server/service/edge/rpc/constructor/DeviceUpdateMsgConstructor.java +++ b/application/src/main/java/org/thingsboard/server/service/edge/rpc/constructor/DeviceUpdateMsgConstructor.java @@ -19,6 +19,7 @@ import lombok.extern.slf4j.Slf4j; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.stereotype.Component; import org.thingsboard.server.common.data.Device; +import org.thingsboard.server.common.data.id.DeviceId; import org.thingsboard.server.common.data.security.DeviceCredentials; import org.thingsboard.server.dao.device.DeviceCredentialsService; import org.thingsboard.server.gen.edge.DeviceUpdateMsg; @@ -58,4 +59,11 @@ public class DeviceUpdateMsgConstructor { } return builder.build(); } + + public DeviceUpdateMsg constructDeviceDeleteMsg(DeviceId deviceId) { + return DeviceUpdateMsg.newBuilder() + .setMsgType(UpdateMsgType.ENTITY_DELETED_RPC_MESSAGE) + .setIdMSB(deviceId.getId().getMostSignificantBits()) + .setIdLSB(deviceId.getId().getLeastSignificantBits()).build(); + } } diff --git a/application/src/main/java/org/thingsboard/server/service/edge/rpc/constructor/EntityViewUpdateMsgConstructor.java b/application/src/main/java/org/thingsboard/server/service/edge/rpc/constructor/EntityViewUpdateMsgConstructor.java index d52bc02321..75301c7ed9 100644 --- a/application/src/main/java/org/thingsboard/server/service/edge/rpc/constructor/EntityViewUpdateMsgConstructor.java +++ b/application/src/main/java/org/thingsboard/server/service/edge/rpc/constructor/EntityViewUpdateMsgConstructor.java @@ -18,6 +18,7 @@ package org.thingsboard.server.service.edge.rpc.constructor; import lombok.extern.slf4j.Slf4j; import org.springframework.stereotype.Component; import org.thingsboard.server.common.data.EntityView; +import org.thingsboard.server.common.data.id.EntityViewId; import org.thingsboard.server.gen.edge.EdgeEntityType; import org.thingsboard.server.gen.edge.EntityViewUpdateMsg; import org.thingsboard.server.gen.edge.UpdateMsgType; @@ -50,4 +51,10 @@ public class EntityViewUpdateMsgConstructor { return builder.build(); } + public EntityViewUpdateMsg constructEntityViewDeleteMsg(EntityViewId entityViewId) { + return EntityViewUpdateMsg.newBuilder() + .setMsgType(UpdateMsgType.ENTITY_DELETED_RPC_MESSAGE) + .setIdMSB(entityViewId.getId().getMostSignificantBits()) + .setIdLSB(entityViewId.getId().getLeastSignificantBits()).build(); + } } diff --git a/application/src/main/java/org/thingsboard/server/service/edge/rpc/constructor/RuleChainUpdateMsgConstructor.java b/application/src/main/java/org/thingsboard/server/service/edge/rpc/constructor/RuleChainUpdateMsgConstructor.java index 204ea18bcd..52e487d973 100644 --- a/application/src/main/java/org/thingsboard/server/service/edge/rpc/constructor/RuleChainUpdateMsgConstructor.java +++ b/application/src/main/java/org/thingsboard/server/service/edge/rpc/constructor/RuleChainUpdateMsgConstructor.java @@ -19,7 +19,6 @@ import com.fasterxml.jackson.core.JsonProcessingException; import com.fasterxml.jackson.databind.ObjectMapper; import lombok.extern.slf4j.Slf4j; import org.springframework.stereotype.Component; -import org.thingsboard.server.common.data.edge.Edge; import org.thingsboard.server.common.data.id.RuleChainId; import org.thingsboard.server.common.data.rule.NodeConnectionInfo; import org.thingsboard.server.common.data.rule.RuleChain; @@ -126,7 +125,6 @@ public class RuleChainUpdateMsgConstructor { .build(); } - private RuleNodeProto constructNode(RuleNode node) throws JsonProcessingException { return RuleNodeProto.newBuilder() .setIdMSB(node.getId().getId().getMostSignificantBits()) @@ -139,4 +137,11 @@ public class RuleChainUpdateMsgConstructor { .build(); } + public RuleChainUpdateMsg constructRuleChainDeleteMsg(RuleChainId ruleChainId) { + return RuleChainUpdateMsg.newBuilder() + .setMsgType(UpdateMsgType.ENTITY_DELETED_RPC_MESSAGE) + .setIdMSB(ruleChainId.getId().getMostSignificantBits()) + .setIdLSB(ruleChainId.getId().getLeastSignificantBits()).build(); + } + } diff --git a/application/src/main/java/org/thingsboard/server/service/edge/rpc/constructor/UserUpdateMsgConstructor.java b/application/src/main/java/org/thingsboard/server/service/edge/rpc/constructor/UserUpdateMsgConstructor.java index 06744530dc..b994d8fdc4 100644 --- a/application/src/main/java/org/thingsboard/server/service/edge/rpc/constructor/UserUpdateMsgConstructor.java +++ b/application/src/main/java/org/thingsboard/server/service/edge/rpc/constructor/UserUpdateMsgConstructor.java @@ -19,6 +19,7 @@ import lombok.extern.slf4j.Slf4j; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.stereotype.Component; import org.thingsboard.server.common.data.User; +import org.thingsboard.server.common.data.id.UserId; import org.thingsboard.server.common.data.security.UserCredentials; import org.thingsboard.server.dao.user.UserService; import org.thingsboard.server.dao.util.mapping.JacksonUtil; @@ -61,4 +62,11 @@ public class UserUpdateMsgConstructor { } return builder.build(); } + + public UserUpdateMsg constructUserDeleteMsg(UserId userId) { + return UserUpdateMsg.newBuilder() + .setMsgType(UpdateMsgType.ENTITY_DELETED_RPC_MESSAGE) + .setIdMSB(userId.getId().getMostSignificantBits()) + .setIdLSB(userId.getId().getLeastSignificantBits()).build(); + } } diff --git a/common/dao-api/src/main/java/org/thingsboard/server/dao/edge/EdgeService.java b/common/dao-api/src/main/java/org/thingsboard/server/dao/edge/EdgeService.java index 60c6b24918..733fabc3f9 100644 --- a/common/dao-api/src/main/java/org/thingsboard/server/dao/edge/EdgeService.java +++ b/common/dao-api/src/main/java/org/thingsboard/server/dao/edge/EdgeService.java @@ -70,7 +70,6 @@ public interface EdgeService { ListenableFuture> findEdgeTypesByTenantId(TenantId tenantId); - void assignDefaultRuleChainsToEdge(TenantId tenantId, EdgeId edgeId); ListenableFuture> findEdgesByTenantIdAndRuleChainId(TenantId tenantId, RuleChainId ruleChainId, TimePageLink pageLink); diff --git a/common/queue/src/main/proto/queue.proto b/common/queue/src/main/proto/queue.proto index 54a8148fbe..921a096df0 100644 --- a/common/queue/src/main/proto/queue.proto +++ b/common/queue/src/main/proto/queue.proto @@ -356,14 +356,16 @@ message FromDeviceRPCResponseProto { message EdgeNotificationMsgProto { int64 tenantIdMSB = 1; int64 tenantIdLSB = 2; - string edgeEventType = 3; - string edgeEventAction = 4; - int64 entityIdMSB = 5; - int64 entityIdLSB = 6; - string entityType = 7; - string entityBody = 8; - PostTelemetryMsg postTelemetryMsg = 9; - PostAttributeMsg postAttributesMsg = 10; + int64 edgeIdMSB = 3; + int64 edgeIdLSB = 4; + string edgeEventType = 5; + string edgeEventAction = 6; + int64 entityIdMSB = 7; + int64 entityIdLSB = 8; + string entityType = 9; + string entityBody = 10; + PostTelemetryMsg postTelemetryMsg = 11; + PostAttributeMsg postAttributesMsg = 12; } /**