From 20c794884a5d1c0a2fb8290610e192e84994fc63 Mon Sep 17 00:00:00 2001 From: Bohdan Smetaniuk Date: Thu, 1 Oct 2020 13:55:30 +0300 Subject: [PATCH] attributes updated/post attributes separation --- .../constructor/EntityDataMsgConstructor.java | 9 +++- .../rpc/processor/TelemetryProcessor.java | 30 +++++++++-- common/edge-api/src/main/proto/edge.proto | 5 +- .../rule/engine/edge/TbMsgPushToEdgeNode.java | 50 +++++++++---------- 4 files changed, 61 insertions(+), 33 deletions(-) diff --git a/application/src/main/java/org/thingsboard/server/service/edge/rpc/constructor/EntityDataMsgConstructor.java b/application/src/main/java/org/thingsboard/server/service/edge/rpc/constructor/EntityDataMsgConstructor.java index 0f7fb0523d..5bbea5351f 100644 --- a/application/src/main/java/org/thingsboard/server/service/edge/rpc/constructor/EntityDataMsgConstructor.java +++ b/application/src/main/java/org/thingsboard/server/service/edge/rpc/constructor/EntityDataMsgConstructor.java @@ -18,7 +18,6 @@ package org.thingsboard.server.service.edge.rpc.constructor; import com.google.gson.Gson; import com.google.gson.JsonArray; import com.google.gson.JsonElement; -import com.google.gson.JsonNull; import com.google.gson.JsonObject; import lombok.extern.slf4j.Slf4j; import org.springframework.stereotype.Component; @@ -27,6 +26,7 @@ import org.thingsboard.server.common.data.id.EntityId; import org.thingsboard.server.common.transport.adaptor.JsonConverter; import org.thingsboard.server.gen.edge.AttributeDeleteMsg; import org.thingsboard.server.gen.edge.EntityDataProto; +import org.thingsboard.server.gen.transport.TransportProtos; import java.util.List; @@ -57,7 +57,12 @@ public class EntityDataMsgConstructor { case ATTRIBUTES_UPDATED: try { JsonObject data = entityData.getAsJsonObject(); - builder.setPostAttributesMsg(JsonConverter.convertToAttributesProto(data.getAsJsonObject("kv"))); + TransportProtos.PostAttributeMsg postAttributeMsg = JsonConverter.convertToAttributesProto(data.getAsJsonObject("kv")); + if (data.has("isPostAttributes") && data.getAsJsonPrimitive("isPostAttributes").getAsBoolean()) { + builder.setPostAttributesMsg(postAttributeMsg); + } else { + builder.setAttributesUpdateMsg(postAttributeMsg); + } builder.setPostAttributeScope(data.getAsJsonPrimitive("scope").getAsString()); } catch (Exception e) { log.warn("Can't convert to attributes proto, entityData [{}]", entityData, e); diff --git a/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/TelemetryProcessor.java b/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/TelemetryProcessor.java index cc2cb15d4b..da5dc3746f 100644 --- a/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/TelemetryProcessor.java +++ b/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/TelemetryProcessor.java @@ -68,13 +68,16 @@ public class TelemetryProcessor extends BaseProcessor { public List> onTelemetryUpdate(TenantId tenantId, EntityDataProto entityData) { List> result = new ArrayList<>(); EntityId entityId = constructEntityId(entityData); - if ((entityData.hasPostAttributesMsg() || entityData.hasPostTelemetryMsg()) && entityId != null) { + if ((entityData.hasPostAttributesMsg() || entityData.hasPostTelemetryMsg() || entityData.hasAttributesUpdateMsg()) && entityId != null) { TbMsgMetaData metaData = constructBaseMsgMetadata(tenantId, entityId); metaData.putValue(DataConstants.MSG_SOURCE_KEY, DataConstants.EDGE_MSG_SOURCE); if (entityData.hasPostAttributesMsg()) { - metaData.putValue("scope", entityData.getPostAttributeScope()); result.add(processPostAttributes(tenantId, entityId, entityData.getPostAttributesMsg(), metaData)); } + if (entityData.hasAttributesUpdateMsg()) { + metaData.putValue("scope", entityData.getPostAttributeScope()); + result.add(processAttributesUpdate(tenantId, entityId, entityData.getAttributesUpdateMsg(), metaData)); + } if (entityData.hasPostTelemetryMsg()) { result.add(processPostTelemetry(tenantId, entityId, entityData.getPostTelemetryMsg(), metaData)); } @@ -139,6 +142,25 @@ public class TelemetryProcessor extends BaseProcessor { } private ListenableFuture processPostAttributes(TenantId tenantId, EntityId entityId, TransportProtos.PostAttributeMsg msg, TbMsgMetaData metaData) { + SettableFuture futureToSet = SettableFuture.create(); + JsonObject json = JsonUtils.getJsonObject(msg.getKvList()); + TbMsg tbMsg = TbMsg.newMsg(SessionMsgType.POST_ATTRIBUTES_REQUEST.name(), entityId, metaData, gson.toJson(json)); + tbClusterService.pushMsgToRuleEngine(tenantId, tbMsg.getOriginator(), tbMsg, new TbQueueCallback() { + @Override + public void onSuccess(TbQueueMsgMetadata metadata) { + futureToSet.set(null); + } + + @Override + public void onFailure(Throwable t) { + log.error("Can't process post attributes [{}]", msg, t); + futureToSet.setException(t); + } + }); + return futureToSet; + } + + private ListenableFuture processAttributesUpdate(TenantId tenantId, EntityId entityId, TransportProtos.PostAttributeMsg msg, TbMsgMetaData metaData) { SettableFuture futureToSet = SettableFuture.create(); JsonObject json = JsonUtils.getJsonObject(msg.getKvList()); Set attributes = JsonConverter.convertToAttributes(json); @@ -155,7 +177,7 @@ public class TelemetryProcessor extends BaseProcessor { @Override public void onFailure(Throwable t) { - log.error("Can't process post attributes [{}]", msg, t); + log.error("Can't process attributes update [{}]", msg, t); futureToSet.setException(t); } }); @@ -163,7 +185,7 @@ public class TelemetryProcessor extends BaseProcessor { @Override public void onFailure(Throwable t) { - log.error("Can't process post attributes [{}]", msg, t); + log.error("Can't process attributes update [{}]", msg, t); futureToSet.setException(t); } }, dbCallbackExecutorService); diff --git a/common/edge-api/src/main/proto/edge.proto b/common/edge-api/src/main/proto/edge.proto index c2ce170401..e5aa03cfa2 100644 --- a/common/edge-api/src/main/proto/edge.proto +++ b/common/edge-api/src/main/proto/edge.proto @@ -102,8 +102,9 @@ message EntityDataProto { string entityType = 3; transport.PostTelemetryMsg postTelemetryMsg = 4; transport.PostAttributeMsg postAttributesMsg = 5; - string postAttributeScope = 6; - AttributeDeleteMsg attributeDeleteMsg = 7; + transport.PostAttributeMsg attributesUpdateMsg = 6; + string postAttributeScope = 7; + AttributeDeleteMsg attributeDeleteMsg = 8; // transport.ToDeviceRpcRequestMsg ??? } diff --git a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/edge/TbMsgPushToEdgeNode.java b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/edge/TbMsgPushToEdgeNode.java index 8d00507a7d..b510af10b0 100644 --- a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/edge/TbMsgPushToEdgeNode.java +++ b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/edge/TbMsgPushToEdgeNode.java @@ -138,16 +138,37 @@ public class TbMsgPushToEdgeNode implements TbNode { } private EdgeEvent buildEdgeEvent(TbMsg msg, TbContext ctx) throws JsonProcessingException { - if (DataConstants.ALARM.equals(msg.getType())) { + String msgType = msg.getType(); + if (DataConstants.ALARM.equals(msgType)) { return buildEdgeEvent(ctx.getTenantId(), ActionType.ADDED, getUUIDFromMsgData(msg), EdgeEventType.ALARM, null); } else { EdgeEventType edgeEventTypeByEntityType = EdgeUtils.getEdgeEventTypeByEntityType(msg.getOriginator().getEntityType()); if (edgeEventTypeByEntityType == null) { return null; } - ActionType actionType = getActionTypeByMsgType(msg.getType()); - JsonNode entityBody = getEntityBody(actionType, msg.getData(), msg.getMetaData().getData()); - return buildEdgeEvent(ctx.getTenantId(), actionType, msg.getOriginator().getId(), edgeEventTypeByEntityType, entityBody); + ActionType actionType = getActionTypeByMsgType(msgType); + Map entityBody = new HashMap<>(); + Map metadata = msg.getMetaData().getData(); + JsonNode dataJson = json.readTree(msg.getData()); + switch (actionType) { + case ATTRIBUTES_UPDATED: + entityBody.put("kv", dataJson); + entityBody.put("scope", metadata.get("scope")); + if (SessionMsgType.POST_ATTRIBUTES_REQUEST.name().equals(msgType)) { + entityBody.put("isPostAttributes", true); + } + break; + case ATTRIBUTES_DELETED: + List keys = json.treeToValue(dataJson.get("attributes"), List.class); + entityBody.put("keys", keys); + entityBody.put("scope", metadata.get("scope")); + break; + case TIMESERIES_UPDATED: + entityBody.put("data", dataJson); + entityBody.put("ts", metadata.get("ts")); + break; + } + return buildEdgeEvent(ctx.getTenantId(), actionType, msg.getOriginator().getId(), edgeEventTypeByEntityType, json.valueToTree(entityBody)); } } @@ -161,27 +182,6 @@ public class TbMsgPushToEdgeNode implements TbNode { return edgeEvent; } - private JsonNode getEntityBody(ActionType actionType, String data, Map metadata) throws JsonProcessingException { - Map entityBody = new HashMap<>(); - JsonNode dataJson = json.readTree(data); - switch (actionType) { - case ATTRIBUTES_UPDATED: - entityBody.put("kv", dataJson); - entityBody.put("scope", metadata.get("scope")); - break; - case ATTRIBUTES_DELETED: - List keys = json.treeToValue(dataJson.get("attributes"), List.class); - entityBody.put("keys", keys); - entityBody.put("scope", metadata.get("scope")); - break; - case TIMESERIES_UPDATED: - entityBody.put("data", dataJson); - entityBody.put("ts", metadata.get("ts")); - break; - } - return json.valueToTree(entityBody); - } - private UUID getUUIDFromMsgData(TbMsg msg) throws JsonProcessingException { JsonNode data = json.readTree(msg.getData()).get("id"); String id = json.treeToValue(data.get("id"), String.class);