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

attributes updated/post attributes separation

pull/2436/head
Bohdan Smetaniuk 6 years ago
parent
64e60de9bc
commit
20c794884a
4 changed files with 61 additions and 33 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. 9
      application/src/main/java/org/thingsboard/server/service/edge/rpc/constructor/EntityDataMsgConstructor.java
  2. 30
      application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/TelemetryProcessor.java
  3. 5
      common/edge-api/src/main/proto/edge.proto
  4. 50
      rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/edge/TbMsgPushToEdgeNode.java

9
application/src/main/java/org/thingsboard/server/service/edge/rpc/constructor/EntityDataMsgConstructor.java
View File

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

30
application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/TelemetryProcessor.java
View File

@ -68,13 +68,16 @@ public class TelemetryProcessor extends BaseProcessor {
public List<ListenableFuture<Void>> onTelemetryUpdate(TenantId tenantId, EntityDataProto entityData) {
List<ListenableFuture<Void>> 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<Void> processPostAttributes(TenantId tenantId, EntityId entityId, TransportProtos.PostAttributeMsg msg, TbMsgMetaData metaData) {
SettableFuture<Void> 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<Void> processAttributesUpdate(TenantId tenantId, EntityId entityId, TransportProtos.PostAttributeMsg msg, TbMsgMetaData metaData) {
SettableFuture<Void> futureToSet = SettableFuture.create();
JsonObject json = JsonUtils.getJsonObject(msg.getKvList());
Set<AttributeKvEntry> 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);

5
common/edge-api/src/main/proto/edge.proto
View File

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

50
rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/edge/TbMsgPushToEdgeNode.java
View File

@ -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<String, Object> entityBody = new HashMap<>();
Map<String, String> 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<String> 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<String, String> metadata) throws JsonProcessingException {
Map<String, Object> 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<String> 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);

Write Preview
Loading…
Cancel
Save