tsai
/
thingsboard
mirror of https://github.com/thingsboard/thingsboard.git
1
0
Fork 0
Code Issues Projects Releases Wiki Activity
Browse Source

Refactored telemetry edge processor to reuse in whitelabeling processor

pull/7478/head
Volodymyr Babak 4 years ago
parent
60998c6f6b
commit
b13ec070cf
2 changed files with 34 additions and 28 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Split View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 33
      application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/BaseEdgeProcessor.java
  2. 29
      application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/TelemetryEdgeProcessor.java

33
application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/BaseEdgeProcessor.java
View File

@ -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;
}
}
}

29
application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/TelemetryEdgeProcessor.java
View File

@ -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<ListenableFuture<Void>> processTelemetryFromEdge(TenantId tenantId, EntityDataProto entityData) {
log.trace("[{}] processTelemetryFromEdge [{}]", tenantId, entityData);
List<ListenableFuture<Void>> 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<TbMsgMetaData, CustomerId> 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()) {

Write Preview
Loading…
Cancel
Save