From b13ec070cf9f8822d4e46b226d5fe06079e93994 Mon Sep 17 00:00:00 2001 From: Volodymyr Babak Date: Thu, 27 Oct 2022 15:50:09 +0300 Subject: [PATCH] Refactored telemetry edge processor to reuse in whitelabeling processor --- .../edge/rpc/processor/BaseEdgeProcessor.java | 33 +++++++++++++++++++ .../rpc/processor/TelemetryEdgeProcessor.java | 29 +--------------- 2 files changed, 34 insertions(+), 28 deletions(-) diff --git a/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/BaseEdgeProcessor.java b/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/BaseEdgeProcessor.java index a5b315176a..f404b21782 100644 --- a/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/BaseEdgeProcessor.java +++ b/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/BaseEdgeProcessor.java @@ -24,15 +24,22 @@ import org.springframework.context.annotation.Lazy; import org.thingsboard.server.cluster.TbClusterService; import org.thingsboard.server.common.data.Device; import org.thingsboard.server.common.data.EdgeUtils; +import org.thingsboard.server.common.data.EntityType; import org.thingsboard.server.common.data.edge.Edge; import org.thingsboard.server.common.data.edge.EdgeEvent; import org.thingsboard.server.common.data.edge.EdgeEventActionType; import org.thingsboard.server.common.data.edge.EdgeEventType; +import org.thingsboard.server.common.data.id.AssetId; +import org.thingsboard.server.common.data.id.CustomerId; +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; +import org.thingsboard.server.common.data.id.UserId; import org.thingsboard.server.common.data.page.PageData; import org.thingsboard.server.common.data.page.PageLink; import org.thingsboard.server.common.data.rule.RuleChain; @@ -425,4 +432,30 @@ public abstract class BaseEdgeProcessor { return Futures.immediateFuture(null); } } + + protected EntityId constructEntityId(String entityTypeStr, long entityIdMSB, long entityIdLSB) { + EntityType entityType = EntityType.valueOf(entityTypeStr); + switch (entityType) { + case DEVICE: + return new DeviceId(new UUID(entityIdMSB, entityIdLSB)); + case ASSET: + return new AssetId(new UUID(entityIdMSB, entityIdLSB)); + case ENTITY_VIEW: + return new EntityViewId(new UUID(entityIdMSB, entityIdLSB)); + case DASHBOARD: + return new DashboardId(new UUID(entityIdMSB, entityIdLSB)); + case TENANT: + return TenantId.fromUUID(new UUID(entityIdMSB, entityIdLSB)); + case CUSTOMER: + return new CustomerId(new UUID(entityIdMSB, entityIdLSB)); + case USER: + return new UserId(new UUID(entityIdMSB, entityIdLSB)); + case EDGE: + return new EdgeId(new UUID(entityIdMSB, entityIdLSB)); + default: + log.warn("Unsupported entity type [{}] during construct of entity id. entityIdMSB [{}], entityIdLSB [{}]", + entityTypeStr, entityIdMSB, entityIdLSB); + return null; + } + } } diff --git a/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/TelemetryEdgeProcessor.java b/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/TelemetryEdgeProcessor.java index e326616963..722b33b426 100644 --- a/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/TelemetryEdgeProcessor.java +++ b/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/TelemetryEdgeProcessor.java @@ -48,7 +48,6 @@ import org.thingsboard.server.common.data.id.EntityId; import org.thingsboard.server.common.data.id.EntityViewId; import org.thingsboard.server.common.data.id.RuleChainId; import org.thingsboard.server.common.data.id.TenantId; -import org.thingsboard.server.common.data.id.UserId; import org.thingsboard.server.common.data.kv.AttributeKey; import org.thingsboard.server.common.data.kv.AttributeKvEntry; import org.thingsboard.server.common.msg.TbMsg; @@ -75,7 +74,6 @@ import java.util.ArrayList; import java.util.HashSet; import java.util.List; import java.util.Set; -import java.util.UUID; @Component @Slf4j @@ -94,7 +92,7 @@ public class TelemetryEdgeProcessor extends BaseEdgeProcessor { public List> processTelemetryFromEdge(TenantId tenantId, EntityDataProto entityData) { log.trace("[{}] processTelemetryFromEdge [{}]", tenantId, entityData); List> result = new ArrayList<>(); - EntityId entityId = constructEntityId(entityData); + EntityId entityId = constructEntityId(entityData.getEntityType(), entityData.getEntityIdMSB(), entityData.getEntityIdLSB()); if ((entityData.hasPostAttributesMsg() || entityData.hasPostTelemetryMsg() || entityData.hasAttributesUpdatedMsg()) && entityId != null) { Pair pair = getBaseMsgMetadataAndCustomerId(tenantId, entityId); TbMsgMetaData metaData = pair.getKey(); @@ -310,31 +308,6 @@ public class TelemetryEdgeProcessor extends BaseEdgeProcessor { return futureToSet; } - private EntityId constructEntityId(EntityDataProto entityData) { - EntityType entityType = EntityType.valueOf(entityData.getEntityType()); - switch (entityType) { - case DEVICE: - return new DeviceId(new UUID(entityData.getEntityIdMSB(), entityData.getEntityIdLSB())); - case ASSET: - return new AssetId(new UUID(entityData.getEntityIdMSB(), entityData.getEntityIdLSB())); - case ENTITY_VIEW: - return new EntityViewId(new UUID(entityData.getEntityIdMSB(), entityData.getEntityIdLSB())); - case DASHBOARD: - return new DashboardId(new UUID(entityData.getEntityIdMSB(), entityData.getEntityIdLSB())); - case TENANT: - return TenantId.fromUUID(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())); - case EDGE: - return new EdgeId(new UUID(entityData.getEntityIdMSB(), entityData.getEntityIdLSB())); - default: - log.warn("Unsupported entity type [{}] during construct of entity id. EntityDataProto [{}]", entityData.getEntityType(), entityData); - return null; - } - } - public DownlinkMsg convertTelemetryEventToDownlink(EdgeEvent edgeEvent) throws JsonProcessingException { EntityId entityId; switch (edgeEvent.getType()) {