From 9c8ceaa37e72ffe1f21532752a777c46f6ca6459 Mon Sep 17 00:00:00 2001 From: Volodymyr Babak Date: Fri, 28 Aug 2020 14:14:07 +0300 Subject: [PATCH] Properly handle entities assign/unassign to/from customer --- .../server/controller/AssetController.java | 13 +- .../server/controller/BaseController.java | 11 ++ .../server/controller/CustomerController.java | 5 + .../controller/DashboardController.java | 14 +- .../server/controller/DeviceController.java | 11 +- .../server/controller/EdgeController.java | 8 +- .../controller/EntityViewController.java | 13 +- .../controller/RuleChainController.java | 5 +- .../edge/DefaultEdgeNotificationService.java | 141 ++++++++++++++---- .../service/edge/rpc/EdgeGrpcSession.java | 90 +++++++---- .../AssetUpdateMsgConstructor.java | 9 +- .../DashboardUpdateMsgConstructor.java | 8 +- .../DeviceUpdateMsgConstructor.java | 8 +- .../EntityViewUpdateMsgConstructor.java | 8 +- .../constructor/UserUpdateMsgConstructor.java | 12 +- .../edge/rpc/init/DefaultSyncEdgeService.java | 6 +- .../thingsboard/edge/rpc/EdgeGrpcClient.java | 10 +- common/edge-api/src/main/proto/edge.proto | 34 +++-- .../server/dao/asset/BaseAssetService.java | 8 +- .../server/dao/device/DeviceServiceImpl.java | 10 +- .../server/dao/edge/EdgeServiceImpl.java | 14 +- .../dao/entity/AbstractEntityService.java | 26 ++++ .../dao/entityview/EntityViewServiceImpl.java | 12 ++ 23 files changed, 366 insertions(+), 110 deletions(-) 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 0877937afa..2f5b848b29 100644 --- a/application/src/main/java/org/thingsboard/server/controller/AssetController.java +++ b/application/src/main/java/org/thingsboard/server/controller/AssetController.java @@ -84,7 +84,7 @@ public class AssetController extends BaseController { try { asset.setTenantId(getCurrentUser().getTenantId()); - checkEntity(asset.getId(), asset, Resource.ASSET); + checkEntity(asset.getId(), asset, Resource.ASSET); Asset savedAsset = checkNotNull(assetService.saveAsset(asset)); @@ -92,8 +92,9 @@ public class AssetController extends BaseController { savedAsset.getCustomerId(), asset.getId() == null ? ActionType.ADDED : ActionType.UPDATED, null); - sendNotificationMsgToEdgeService(getTenantId(), savedAsset.getId(), - asset.getId() == null ? ActionType.ADDED : ActionType.UPDATED); + if (asset.getId() != null) { + sendNotificationMsgToEdgeService(savedAsset.getTenantId(), savedAsset.getId(), ActionType.UPDATED); + } return savedAsset; } catch (Exception e) { @@ -147,6 +148,9 @@ public class AssetController extends BaseController { savedAsset.getCustomerId(), ActionType.ASSIGNED_TO_CUSTOMER, null, strAssetId, strCustomerId, customer.getName()); + sendNotificationMsgToEdgeService(savedAsset.getTenantId(), savedAsset.getId(), + customerId, ActionType.ASSIGNED_TO_CUSTOMER); + return savedAsset; } catch (Exception e) { @@ -178,6 +182,9 @@ public class AssetController extends BaseController { asset.getCustomerId(), ActionType.UNASSIGNED_FROM_CUSTOMER, null, strAssetId, customer.getId().toString(), customer.getName()); + sendNotificationMsgToEdgeService(savedAsset.getTenantId(), savedAsset.getId(), + customer.getId(), ActionType.UNASSIGNED_FROM_CUSTOMER); + 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 390b2187fa..79fa90cbdf 100644 --- a/application/src/main/java/org/thingsboard/server/controller/BaseController.java +++ b/application/src/main/java/org/thingsboard/server/controller/BaseController.java @@ -726,6 +726,17 @@ public abstract class BaseController { } } + protected void sendNotificationMsgToEdgeService(TenantId tenantId, EntityId entityId, CustomerId customerId, ActionType edgeEventAction) { + EdgeEventType edgeEventType = EdgeUtils.getEdgeEventTypeByEntityType(entityId.getEntityType()); + try { + if (edgeEventType != null) { + sendNotificationMsgToEdgeService(tenantId, null, entityId, json.writeValueAsString(customerId), edgeEventType, edgeEventAction); + } + } catch (Exception e) { + log.warn("Failed to push assign/unassign to/from customer to core: {}", customerId, e); + } + } + protected void sendNotificationMsgToEdgeService(TenantId tenantId, EntityRelation relation, ActionType edgeEventAction) { try { if (!relation.getFrom().getEntityType().equals(EntityType.EDGE) && diff --git a/application/src/main/java/org/thingsboard/server/controller/CustomerController.java b/application/src/main/java/org/thingsboard/server/controller/CustomerController.java index 75dee42ff8..6e7ca50418 100644 --- a/application/src/main/java/org/thingsboard/server/controller/CustomerController.java +++ b/application/src/main/java/org/thingsboard/server/controller/CustomerController.java @@ -108,6 +108,10 @@ public class CustomerController extends BaseController { savedCustomer.getId(), customer.getId() == null ? ActionType.ADDED : ActionType.UPDATED, null); + if (customer.getId() != null) { + sendNotificationMsgToEdgeService(savedCustomer.getTenantId(), savedCustomer.getId(),ActionType.UPDATED); + } + return savedCustomer; } catch (Exception e) { @@ -132,6 +136,7 @@ public class CustomerController extends BaseController { customer.getId(), ActionType.DELETED, null, strCustomerId); + sendNotificationMsgToEdgeService(getTenantId(), customerId, ActionType.DELETED); } catch (Exception e) { logEntityAction(emptyId(EntityType.CUSTOMER), 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 0ae455d8f4..31c7a7d0b7 100644 --- a/application/src/main/java/org/thingsboard/server/controller/DashboardController.java +++ b/application/src/main/java/org/thingsboard/server/controller/DashboardController.java @@ -117,8 +117,9 @@ public class DashboardController extends BaseController { null, dashboard.getId() == null ? ActionType.ADDED : ActionType.UPDATED, null); - sendNotificationMsgToEdgeService(savedDashboard.getTenantId(), savedDashboard.getId(), - dashboard.getId() == null ? ActionType.ADDED : ActionType.UPDATED); + if (dashboard.getId() != null) { + sendNotificationMsgToEdgeService(savedDashboard.getTenantId(), savedDashboard.getId(), ActionType.UPDATED); + } return savedDashboard; } catch (Exception e) { @@ -175,6 +176,7 @@ public class DashboardController extends BaseController { customerId, ActionType.ASSIGNED_TO_CUSTOMER, null, strDashboardId, strCustomerId, customer.getName()); + sendNotificationMsgToEdgeService(savedDashboard.getTenantId(), savedDashboard.getId(), customerId, ActionType.ASSIGNED_TO_CUSTOMER); return savedDashboard; } catch (Exception e) { @@ -206,6 +208,8 @@ public class DashboardController extends BaseController { customerId, ActionType.UNASSIGNED_FROM_CUSTOMER, null, strDashboardId, customer.getId().toString(), customer.getName()); + sendNotificationMsgToEdgeService(savedDashboard.getTenantId(), savedDashboard.getId(), customerId, ActionType.UNASSIGNED_FROM_CUSTOMER); + return savedDashboard; } catch (Exception e) { @@ -261,6 +265,7 @@ public class DashboardController extends BaseController { logEntityAction(dashboardId, savedDashboard, customerId, ActionType.ASSIGNED_TO_CUSTOMER, null, strDashboardId, customerId.toString(), customerInfo.getTitle()); + sendNotificationMsgToEdgeService(savedDashboard.getTenantId(), savedDashboard.getId(), customerId, ActionType.ASSIGNED_TO_CUSTOMER); } for (CustomerId customerId : removedCustomerIds) { ShortCustomerInfo customerInfo = dashboard.getAssignedCustomerInfo(customerId); @@ -268,7 +273,7 @@ public class DashboardController extends BaseController { logEntityAction(dashboardId, dashboard, customerId, ActionType.UNASSIGNED_FROM_CUSTOMER, null, strDashboardId, customerId.toString(), customerInfo.getTitle()); - + sendNotificationMsgToEdgeService(savedDashboard.getTenantId(), savedDashboard.getId(), customerId, ActionType.UNASSIGNED_FROM_CUSTOMER); } return savedDashboard; } @@ -312,6 +317,7 @@ public class DashboardController extends BaseController { logEntityAction(dashboardId, savedDashboard, customerId, ActionType.ASSIGNED_TO_CUSTOMER, null, strDashboardId, customerId.toString(), customerInfo.getTitle()); + sendNotificationMsgToEdgeService(savedDashboard.getTenantId(), savedDashboard.getId(), customerId, ActionType.ASSIGNED_TO_CUSTOMER); } return savedDashboard; } @@ -355,7 +361,7 @@ public class DashboardController extends BaseController { logEntityAction(dashboardId, dashboard, customerId, ActionType.UNASSIGNED_FROM_CUSTOMER, null, strDashboardId, customerId.toString(), customerInfo.getTitle()); - + sendNotificationMsgToEdgeService(savedDashboard.getTenantId(), savedDashboard.getId(), customerId, ActionType.UNASSIGNED_FROM_CUSTOMER); } return savedDashboard; } 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 b65d6387c2..da9e352dc9 100644 --- a/application/src/main/java/org/thingsboard/server/controller/DeviceController.java +++ b/application/src/main/java/org/thingsboard/server/controller/DeviceController.java @@ -108,8 +108,9 @@ public class DeviceController extends BaseController { tbClusterService.pushMsgToCore(new DeviceNameOrTypeUpdateMsg(savedDevice.getTenantId(), savedDevice.getId(), savedDevice.getName(), savedDevice.getType()), null); - sendNotificationMsgToEdgeService(savedDevice.getTenantId(), savedDevice.getId(), - device.getId() == null ? ActionType.ADDED : ActionType.UPDATED); + if (device.getId() != null) { + sendNotificationMsgToEdgeService(savedDevice.getTenantId(), savedDevice.getId(),ActionType.UPDATED); + } logEntityAction(savedDevice.getId(), savedDevice, savedDevice.getCustomerId(), @@ -174,6 +175,9 @@ public class DeviceController extends BaseController { savedDevice.getCustomerId(), ActionType.ASSIGNED_TO_CUSTOMER, null, strDeviceId, strCustomerId, customer.getName()); + sendNotificationMsgToEdgeService(savedDevice.getTenantId(), savedDevice.getId(), + customerId, ActionType.ASSIGNED_TO_CUSTOMER); + return savedDevice; } catch (Exception e) { logEntityAction(emptyId(EntityType.DEVICE), null, @@ -202,6 +206,9 @@ public class DeviceController extends BaseController { device.getCustomerId(), ActionType.UNASSIGNED_FROM_CUSTOMER, null, strDeviceId, customer.getId().toString(), customer.getName()); + sendNotificationMsgToEdgeService(savedDevice.getTenantId(), savedDevice.getId(), + customer.getId(), ActionType.UNASSIGNED_FROM_CUSTOMER); + return savedDevice; } catch (Exception e) { logEntityAction(emptyId(EntityType.DEVICE), null, diff --git a/application/src/main/java/org/thingsboard/server/controller/EdgeController.java b/application/src/main/java/org/thingsboard/server/controller/EdgeController.java index 19b82b3b85..44be68b4c2 100644 --- a/application/src/main/java/org/thingsboard/server/controller/EdgeController.java +++ b/application/src/main/java/org/thingsboard/server/controller/EdgeController.java @@ -174,6 +174,9 @@ public class EdgeController extends BaseController { Edge savedEdge = checkNotNull(edgeService.assignEdgeToCustomer(getCurrentUser().getTenantId(), edgeId, customerId)); + tbClusterService.onEntityStateChange(getTenantId(), edgeId, + ComponentLifecycleEvent.UPDATED); + logEntityAction(edgeId, savedEdge, savedEdge.getCustomerId(), ActionType.ASSIGNED_TO_CUSTOMER, null, strEdgeId, strCustomerId, customer.getName()); @@ -205,12 +208,15 @@ public class EdgeController extends BaseController { Edge savedEdge = checkNotNull(edgeService.unassignEdgeFromCustomer(getCurrentUser().getTenantId(), edgeId)); + tbClusterService.onEntityStateChange(getTenantId(), edgeId, + ComponentLifecycleEvent.UPDATED); + logEntityAction(edgeId, edge, edge.getCustomerId(), ActionType.UNASSIGNED_FROM_CUSTOMER, null, strEdgeId, customer.getId().toString(), customer.getName()); sendNotificationMsgToEdgeService(savedEdge.getTenantId(), savedEdge.getId(), - edge.getCustomerId(), ActionType.UNASSIGNED_FROM_CUSTOMER); + customer.getId(), ActionType.UNASSIGNED_FROM_CUSTOMER); return savedEdge; } 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 821dbcbd70..b4762accab 100644 --- a/application/src/main/java/org/thingsboard/server/controller/EntityViewController.java +++ b/application/src/main/java/org/thingsboard/server/controller/EntityViewController.java @@ -118,8 +118,10 @@ public class EntityViewController extends BaseController { logEntityAction(savedEntityView.getId(), savedEntityView, null, entityView.getId() == null ? ActionType.ADDED : ActionType.UPDATED, null); - sendNotificationMsgToEdgeService(getTenantId(), savedEntityView.getId(), - entityView.getId() == null ? ActionType.ADDED : ActionType.UPDATED); + if (entityView.getId() != null) { + sendNotificationMsgToEdgeService(savedEntityView.getTenantId(), savedEntityView.getId(), ActionType.UPDATED); + } + return savedEntityView; } catch (Exception e) { logEntityAction(emptyId(EntityType.ENTITY_VIEW), entityView, null, @@ -231,6 +233,10 @@ public class EntityViewController extends BaseController { logEntityAction(entityViewId, savedEntityView, savedEntityView.getCustomerId(), ActionType.ASSIGNED_TO_CUSTOMER, null, strEntityViewId, strCustomerId, customer.getName()); + + sendNotificationMsgToEdgeService(savedEntityView.getTenantId(), savedEntityView.getId(), + customerId, ActionType.ASSIGNED_TO_CUSTOMER); + return savedEntityView; } catch (Exception e) { logEntityAction(emptyId(EntityType.ENTITY_VIEW), null, @@ -257,6 +263,9 @@ public class EntityViewController extends BaseController { entityView.getCustomerId(), ActionType.UNASSIGNED_FROM_CUSTOMER, null, strEntityViewId, customer.getId().toString(), customer.getName()); + sendNotificationMsgToEdgeService(savedEntityView.getTenantId(), savedEntityView.getId(), + customer.getId(), ActionType.UNASSIGNED_FROM_CUSTOMER); + return savedEntityView; } catch (Exception e) { logEntityAction(emptyId(EntityType.ENTITY_VIEW), null, 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 4d2e78bc67..3b483782d9 100644 --- a/application/src/main/java/org/thingsboard/server/controller/RuleChainController.java +++ b/application/src/main/java/org/thingsboard/server/controller/RuleChainController.java @@ -145,8 +145,9 @@ public class RuleChainController extends BaseController { created ? ActionType.ADDED : ActionType.UPDATED, null); if (RuleChainType.EDGE.equals(savedRuleChain.getType())) { - sendNotificationMsgToEdgeService(savedRuleChain.getTenantId(), savedRuleChain.getId(), - created ? ActionType.ADDED : ActionType.UPDATED); + if (!created) { + sendNotificationMsgToEdgeService(savedRuleChain.getTenantId(), savedRuleChain.getId(), ActionType.UPDATED); + } } 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 279ec96064..0a85dcacf5 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 @@ -54,7 +54,6 @@ import org.thingsboard.server.common.msg.queue.TbCallback; import org.thingsboard.server.dao.alarm.AlarmService; import org.thingsboard.server.dao.edge.EdgeEventService; import org.thingsboard.server.dao.edge.EdgeService; -import org.thingsboard.server.dao.model.ModelConstants; import org.thingsboard.server.dao.relation.RelationService; import org.thingsboard.server.dao.rule.RuleChainService; import org.thingsboard.server.dao.user.UserService; @@ -166,9 +165,14 @@ public class DefaultEdgeNotificationService implements EdgeNotificationService { case ENTITY_VIEW: case DASHBOARD: case RULE_CHAIN: + processEntity(tenantId, edgeNotificationMsg); + break; + case CUSTOMER: + processCustomer(tenantId, edgeNotificationMsg); + break; case WIDGETS_BUNDLE: case WIDGET_TYPE: - processEntity(tenantId, edgeNotificationMsg); + processWidgetBundleOrWidgetType(tenantId, edgeNotificationMsg); break; case ALARM: processAlarm(tenantId, edgeNotificationMsg); @@ -188,26 +192,24 @@ public class DefaultEdgeNotificationService implements EdgeNotificationService { } private void processEdge(TenantId tenantId, TransportProtos.EdgeNotificationMsgProto edgeNotificationMsg) { - // TODO: voba - handle edge updates try { ActionType edgeEventActionType = ActionType.valueOf(edgeNotificationMsg.getEdgeEventAction()); EdgeId edgeId = new EdgeId(new UUID(edgeNotificationMsg.getEdgeIdMSB(), edgeNotificationMsg.getEdgeIdLSB())); + ListenableFuture edgeFuture; switch (edgeEventActionType) { case ASSIGNED_TO_CUSTOMER: - case UNASSIGNED_FROM_CUSTOMER: CustomerId customerId = mapper.readValue(edgeNotificationMsg.getEntityBody(), CustomerId.class); - ListenableFuture edgeFuture = edgeService.findEdgeByIdAsync(tenantId, edgeId); + edgeFuture = edgeService.findEdgeByIdAsync(tenantId, edgeId); Futures.addCallback(edgeFuture, new FutureCallback() { @Override public void onSuccess(@Nullable Edge edge) { - if (edge != null && customerId != null && !EntityId.NULL_UUID.equals(customerId.getId())) { - ActionType actionType = ActionType.ASSIGNED_TO_CUSTOMER.equals(edgeEventActionType) ? ActionType.ADDED : ActionType.DELETED; - saveEdgeEvent(edge.getTenantId(), edge.getId(), EdgeEventType.CUSTOMER, actionType, customerId, null); + if (edge != null && !customerId.isNullUid()) { + saveEdgeEvent(edge.getTenantId(), edge.getId(), EdgeEventType.CUSTOMER, ActionType.ADDED, customerId, null); TextPageData pageData = userService.findCustomerUsers(tenantId, customerId, new TextPageLink(Integer.MAX_VALUE)); if (pageData != null && pageData.getData() != null && !pageData.getData().isEmpty()) { - log.trace("[{}] [{}] user(s) are going to be {} to edge.", edge.getId(), pageData.getData().size(), actionType.name()); + log.trace("[{}] [{}] user(s) are going to be added to edge.", edge.getId(), pageData.getData().size()); for (User user : pageData.getData()) { - saveEdgeEvent(edge.getTenantId(), edge.getId(), EdgeEventType.USER, actionType, user.getId(), null); + saveEdgeEvent(edge.getTenantId(), edge.getId(), EdgeEventType.USER, ActionType.ADDED, user.getId(), null); } } } @@ -219,44 +221,129 @@ public class DefaultEdgeNotificationService implements EdgeNotificationService { } }, dbCallbackExecutorService); break; + case UNASSIGNED_FROM_CUSTOMER: + CustomerId customerIdToDelete = mapper.readValue(edgeNotificationMsg.getEntityBody(), CustomerId.class); + edgeFuture = edgeService.findEdgeByIdAsync(tenantId, edgeId); + Futures.addCallback(edgeFuture, new FutureCallback() { + @Override + public void onSuccess(@Nullable Edge edge) { + if (edge != null && !customerIdToDelete.isNullUid()) { + saveEdgeEvent(edge.getTenantId(), edge.getId(), EdgeEventType.CUSTOMER, ActionType.DELETED, customerIdToDelete, null); + } + } + + @Override + public void onFailure(Throwable t) { + log.error("Can't find edge by id [{}]", edgeNotificationMsg, t); + } + }, dbCallbackExecutorService); + break; } } catch (Exception e) { log.error("Exception during processing edge event", e); } } - private void processEntity(TenantId tenantId, TransportProtos.EdgeNotificationMsgProto edgeNotificationMsg) { + private void processWidgetBundleOrWidgetType(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: - case CREDENTIALS_UPDATED: - if (edgeEventType.equals(EdgeEventType.WIDGETS_BUNDLE) || edgeEventType.equals(EdgeEventType.WIDGET_TYPE)) { - TextPageData edgesByTenantId = edgeService.findEdgesByTenantId(tenantId, new TextPageLink(Integer.MAX_VALUE)); - if (edgesByTenantId != null && edgesByTenantId.getData() != null && !edgesByTenantId.getData().isEmpty()) { - for (Edge edge : edgesByTenantId.getData()) { + 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; + } + } + + private void processCustomer(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())); + TextPageData edgesByTenantId = edgeService.findEdgesByTenantId(tenantId, new TextPageLink(Integer.MAX_VALUE)); + if (edgesByTenantId != null && edgesByTenantId.getData() != null && !edgesByTenantId.getData().isEmpty()) { + for (Edge edge : edgesByTenantId.getData()) { + switch (edgeEventActionType) { + case UPDATED: + if (!edge.getCustomerId().isNullUid() && edge.getCustomerId().equals(entityId)) { saveEdgeEvent(tenantId, edge.getId(), edgeEventType, edgeEventActionType, entityId, null); } + break; + case DELETED: + saveEdgeEvent(tenantId, edge.getId(), edgeEventType, edgeEventActionType, entityId, null); + break; + } + } + } + } + + private void processEntity(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())); + ListenableFuture> edgeIdsFuture; + switch (edgeEventActionType) { + case ADDED: // used only for USER entity + case UPDATED: + case CREDENTIALS_UPDATED: + edgeIdsFuture = findRelatedEdgeIdsByEntityId(tenantId, entityId); + Futures.addCallback(edgeIdsFuture, new FutureCallback>() { + @Override + public void onSuccess(@Nullable List edgeIds) { + if (edgeIds != null && !edgeIds.isEmpty()) { + for (EdgeId edgeId : edgeIds) { + saveEdgeEvent(tenantId, edgeId, edgeEventType, edgeEventActionType, entityId, null); + } + } } - } else { - ListenableFuture> edgeIdsFuture = findRelatedEdgeIdsByEntityId(tenantId, entityId); - Futures.transform(edgeIdsFuture, edgeIds -> { + @Override + public void onFailure(Throwable throwable) { + log.error("Failed to find related edge ids [{}]", edgeNotificationMsg, throwable); + } + }, dbCallbackExecutorService); + break; + case ASSIGNED_TO_CUSTOMER: + case UNASSIGNED_FROM_CUSTOMER: + edgeIdsFuture = findRelatedEdgeIdsByEntityId(tenantId, entityId); + Futures.addCallback(edgeIdsFuture, new FutureCallback>() { + @Override + public void onSuccess(@Nullable List edgeIds) { if (edgeIds != null && !edgeIds.isEmpty()) { for (EdgeId edgeId : edgeIds) { try { - saveEdgeEvent(tenantId, edgeId, edgeEventType, edgeEventActionType, entityId, null); + CustomerId customerId = mapper.readValue(edgeNotificationMsg.getEntityBody(), CustomerId.class); + ListenableFuture future = edgeService.findEdgeByIdAsync(tenantId, edgeId); + Futures.addCallback(future, new FutureCallback() { + @Override + public void onSuccess(@Nullable Edge edge) { + if (edge != null && edge.getCustomerId() != null && + !edge.getCustomerId().isNullUid() && edge.getCustomerId().equals(customerId)) { + saveEdgeEvent(tenantId, edgeId, edgeEventType, edgeEventActionType, entityId, null); + } + } + @Override + public void onFailure(Throwable throwable) { + log.error("Failed to find edge by id [{}]", edgeNotificationMsg, throwable); + } + }, dbCallbackExecutorService); } catch (Exception e) { - log.error("[{}] Failed to push event to edge, edgeId [{}], edgeEventType [{}], edgeEventActionType [{}], entityId [{}]", - tenantId, edgeId, edgeEventType, edgeEventActionType, entityId, e); + log.error("Can't parse customer id from entity body [{}]", edgeNotificationMsg, e); } } } - return null; - }, dbCallbackExecutorService); - } + } + + @Override + public void onFailure(Throwable throwable) { + log.error("Failed to find related edge ids [{}]", edgeNotificationMsg, throwable); + } + }, dbCallbackExecutorService); break; case DELETED: TextPageData edgesByTenantId = edgeService.findEdgesByTenantId(tenantId, new TextPageLink(Integer.MAX_VALUE)); @@ -394,7 +481,7 @@ public class DefaultEdgeNotificationService implements EdgeNotificationService { case USER: User userById = userService.findUserById(tenantId, new UserId(entityId.getId())); TextPageData edges; - if (userById.getCustomerId() == null || userById.getCustomerId().getId().equals(ModelConstants.NULL_UUID)) { + if (userById.getCustomerId() == null || userById.getCustomerId().isNullUid()) { edges = edgeService.findEdgesByTenantId(tenantId, new TextPageLink(Integer.MAX_VALUE)); } else { edges = edgeService.findEdgesByTenantIdAndCustomerId(tenantId, new CustomerId(entityId.getId()), new TextPageLink(Integer.MAX_VALUE)); 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 121772a01b..e036349ab1 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 @@ -17,14 +17,13 @@ package org.thingsboard.server.service.edge.rpc; import com.datastax.driver.core.utils.UUIDs; import com.fasterxml.jackson.core.JsonProcessingException; -import com.fasterxml.jackson.databind.JsonNode; import com.fasterxml.jackson.databind.ObjectMapper; -import com.fasterxml.jackson.databind.node.NullNode; import com.fasterxml.jackson.databind.node.ObjectNode; import com.google.common.util.concurrent.FutureCallback; import com.google.common.util.concurrent.Futures; import com.google.common.util.concurrent.ListenableFuture; import com.google.common.util.concurrent.MoreExecutors; +import com.google.common.util.concurrent.SettableFuture; import com.google.gson.Gson; import com.google.gson.JsonElement; import com.google.gson.JsonObject; @@ -40,6 +39,7 @@ import org.thingsboard.server.common.data.DataConstants; import org.thingsboard.server.common.data.Device; import org.thingsboard.server.common.data.EntityType; import org.thingsboard.server.common.data.EntityView; +import org.thingsboard.server.common.data.HasCustomerId; import org.thingsboard.server.common.data.User; import org.thingsboard.server.common.data.alarm.Alarm; import org.thingsboard.server.common.data.alarm.AlarmSeverity; @@ -73,7 +73,6 @@ 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.data.security.Authority; import org.thingsboard.server.common.data.security.DeviceCredentials; import org.thingsboard.server.common.data.security.DeviceCredentialsType; import org.thingsboard.server.common.data.security.UserCredentials; @@ -138,8 +137,6 @@ import java.util.concurrent.locks.ReentrantLock; import java.util.function.BiConsumer; import java.util.function.Consumer; -import static org.thingsboard.server.gen.edge.UpdateMsgType.ENTITY_CREATED_RPC_MESSAGE; - @Slf4j @Data public final class EdgeGrpcSession implements Closeable { @@ -185,13 +182,14 @@ public final class EdgeGrpcSession implements Closeable { public void onNext(RequestMsg requestMsg) { if (!connected && requestMsg.getMsgType().equals(RequestMsgType.CONNECT_RPC_MESSAGE)) { ConnectResponseMsg responseMsg = processConnect(requestMsg.getConnectRequestMsg()); - sendResponseMsg(ResponseMsg.newBuilder() + outputStream.onNext(ResponseMsg.newBuilder() .setConnectResponseMsg(responseMsg) .build()); if (ConnectResponseCode.ACCEPTED != responseMsg.getResponseCode()) { outputStream.onError(new RuntimeException(responseMsg.getErrorMsg())); } if (ConnectResponseCode.ACCEPTED == responseMsg.getResponseCode()) { + connected = true; ctx.getSyncEdgeService().sync(edge); } } @@ -212,6 +210,7 @@ public final class EdgeGrpcSession implements Closeable { @Override public void onCompleted() { + connected = false; sessionCloseListener.accept(edge.getId()); outputStream.onCompleted(); } @@ -257,6 +256,10 @@ public final class EdgeGrpcSession implements Closeable { try { responseMsgLock.lock(); outputStream.onNext(responseMsg); + } catch (Exception e) { + log.error("Failed to send response message [{}]", responseMsg, e); + connected = false; + sessionCloseListener.accept(edge.getId()); } finally { responseMsgLock.unlock(); } @@ -337,14 +340,16 @@ public final class EdgeGrpcSession implements Closeable { switch (edgeEventAction) { case UPDATED: case ADDED: - case ASSIGNED_TO_EDGE: case DELETED: + case ASSIGNED_TO_EDGE: case UNASSIGNED_FROM_EDGE: case ALARM_ACK: case ALARM_CLEAR: case CREDENTIALS_UPDATED: case RELATION_ADD_OR_UPDATE: case RELATION_DELETED: + case ASSIGNED_TO_CUSTOMER: + case UNASSIGNED_FROM_CUSTOMER: downlinkMsg = processEntityMessage(edgeEvent, edgeEventAction); break; case ATTRIBUTES_UPDATED: @@ -464,10 +469,13 @@ public final class EdgeGrpcSession implements Closeable { case ADDED: case UPDATED: case ASSIGNED_TO_EDGE: + case ASSIGNED_TO_CUSTOMER: + case UNASSIGNED_FROM_CUSTOMER: Device device = ctx.getDeviceService().findDeviceById(edgeEvent.getTenantId(), deviceId); if (device != null) { + CustomerId customerId = getCustomerIdIfEdgeAssignedToCustomer(device); DeviceUpdateMsg deviceUpdateMsg = - ctx.getDeviceUpdateMsgConstructor().constructDeviceUpdatedMsg(msgType, device); + ctx.getDeviceUpdateMsgConstructor().constructDeviceUpdatedMsg(msgType, device, customerId); downlinkMsg = DownlinkMsg.newBuilder() .addAllDeviceUpdateMsg(Collections.singletonList(deviceUpdateMsg)) .build(); @@ -502,10 +510,13 @@ public final class EdgeGrpcSession implements Closeable { case ADDED: case UPDATED: case ASSIGNED_TO_EDGE: + case ASSIGNED_TO_CUSTOMER: + case UNASSIGNED_FROM_CUSTOMER: Asset asset = ctx.getAssetService().findAssetById(edgeEvent.getTenantId(), assetId); if (asset != null) { + CustomerId customerId = getCustomerIdIfEdgeAssignedToCustomer(asset); AssetUpdateMsg assetUpdateMsg = - ctx.getAssetUpdateMsgConstructor().constructAssetUpdatedMsg(msgType, asset); + ctx.getAssetUpdateMsgConstructor().constructAssetUpdatedMsg(msgType, asset, customerId); downlinkMsg = DownlinkMsg.newBuilder() .addAllAssetUpdateMsg(Collections.singletonList(assetUpdateMsg)) .build(); @@ -530,10 +541,13 @@ public final class EdgeGrpcSession implements Closeable { case ADDED: case UPDATED: case ASSIGNED_TO_EDGE: + case ASSIGNED_TO_CUSTOMER: + case UNASSIGNED_FROM_CUSTOMER: EntityView entityView = ctx.getEntityViewService().findEntityViewById(edgeEvent.getTenantId(), entityViewId); if (entityView != null) { + CustomerId customerId = getCustomerIdIfEdgeAssignedToCustomer(entityView); EntityViewUpdateMsg entityViewUpdateMsg = - ctx.getEntityViewUpdateMsgConstructor().constructEntityViewUpdatedMsg(msgType, entityView); + ctx.getEntityViewUpdateMsgConstructor().constructEntityViewUpdatedMsg(msgType, entityView, customerId); downlinkMsg = DownlinkMsg.newBuilder() .addAllEntityViewUpdateMsg(Collections.singletonList(entityViewUpdateMsg)) .build(); @@ -558,10 +572,16 @@ public final class EdgeGrpcSession implements Closeable { case ADDED: case UPDATED: case ASSIGNED_TO_EDGE: + case ASSIGNED_TO_CUSTOMER: + case UNASSIGNED_FROM_CUSTOMER: Dashboard dashboard = ctx.getDashboardService().findDashboardById(edgeEvent.getTenantId(), dashboardId); if (dashboard != null) { + CustomerId customerId = null; + if (!edge.getCustomerId().isNullUid() && dashboard.isAssignedToCustomer(edge.getCustomerId())) { + customerId = edge.getCustomerId(); + } DashboardUpdateMsg dashboardUpdateMsg = - ctx.getDashboardUpdateMsgConstructor().constructDashboardUpdatedMsg(msgType, dashboard); + ctx.getDashboardUpdateMsgConstructor().constructDashboardUpdatedMsg(msgType, dashboard, customerId); downlinkMsg = DownlinkMsg.newBuilder() .addAllDashboardUpdateMsg(Collections.singletonList(dashboardUpdateMsg)) .build(); @@ -654,19 +674,15 @@ public final class EdgeGrpcSession implements Closeable { switch (edgeActionType) { case ADDED: case UPDATED: - case ASSIGNED_TO_EDGE: User user = ctx.getUserService().findUserById(edgeEvent.getTenantId(), userId); if (user != null) { - boolean fullAccess = Authority.TENANT_ADMIN.equals(user.getAuthority()); - setFullAccess(user, fullAccess); - + CustomerId customerId = getCustomerIdIfEdgeAssignedToCustomer(user); downlinkMsg = DownlinkMsg.newBuilder() - .addAllUserUpdateMsg(Collections.singletonList(ctx.getUserUpdateMsgConstructor().constructUserUpdatedMsg(msgType, user))) + .addAllUserUpdateMsg(Collections.singletonList(ctx.getUserUpdateMsgConstructor().constructUserUpdatedMsg(msgType, user, customerId))) .build(); } break; case DELETED: - case UNASSIGNED_FROM_EDGE: downlinkMsg = DownlinkMsg.newBuilder() .addAllUserUpdateMsg(Collections.singletonList(ctx.getUserUpdateMsgConstructor().constructUserDeleteMsg(userId))) .build(); @@ -684,13 +700,12 @@ public final class EdgeGrpcSession implements Closeable { return downlinkMsg; } - private void setFullAccess(User user, boolean isFullAccess) { - JsonNode additionalInfo = user.getAdditionalInfo(); - if (additionalInfo == null || additionalInfo instanceof NullNode) { - additionalInfo = mapper.createObjectNode(); + private CustomerId getCustomerIdIfEdgeAssignedToCustomer(HasCustomerId hasCustomerIdEntity) { + if (!edge.getCustomerId().isNullUid() && edge.getCustomerId().equals(hasCustomerIdEntity.getCustomerId())) { + return edge.getCustomerId(); + } else { + return null; } - ((ObjectNode) additionalInfo).put("isFullAccess", isFullAccess); - user.setAdditionalInfo(additionalInfo); } private DownlinkMsg processRelation(EdgeEvent edgeEvent, UpdateMsgType msgType) { @@ -781,11 +796,13 @@ public final class EdgeGrpcSession implements Closeable { switch (actionType) { case UPDATED: case CREDENTIALS_UPDATED: + case ASSIGNED_TO_CUSTOMER: + case UNASSIGNED_FROM_CUSTOMER: return UpdateMsgType.ENTITY_UPDATED_RPC_MESSAGE; case ADDED: case ASSIGNED_TO_EDGE: case RELATION_ADD_OR_UPDATE: - return ENTITY_CREATED_RPC_MESSAGE; + return UpdateMsgType.ENTITY_CREATED_RPC_MESSAGE; case DELETED: case UNASSIGNED_FROM_EDGE: case RELATION_DELETED: @@ -923,6 +940,8 @@ public final class EdgeGrpcSession implements Closeable { return new TenantId(new UUID(entityData.getEntityIdMSB(), entityData.getEntityIdLSB())); case CUSTOMER: return new CustomerId(new UUID(entityData.getEntityIdMSB(), entityData.getEntityIdLSB())); + case USER: + return new UserId(new UUID(entityData.getEntityIdMSB(), entityData.getEntityIdLSB())); default: log.warn("Unsupported entity type [{}] during construct of entity id. EntityDataProto [{}]", entityData.getEntityType(), entityData); return null; @@ -930,14 +949,24 @@ public final class EdgeGrpcSession implements Closeable { } private ListenableFuture processPostTelemetry(EntityId entityId, TransportProtos.PostTelemetryMsg msg, TbMsgMetaData metaData) { + SettableFuture futureToSet = SettableFuture.create(); for (TransportProtos.TsKvListProto tsKv : msg.getTsKvListList()) { JsonObject json = JsonUtils.getJsonObject(tsKv.getKvList()); metaData.putValue("ts", tsKv.getTs() + ""); TbMsg tbMsg = TbMsg.newMsg(SessionMsgType.POST_TELEMETRY_REQUEST.name(), entityId, metaData, gson.toJson(json)); - // TODO: voba - verify that null callback is OK - ctx.getTbClusterService().pushMsgToRuleEngine(edge.getTenantId(), tbMsg.getOriginator(), tbMsg, null); + ctx.getTbClusterService().pushMsgToRuleEngine(edge.getTenantId(), tbMsg.getOriginator(), tbMsg, new TbQueueCallback() { + @Override + public void onSuccess(TbQueueMsgMetadata metadata) { + futureToSet.set(null); + } + + @Override + public void onFailure(Throwable t) { + futureToSet.setException(t); + } + }); } - return Futures.immediateFuture(null); + return futureToSet; } private ListenableFuture processPostAttributes(EntityId entityId, TransportProtos.PostAttributeMsg msg, TbMsgMetaData metaData) { @@ -957,7 +986,8 @@ public final class EdgeGrpcSession implements Closeable { if (device != null) { // device with this name already exists on the cloud - update ID on the edge if (!device.getId().equals(edgeDeviceId)) { - DeviceUpdateMsg d = ctx.getDeviceUpdateMsgConstructor().constructDeviceUpdatedMsg(UpdateMsgType.DEVICE_CONFLICT_RPC_MESSAGE, device); + CustomerId customerId = getCustomerIdIfEdgeAssignedToCustomer(device); + DeviceUpdateMsg d = ctx.getDeviceUpdateMsgConstructor().constructDeviceUpdatedMsg(UpdateMsgType.DEVICE_CONFLICT_RPC_MESSAGE, device, customerId); DownlinkMsg downlinkMsg = DownlinkMsg.newBuilder() .addAllDeviceUpdateMsg(Collections.singletonList(d)) .build(); @@ -970,7 +1000,8 @@ public final class EdgeGrpcSession implements Closeable { if (deviceById != null) { // this ID already used by other device - create new device and update ID on the edge device = createDevice(deviceUpdateMsg); - DeviceUpdateMsg d = ctx.getDeviceUpdateMsgConstructor().constructDeviceUpdatedMsg(UpdateMsgType.DEVICE_CONFLICT_RPC_MESSAGE, device); + CustomerId customerId = getCustomerIdIfEdgeAssignedToCustomer(device); + DeviceUpdateMsg d = ctx.getDeviceUpdateMsgConstructor().constructDeviceUpdatedMsg(UpdateMsgType.DEVICE_CONFLICT_RPC_MESSAGE, device, customerId); DownlinkMsg downlinkMsg = DownlinkMsg.newBuilder() .addAllDeviceUpdateMsg(Collections.singletonList(d)) .build(); @@ -1250,7 +1281,6 @@ public final class EdgeGrpcSession implements Closeable { edge = optional.get(); try { if (edge.getSecret().equals(request.getEdgeSecret())) { - connected = true; sessionOpenListener.accept(edge.getId(), this); return ConnectResponseMsg.newBuilder() .setResponseCode(ConnectResponseCode.ACCEPTED) 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 c404fe5c41..ddcb531583 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 @@ -17,8 +17,11 @@ package org.thingsboard.server.service.edge.rpc.constructor; import lombok.extern.slf4j.Slf4j; import org.springframework.stereotype.Component; +import org.thingsboard.server.common.data.Customer; import org.thingsboard.server.common.data.asset.Asset; import org.thingsboard.server.common.data.id.AssetId; +import org.thingsboard.server.common.data.id.CustomerId; +import org.thingsboard.server.common.data.id.EntityId; import org.thingsboard.server.gen.edge.AssetUpdateMsg; import org.thingsboard.server.gen.edge.UpdateMsgType; @@ -26,7 +29,7 @@ import org.thingsboard.server.gen.edge.UpdateMsgType; @Slf4j public class AssetUpdateMsgConstructor { - public AssetUpdateMsg constructAssetUpdatedMsg(UpdateMsgType msgType, Asset asset) { + public AssetUpdateMsg constructAssetUpdatedMsg(UpdateMsgType msgType, Asset asset, CustomerId customerId) { AssetUpdateMsg.Builder builder = AssetUpdateMsg.newBuilder() .setMsgType(msgType) .setIdMSB(asset.getId().getId().getMostSignificantBits()) @@ -36,6 +39,10 @@ public class AssetUpdateMsgConstructor { if (asset.getLabel() != null) { builder.setLabel(asset.getLabel()); } + if (customerId != null) { + builder.setCustomerIdMSB(customerId.getId().getMostSignificantBits()); + builder.setCustomerIdLSB(customerId.getId().getLeastSignificantBits()); + } return builder.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 6c52464326..69b9d8dbd2 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 @@ -18,7 +18,9 @@ package org.thingsboard.server.service.edge.rpc.constructor; import lombok.extern.slf4j.Slf4j; import org.springframework.stereotype.Component; import org.thingsboard.server.common.data.Dashboard; +import org.thingsboard.server.common.data.id.CustomerId; import org.thingsboard.server.common.data.id.DashboardId; +import org.thingsboard.server.common.data.id.EntityId; import org.thingsboard.server.dao.util.mapping.JacksonUtil; import org.thingsboard.server.gen.edge.DashboardUpdateMsg; import org.thingsboard.server.gen.edge.UpdateMsgType; @@ -27,13 +29,17 @@ import org.thingsboard.server.gen.edge.UpdateMsgType; @Slf4j public class DashboardUpdateMsgConstructor { - public DashboardUpdateMsg constructDashboardUpdatedMsg(UpdateMsgType msgType, Dashboard dashboard) { + public DashboardUpdateMsg constructDashboardUpdatedMsg(UpdateMsgType msgType, Dashboard dashboard, CustomerId customerId) { DashboardUpdateMsg.Builder builder = DashboardUpdateMsg.newBuilder() .setMsgType(msgType) .setIdMSB(dashboard.getId().getId().getMostSignificantBits()) .setIdLSB(dashboard.getId().getId().getLeastSignificantBits()) .setTitle(dashboard.getTitle()) .setConfiguration(JacksonUtil.toString(dashboard.getConfiguration())); + if (customerId != null) { + builder.setCustomerIdMSB(customerId.getId().getMostSignificantBits()); + builder.setCustomerIdLSB(customerId.getId().getLeastSignificantBits()); + } return builder.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 dcef0a57c6..885511b3db 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 @@ -18,7 +18,9 @@ package org.thingsboard.server.service.edge.rpc.constructor; import lombok.extern.slf4j.Slf4j; import org.springframework.stereotype.Component; import org.thingsboard.server.common.data.Device; +import org.thingsboard.server.common.data.id.CustomerId; import org.thingsboard.server.common.data.id.DeviceId; +import org.thingsboard.server.common.data.id.EntityId; import org.thingsboard.server.common.data.security.DeviceCredentials; import org.thingsboard.server.gen.edge.DeviceCredentialsUpdateMsg; import org.thingsboard.server.gen.edge.DeviceUpdateMsg; @@ -28,7 +30,7 @@ import org.thingsboard.server.gen.edge.UpdateMsgType; @Slf4j public class DeviceUpdateMsgConstructor { - public DeviceUpdateMsg constructDeviceUpdatedMsg(UpdateMsgType msgType, Device device) { + public DeviceUpdateMsg constructDeviceUpdatedMsg(UpdateMsgType msgType, Device device, CustomerId customerId) { DeviceUpdateMsg.Builder builder = DeviceUpdateMsg.newBuilder() .setMsgType(msgType) .setIdMSB(device.getId().getId().getMostSignificantBits()) @@ -38,6 +40,10 @@ public class DeviceUpdateMsgConstructor { if (device.getLabel() != null) { builder.setLabel(device.getLabel()); } + if (customerId != null) { + builder.setCustomerIdMSB(customerId.getId().getMostSignificantBits()); + builder.setCustomerIdLSB(customerId.getId().getLeastSignificantBits()); + } return builder.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 2ca0b4fcbe..af61e9e4ad 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,8 @@ 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.CustomerId; +import org.thingsboard.server.common.data.id.EntityId; import org.thingsboard.server.common.data.id.EntityViewId; import org.thingsboard.server.gen.edge.EdgeEntityType; import org.thingsboard.server.gen.edge.EntityViewUpdateMsg; @@ -27,7 +29,7 @@ import org.thingsboard.server.gen.edge.UpdateMsgType; @Slf4j public class EntityViewUpdateMsgConstructor { - public EntityViewUpdateMsg constructEntityViewUpdatedMsg(UpdateMsgType msgType, EntityView entityView) { + public EntityViewUpdateMsg constructEntityViewUpdatedMsg(UpdateMsgType msgType, EntityView entityView, CustomerId customerId) { EdgeEntityType entityType; switch (entityView.getEntityId().getEntityType()) { case DEVICE: @@ -48,6 +50,10 @@ public class EntityViewUpdateMsgConstructor { .setEntityIdMSB(entityView.getEntityId().getId().getMostSignificantBits()) .setEntityIdLSB(entityView.getEntityId().getId().getLeastSignificantBits()) .setEntityType(entityType); + if (customerId != null) { + builder.setCustomerIdMSB(customerId.getId().getMostSignificantBits()); + builder.setCustomerIdLSB(customerId.getId().getLeastSignificantBits()); + } return builder.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 b50e0c2b5e..4240d21292 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 @@ -18,6 +18,8 @@ package org.thingsboard.server.service.edge.rpc.constructor; import lombok.extern.slf4j.Slf4j; import org.springframework.stereotype.Component; import org.thingsboard.server.common.data.User; +import org.thingsboard.server.common.data.id.CustomerId; +import org.thingsboard.server.common.data.id.EntityId; import org.thingsboard.server.common.data.id.UserId; import org.thingsboard.server.common.data.security.UserCredentials; import org.thingsboard.server.dao.util.mapping.JacksonUtil; @@ -25,20 +27,22 @@ import org.thingsboard.server.gen.edge.UpdateMsgType; import org.thingsboard.server.gen.edge.UserCredentialsUpdateMsg; import org.thingsboard.server.gen.edge.UserUpdateMsg; +import java.util.UUID; + @Component @Slf4j public class UserUpdateMsgConstructor { - public UserUpdateMsg constructUserUpdatedMsg(UpdateMsgType msgType, User user) { + public UserUpdateMsg constructUserUpdatedMsg(UpdateMsgType msgType, User user, CustomerId customerId) { UserUpdateMsg.Builder builder = UserUpdateMsg.newBuilder() .setMsgType(msgType) .setIdMSB(user.getId().getId().getMostSignificantBits()) .setIdLSB(user.getId().getId().getLeastSignificantBits()) .setEmail(user.getEmail()) .setAuthority(user.getAuthority().name()); - if (user.getCustomerId() != null) { - builder.setCustomerIdMSB(user.getCustomerId().getId().getMostSignificantBits()); - builder.setCustomerIdLSB(user.getCustomerId().getId().getLeastSignificantBits()); + if (customerId != null) { + builder.setCustomerIdMSB(customerId.getId().getMostSignificantBits()); + builder.setCustomerIdLSB(customerId.getId().getLeastSignificantBits()); } if (user.getFirstName() != null) { builder.setFirstName(user.getFirstName()); diff --git a/application/src/main/java/org/thingsboard/server/service/edge/rpc/init/DefaultSyncEdgeService.java b/application/src/main/java/org/thingsboard/server/service/edge/rpc/init/DefaultSyncEdgeService.java index cfe8f86fd6..27ea1379f5 100644 --- a/application/src/main/java/org/thingsboard/server/service/edge/rpc/init/DefaultSyncEdgeService.java +++ b/application/src/main/java/org/thingsboard/server/service/edge/rpc/init/DefaultSyncEdgeService.java @@ -140,14 +140,14 @@ public class DefaultSyncEdgeService implements SyncEdgeService { @Override public void sync(Edge edge) { try { - syncUsers(edge); + syncWidgetsBundleAndWidgetTypes(edge); + syncAdminSettings(edge); syncRuleChains(edge); + syncUsers(edge); syncDevices(edge); syncAssets(edge); syncEntityViews(edge); syncDashboards(edge); - syncWidgetsBundleAndWidgetTypes(edge); - syncAdminSettings(edge); } catch (Exception e) { log.error("Exception during sync process", e); } diff --git a/common/edge-api/src/main/java/org/thingsboard/edge/rpc/EdgeGrpcClient.java b/common/edge-api/src/main/java/org/thingsboard/edge/rpc/EdgeGrpcClient.java index 45201d3229..57c1cf272e 100644 --- a/common/edge-api/src/main/java/org/thingsboard/edge/rpc/EdgeGrpcClient.java +++ b/common/edge-api/src/main/java/org/thingsboard/edge/rpc/EdgeGrpcClient.java @@ -104,7 +104,10 @@ public class EdgeGrpcClient implements EdgeRpcClient { @Override public void disconnect() throws InterruptedException { - inputStream.onCompleted(); + try { + inputStream.onCompleted(); + } catch (Exception e) { + } if (channel != null) { channel.shutdown().awaitTermination(timeoutSecs, TimeUnit.SECONDS); } @@ -151,6 +154,11 @@ public class EdgeGrpcClient implements EdgeRpcClient { onEdgeUpdate.accept(connectResponseMsg.getConfiguration()); } else { log.error("[{}] Failed to establish the connection! Code: {}. Error message: {}.", edgeKey, connectResponseMsg.getResponseCode(), connectResponseMsg.getErrorMsg()); + try { + EdgeGrpcClient.this.disconnect(); + } catch (InterruptedException e) { + log.error("[{}] Got interruption during disconnect!", edgeKey, e); + } onError.accept(new EdgeConnectionException("Failed to establish the connection! Response code: " + connectResponseMsg.getResponseCode().name())); } } else if (responseMsg.hasUplinkResponseMsg()) { diff --git a/common/edge-api/src/main/proto/edge.proto b/common/edge-api/src/main/proto/edge.proto index 3d91595a2f..35be11839f 100644 --- a/common/edge-api/src/main/proto/edge.proto +++ b/common/edge-api/src/main/proto/edge.proto @@ -154,17 +154,21 @@ message DashboardUpdateMsg { UpdateMsgType msgType = 1; int64 idMSB = 2; int64 idLSB = 3; - string title = 4; - string configuration = 5; + int64 customerIdMSB = 4; + int64 customerIdLSB = 5; + string title = 6; + string configuration = 7; } message DeviceUpdateMsg { UpdateMsgType msgType = 1; int64 idMSB = 2; int64 idLSB = 3; - string name = 4; - string type = 5; - string label = 6; + int64 customerIdMSB = 4; + int64 customerIdLSB = 5; + string name = 6; + string type = 7; + string label = 8; } message DeviceCredentialsUpdateMsg { @@ -179,20 +183,24 @@ message AssetUpdateMsg { UpdateMsgType msgType = 1; int64 idMSB = 2; int64 idLSB = 3; - string name = 4; - string type = 5; - string label = 6; + int64 customerIdMSB = 4; + int64 customerIdLSB = 5; + string name = 6; + string type = 7; + string label = 8; } message EntityViewUpdateMsg { UpdateMsgType msgType = 1; int64 idMSB = 2; int64 idLSB = 3; - string name = 4; - string type = 5; - int64 entityIdMSB = 6; - int64 entityIdLSB = 7; - EdgeEntityType entityType = 8; + int64 customerIdMSB = 4; + int64 customerIdLSB = 5; + string name = 6; + string type = 7; + int64 entityIdMSB = 8; + int64 entityIdLSB = 9; + EdgeEntityType entityType = 10; } message AlarmUpdateMsg { diff --git a/dao/src/main/java/org/thingsboard/server/dao/asset/BaseAssetService.java b/dao/src/main/java/org/thingsboard/server/dao/asset/BaseAssetService.java index 62f68e962c..d23f840938 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/asset/BaseAssetService.java +++ b/dao/src/main/java/org/thingsboard/server/dao/asset/BaseAssetService.java @@ -30,7 +30,6 @@ import org.springframework.cache.annotation.Cacheable; import org.springframework.stereotype.Service; import org.springframework.util.StringUtils; import org.thingsboard.server.common.data.Customer; -import org.thingsboard.server.common.data.Device; import org.thingsboard.server.common.data.EntitySubtype; import org.thingsboard.server.common.data.EntityType; import org.thingsboard.server.common.data.EntityView; @@ -53,7 +52,6 @@ import org.thingsboard.server.common.data.relation.RelationTypeGroup; import org.thingsboard.server.dao.customer.CustomerDao; import org.thingsboard.server.dao.edge.EdgeService; import org.thingsboard.server.dao.entity.AbstractEntityService; -import org.thingsboard.server.dao.entityview.EntityViewService; import org.thingsboard.server.dao.exception.DataValidationException; import org.thingsboard.server.dao.service.DataValidator; import org.thingsboard.server.dao.service.PaginatedRemover; @@ -92,9 +90,6 @@ public class BaseAssetService extends AbstractEntityService implements AssetServ @Autowired private CustomerDao customerDao; - @Autowired - private EntityViewService entityViewService; - @Autowired private EdgeService edgeService; @@ -317,6 +312,9 @@ public class BaseAssetService extends AbstractEntityService implements AssetServ if (edge == null) { throw new DataValidationException("Can't unassign asset from non-existent edge!"); } + + checkAssignedEntityViewsToEdge(tenantId, assetId, edgeId); + try { deleteRelation(tenantId, new EntityRelation(edgeId, assetId, EntityRelation.CONTAINS_TYPE, RelationTypeGroup.EDGE)); } catch (ExecutionException | InterruptedException e) { diff --git a/dao/src/main/java/org/thingsboard/server/dao/device/DeviceServiceImpl.java b/dao/src/main/java/org/thingsboard/server/dao/device/DeviceServiceImpl.java index c8f24262a1..31d3f1a7ab 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/device/DeviceServiceImpl.java +++ b/dao/src/main/java/org/thingsboard/server/dao/device/DeviceServiceImpl.java @@ -41,7 +41,6 @@ import org.thingsboard.server.common.data.id.CustomerId; import org.thingsboard.server.common.data.id.DeviceId; import org.thingsboard.server.common.data.id.EdgeId; import org.thingsboard.server.common.data.id.EntityId; -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; @@ -50,17 +49,14 @@ import org.thingsboard.server.common.data.page.TimePageLink; import org.thingsboard.server.common.data.relation.EntityRelation; import org.thingsboard.server.common.data.relation.EntitySearchDirection; import org.thingsboard.server.common.data.relation.RelationTypeGroup; -import org.thingsboard.server.common.data.rule.RuleChain; import org.thingsboard.server.common.data.security.DeviceCredentials; import org.thingsboard.server.common.data.security.DeviceCredentialsType; import org.thingsboard.server.dao.customer.CustomerDao; import org.thingsboard.server.dao.edge.EdgeService; import org.thingsboard.server.dao.entity.AbstractEntityService; -import org.thingsboard.server.dao.entityview.EntityViewService; import org.thingsboard.server.dao.exception.DataValidationException; import org.thingsboard.server.dao.service.DataValidator; import org.thingsboard.server.dao.service.PaginatedRemover; -import org.thingsboard.server.dao.service.Validator; import org.thingsboard.server.dao.tenant.TenantDao; import javax.annotation.Nullable; @@ -102,9 +98,6 @@ public class DeviceServiceImpl extends AbstractEntityService implements DeviceSe @Autowired private DeviceCredentialsService deviceCredentialsService; - @Autowired - private EntityViewService entityViewService; - @Autowired private EdgeService edgeService; @@ -357,6 +350,9 @@ public class DeviceServiceImpl extends AbstractEntityService implements DeviceSe if (edge == null) { throw new DataValidationException("Can't unassign device from non-existent edge!"); } + + checkAssignedEntityViewsToEdge(tenantId, deviceId, edgeId); + try { deleteRelation(tenantId, new EntityRelation(edgeId, deviceId, EntityRelation.CONTAINS_TYPE, RelationTypeGroup.EDGE)); } catch (ExecutionException | InterruptedException e) { diff --git a/dao/src/main/java/org/thingsboard/server/dao/edge/EdgeServiceImpl.java b/dao/src/main/java/org/thingsboard/server/dao/edge/EdgeServiceImpl.java index 01f82faa17..0cc5c056d8 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/edge/EdgeServiceImpl.java +++ b/dao/src/main/java/org/thingsboard/server/dao/edge/EdgeServiceImpl.java @@ -41,14 +41,15 @@ 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.EntitySearchDirection; import org.thingsboard.server.common.data.rule.RuleChain; +import org.thingsboard.server.dao.asset.AssetService; import org.thingsboard.server.dao.customer.CustomerDao; import org.thingsboard.server.dao.dashboard.DashboardService; +import org.thingsboard.server.dao.device.DeviceService; import org.thingsboard.server.dao.entity.AbstractEntityService; +import org.thingsboard.server.dao.entityview.EntityViewService; import org.thingsboard.server.dao.exception.DataValidationException; import org.thingsboard.server.dao.relation.RelationService; import org.thingsboard.server.dao.rule.RuleChainService; @@ -94,6 +95,15 @@ public class EdgeServiceImpl extends AbstractEntityService implements EdgeServic @Autowired private CacheManager cacheManager; + @Autowired + private AssetService assetService; + + @Autowired + private DeviceService deviceService; + + @Autowired + private EntityViewService entityViewService; + @Autowired private DashboardService dashboardService; diff --git a/dao/src/main/java/org/thingsboard/server/dao/entity/AbstractEntityService.java b/dao/src/main/java/org/thingsboard/server/dao/entity/AbstractEntityService.java index 523df59db5..deb2f75c6e 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/entity/AbstractEntityService.java +++ b/dao/src/main/java/org/thingsboard/server/dao/entity/AbstractEntityService.java @@ -19,12 +19,18 @@ import lombok.extern.slf4j.Slf4j; import org.hibernate.exception.ConstraintViolationException; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Value; +import org.thingsboard.server.common.data.EntityView; +import org.thingsboard.server.common.data.id.EdgeId; import org.thingsboard.server.common.data.id.EntityId; import org.thingsboard.server.common.data.id.TenantId; import org.thingsboard.server.common.data.relation.EntityRelation; +import org.thingsboard.server.common.data.relation.RelationTypeGroup; +import org.thingsboard.server.dao.entityview.EntityViewService; +import org.thingsboard.server.dao.exception.DataValidationException; import org.thingsboard.server.dao.relation.RelationService; import javax.annotation.PostConstruct; +import java.util.List; import java.util.Optional; import java.util.concurrent.ExecutionException; @@ -37,6 +43,9 @@ public abstract class AbstractEntityService { @Autowired protected RelationService relationService; + @Autowired + protected EntityViewService entityViewService; + @Value("${database.entities.type:sql}") private String databaseType; @@ -72,4 +81,21 @@ public abstract class AbstractEntityService { } } + protected void checkAssignedEntityViewsToEdge(TenantId tenantId, EntityId entityId, EdgeId edgeId) { + try { + List entityViews = entityViewService.findEntityViewsByTenantIdAndEntityIdAsync(tenantId, entityId).get(); + if (entityViews != null && !entityViews.isEmpty()) { + EntityView entityView = entityViews.get(0); + Boolean relationExists = relationService.checkRelation(tenantId,edgeId, entityView.getId(), + EntityRelation.CONTAINS_TYPE, RelationTypeGroup.EDGE).get(); + if (relationExists) { + throw new DataValidationException("Can't unassign device/asset from edge that is related to entity view and entity view is assigned to edge!"); + } + } + } catch (ExecutionException | InterruptedException e) { + log.error("Exception while finding entity views for entityId [{}]", entityId, e); + throw new RuntimeException("Exception while finding entity views for entityId [" + entityId + "]", e); + } + } + } diff --git a/dao/src/main/java/org/thingsboard/server/dao/entityview/EntityViewServiceImpl.java b/dao/src/main/java/org/thingsboard/server/dao/entityview/EntityViewServiceImpl.java index 907072e549..8729085ded 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/entityview/EntityViewServiceImpl.java +++ b/dao/src/main/java/org/thingsboard/server/dao/entityview/EntityViewServiceImpl.java @@ -305,6 +305,18 @@ public class EntityViewServiceImpl extends AbstractEntityService implements Enti if (!edge.getTenantId().getId().equals(entityView.getTenantId().getId())) { throw new DataValidationException("Can't assign entityView to edge from different tenant!"); } + + try { + Boolean relationExists = relationService.checkRelation(tenantId, edgeId, entityView.getEntityId(), + EntityRelation.CONTAINS_TYPE, RelationTypeGroup.EDGE).get(); + if (!relationExists) { + throw new DataValidationException("Can't assign entity view to edge because related device/asset doesn't assigned to edge!"); + } + } catch (ExecutionException | InterruptedException e) { + log.error("Exception during relation check", e); + throw new RuntimeException("Exception during relation check", e); + } + try { createRelation(tenantId, new EntityRelation(edgeId, entityViewId, EntityRelation.CONTAINS_TYPE, RelationTypeGroup.EDGE)); } catch (ExecutionException | InterruptedException e) {