From 2b2a513cd49227188f44443b395c0c9bad83f0fd Mon Sep 17 00:00:00 2001 From: Bohdan Smetaniuk Date: Tue, 18 Aug 2020 15:46:49 +0300 Subject: [PATCH] propagate relations delete/update --- .../service/edge/rpc/EdgeGrpcSession.java | 63 +++++++++++++++++++ common/edge-api/src/main/proto/edge.proto | 11 ++-- 2 files changed, 69 insertions(+), 5 deletions(-) diff --git a/application/src/main/java/org/thingsboard/server/service/edge/rpc/EdgeGrpcSession.java b/application/src/main/java/org/thingsboard/server/service/edge/rpc/EdgeGrpcSession.java index 519f4529f1..9b3c565092 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 @@ -47,6 +47,8 @@ import org.thingsboard.server.common.data.asset.Asset; import org.thingsboard.server.common.data.audit.ActionType; import org.thingsboard.server.common.data.edge.Edge; import org.thingsboard.server.common.data.edge.EdgeEvent; +import org.thingsboard.server.common.data.exception.ThingsboardErrorCode; +import org.thingsboard.server.common.data.exception.ThingsboardException; import org.thingsboard.server.common.data.id.AlarmId; import org.thingsboard.server.common.data.id.AssetId; import org.thingsboard.server.common.data.id.CustomerId; @@ -54,6 +56,7 @@ import org.thingsboard.server.common.data.id.DashboardId; 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.EntityIdFactory; import org.thingsboard.server.common.data.id.EntityViewId; import org.thingsboard.server.common.data.id.RuleChainId; import org.thingsboard.server.common.data.id.TenantId; @@ -98,6 +101,7 @@ import org.thingsboard.server.gen.edge.EntityDataProto; import org.thingsboard.server.gen.edge.EntityUpdateMsg; import org.thingsboard.server.gen.edge.EntityViewUpdateMsg; import org.thingsboard.server.gen.edge.RelationRequestMsg; +import org.thingsboard.server.gen.edge.RelationUpdateMsg; import org.thingsboard.server.gen.edge.RequestMsg; import org.thingsboard.server.gen.edge.RequestMsgType; import org.thingsboard.server.gen.edge.ResponseMsg; @@ -813,6 +817,11 @@ public final class EdgeGrpcSession implements Closeable { onAlarmUpdate(alarmUpdateMsg); } } + if (uplinkMsg.getRelationUpdateMsgList() != null && !uplinkMsg.getRelationUpdateMsgList().isEmpty()) { + for (RelationUpdateMsg relationUpdateMsg: uplinkMsg.getRelationUpdateMsgList()) { + onRelationUpdate(relationUpdateMsg); + } + } if (uplinkMsg.getRuleChainMetadataRequestMsgList() != null && !uplinkMsg.getRuleChainMetadataRequestMsgList().isEmpty()) { for (RuleChainMetadataRequestMsg ruleChainMetadataRequestMsg : uplinkMsg.getRuleChainMetadataRequestMsgList()) { ctx.getSyncEdgeService().processRuleChainMetadataRequestMsg(edge, ruleChainMetadataRequestMsg); @@ -1164,6 +1173,60 @@ public final class EdgeGrpcSession implements Closeable { } } + private void onRelationUpdate(RelationUpdateMsg relationUpdateMsg) { + log.info("onRelationUpdate {}", relationUpdateMsg); + try { + EntityRelation entityRelation = new EntityRelation(); + + UUID fromUUID = new UUID(relationUpdateMsg.getFromIdMSB(), relationUpdateMsg.getFromIdLSB()); + EntityId fromId = EntityIdFactory.getByTypeAndUuid(EntityType.valueOf(relationUpdateMsg.getFromEntityType()), fromUUID); + entityRelation.setFrom(fromId); + + UUID toUUID = new UUID(relationUpdateMsg.getToIdMSB(), relationUpdateMsg.getToIdLSB()); + EntityId toId = EntityIdFactory.getByTypeAndUuid(EntityType.valueOf(relationUpdateMsg.getToEntityType()), toUUID); + entityRelation.setTo(toId); + + entityRelation.setType(relationUpdateMsg.getType()); + entityRelation.setTypeGroup(RelationTypeGroup.valueOf(relationUpdateMsg.getTypeGroup())); + entityRelation.setAdditionalInfo(mapper.readTree(relationUpdateMsg.getAdditionalInfo())); + switch (relationUpdateMsg.getMsgType()) { + case ENTITY_CREATED_RPC_MESSAGE: + case ENTITY_UPDATED_RPC_MESSAGE: + if (isEntityExists(edge.getTenantId(), entityRelation.getTo()) + && isEntityExists(edge.getTenantId(), entityRelation.getFrom())) { + ctx.getRelationService().saveRelationAsync(edge.getTenantId(), entityRelation); + } + break; + case ENTITY_DELETED_RPC_MESSAGE: + ctx.getRelationService().deleteRelation(edge.getTenantId(), entityRelation); + break; + case UNRECOGNIZED: + log.error("Unsupported msg type"); + } + } catch (Exception e) { + log.error("Error during relation update msg", e); + } + } + + private boolean isEntityExists(TenantId tenantId, EntityId entityId) throws ThingsboardException { + switch (entityId.getEntityType()) { + case DEVICE: + return ctx.getDeviceService().findDeviceById(tenantId, new DeviceId(entityId.getId())) != null; + case ASSET: + return ctx.getAssetService().findAssetById(tenantId, new AssetId(entityId.getId())) != null; + case ENTITY_VIEW: + return ctx.getEntityViewService().findEntityViewById(tenantId, new EntityViewId(entityId.getId())) != null; + case CUSTOMER: + return ctx.getCustomerService().findCustomerById(tenantId, new CustomerId(entityId.getId())) != null; + case USER: + return ctx.getUserService().findUserById(tenantId, new UserId(entityId.getId())) != null; + case DASHBOARD: + return ctx.getDashboardService().findDashboardById(tenantId, new DashboardId(entityId.getId())) != null; + default: + throw new ThingsboardException("Unsupported entity type " + entityId.getEntityType(), ThingsboardErrorCode.INVALID_ARGUMENTS); + } + } + private ConnectResponseMsg processConnect(ConnectRequestMsg request) { Optional optional = ctx.getEdgeService().findEdgeByRoutingKey(TenantId.SYS_TENANT_ID, request.getEdgeRoutingKey()); if (optional.isPresent()) { diff --git a/common/edge-api/src/main/proto/edge.proto b/common/edge-api/src/main/proto/edge.proto index 8be4be9956..cce4bef73a 100644 --- a/common/edge-api/src/main/proto/edge.proto +++ b/common/edge-api/src/main/proto/edge.proto @@ -348,11 +348,12 @@ message UplinkMsg { repeated DeviceUpdateMsg deviceUpdateMsg = 3; repeated DeviceCredentialsUpdateMsg deviceCredentialsUpdateMsg = 4; repeated AlarmUpdateMsg alarmUpdateMsg = 5; - repeated RuleChainMetadataRequestMsg ruleChainMetadataRequestMsg = 6; - repeated AttributesRequestMsg attributesRequestMsg = 7; - repeated RelationRequestMsg relationRequestMsg = 8; - repeated UserCredentialsRequestMsg userCredentialsRequestMsg = 9; - repeated DeviceCredentialsRequestMsg deviceCredentialsRequestMsg = 10; + repeated RelationUpdateMsg relationUpdateMsg = 6; + repeated RuleChainMetadataRequestMsg ruleChainMetadataRequestMsg = 7; + repeated AttributesRequestMsg attributesRequestMsg = 8; + repeated RelationRequestMsg relationRequestMsg = 9; + repeated UserCredentialsRequestMsg userCredentialsRequestMsg = 10; + repeated DeviceCredentialsRequestMsg deviceCredentialsRequestMsg = 11; } message UplinkResponseMsg {