From 670a218e6da7643f8a15801b0be5d9828ead7ba1 Mon Sep 17 00:00:00 2001 From: Andrii Shvaika Date: Mon, 25 May 2020 15:28:55 +0300 Subject: [PATCH] Added ts to TbMsg --- .../thingsboard/server/common/msg/TbMsg.java | 30 +++++++++++-------- common/message/src/main/proto/tbmsg.proto | 1 + .../engine/telemetry/TbMsgTimeseriesNode.java | 2 +- 3 files changed, 20 insertions(+), 13 deletions(-) diff --git a/common/message/src/main/java/org/thingsboard/server/common/msg/TbMsg.java b/common/message/src/main/java/org/thingsboard/server/common/msg/TbMsg.java index 9da7407552..845138d92c 100644 --- a/common/message/src/main/java/org/thingsboard/server/common/msg/TbMsg.java +++ b/common/message/src/main/java/org/thingsboard/server/common/msg/TbMsg.java @@ -40,6 +40,7 @@ import java.util.UUID; public final class TbMsg implements Serializable { private final UUID id; + private final long ts; private final String type; private final EntityId originator; private final TbMsgMetaData metaData; @@ -51,38 +52,43 @@ public final class TbMsg implements Serializable { transient private final TbMsgCallback callback; public static TbMsg newMsg(String type, EntityId originator, TbMsgMetaData metaData, String data) { - return new TbMsg(UUID.randomUUID(), type, originator, metaData.copy(), TbMsgDataType.JSON, data, null, null, TbMsgCallback.EMPTY); + return new TbMsg(UUID.randomUUID(), System.currentTimeMillis(), type, originator, metaData.copy(), TbMsgDataType.JSON, data, null, null, TbMsgCallback.EMPTY); } public static TbMsg newMsg(String type, EntityId originator, TbMsgMetaData metaData, String data, RuleChainId ruleChainId, RuleNodeId ruleNodeId) { - return new TbMsg(UUID.randomUUID(), type, originator, metaData.copy(), TbMsgDataType.JSON, data, ruleChainId, ruleNodeId, TbMsgCallback.EMPTY); + return new TbMsg(UUID.randomUUID(), System.currentTimeMillis(), type, originator, metaData.copy(), TbMsgDataType.JSON, data, ruleChainId, ruleNodeId, TbMsgCallback.EMPTY); } public static TbMsg newMsg(String type, EntityId originator, TbMsgMetaData metaData, TbMsgDataType dataType, String data) { - return new TbMsg(UUID.randomUUID(), type, originator, metaData.copy(), dataType, data, null, null, TbMsgCallback.EMPTY); + return new TbMsg(UUID.randomUUID(), System.currentTimeMillis(), type, originator, metaData.copy(), dataType, data, null, null, TbMsgCallback.EMPTY); } public static TbMsg newMsg(String type, EntityId originator, TbMsgMetaData metaData, TbMsgDataType dataType, String data, RuleChainId ruleChainId, RuleNodeId ruleNodeId) { - return new TbMsg(UUID.randomUUID(), type, originator, metaData.copy(), dataType, data, ruleChainId, ruleNodeId, TbMsgCallback.EMPTY); + return new TbMsg(UUID.randomUUID(), System.currentTimeMillis(), type, originator, metaData.copy(), dataType, data, ruleChainId, ruleNodeId, TbMsgCallback.EMPTY); } public static TbMsg newMsg(String type, EntityId originator, TbMsgMetaData metaData, String data, TbMsgCallback callback) { - return new TbMsg(UUID.randomUUID(), type, originator, metaData.copy(), TbMsgDataType.JSON, data, null, null, callback); + return new TbMsg(UUID.randomUUID(), System.currentTimeMillis(), type, originator, metaData.copy(), TbMsgDataType.JSON, data, null, null, callback); } public static TbMsg transformMsg(TbMsg origMsg, String type, EntityId originator, TbMsgMetaData metaData, String data) { - return new TbMsg(origMsg.getId(), type, originator, metaData.copy(), origMsg.getDataType(), + return new TbMsg(origMsg.getId(), origMsg.getTs(), type, originator, metaData.copy(), origMsg.getDataType(), data, origMsg.getRuleChainId(), origMsg.getRuleNodeId(), origMsg.getCallback()); } public static TbMsg newMsg(TbMsg tbMsg, RuleChainId ruleChainId, RuleNodeId ruleNodeId) { - return new TbMsg(UUID.randomUUID(), tbMsg.getType(), tbMsg.getOriginator(), tbMsg.getMetaData().copy(), + return new TbMsg(UUID.randomUUID(), tbMsg.getTs(), tbMsg.getType(), tbMsg.getOriginator(), tbMsg.getMetaData().copy(), tbMsg.getDataType(), tbMsg.getData(), ruleChainId, ruleNodeId, TbMsgCallback.EMPTY); } - private TbMsg(UUID id, String type, EntityId originator, TbMsgMetaData metaData, TbMsgDataType dataType, String data, + private TbMsg(UUID id, long ts, String type, EntityId originator, TbMsgMetaData metaData, TbMsgDataType dataType, String data, RuleChainId ruleChainId, RuleNodeId ruleNodeId, TbMsgCallback callback) { this.id = id; + if (ts > 0) { + this.ts = ts; + } else { + this.ts = System.currentTimeMillis(); + } this.type = type; this.originator = originator; this.metaData = metaData; @@ -105,6 +111,7 @@ public final class TbMsg implements Serializable { public static byte[] toByteArray(TbMsg msg) { MsgProtos.TbMsgProto.Builder builder = MsgProtos.TbMsgProto.newBuilder(); builder.setId(msg.getId().toString()); + builder.setTs(msg.getTs()); builder.setType(msg.getType()); builder.setEntityType(msg.getOriginator().getEntityType().name()); builder.setEntityIdMSB(msg.getOriginator().getId().getMostSignificantBits()); @@ -124,7 +131,6 @@ public final class TbMsg implements Serializable { builder.setMetaData(MsgProtos.TbMsgMetaDataProto.newBuilder().putAllData(msg.getMetaData().getData()).build()); } - builder.setDataType(msg.getDataType().ordinal()); builder.setData(msg.getData()); return builder.build().toByteArray(); @@ -144,18 +150,18 @@ public final class TbMsg implements Serializable { ruleNodeId = new RuleNodeId(new UUID(proto.getRuleNodeIdMSB(), proto.getRuleNodeIdLSB())); } TbMsgDataType dataType = TbMsgDataType.values()[proto.getDataType()]; - return new TbMsg(UUID.fromString(proto.getId()), proto.getType(), entityId, metaData, dataType, proto.getData(), ruleChainId, ruleNodeId, callback); + return new TbMsg(UUID.fromString(proto.getId()), proto.getTs(), proto.getType(), entityId, metaData, dataType, proto.getData(), ruleChainId, ruleNodeId, callback); } catch (InvalidProtocolBufferException e) { throw new IllegalStateException("Could not parse protobuf for TbMsg", e); } } public TbMsg copyWithRuleChainId(RuleChainId ruleChainId) { - return new TbMsg(this.id, this.type, this.originator, this.metaData, this.dataType, this.data, ruleChainId, null, callback); + return new TbMsg(this.id, this.ts, this.type, this.originator, this.metaData, this.dataType, this.data, ruleChainId, null, callback); } public TbMsg copyWithRuleNodeId(RuleChainId ruleChainId, RuleNodeId ruleNodeId) { - return new TbMsg(this.id, this.type, this.originator, this.metaData, this.dataType, this.data, ruleChainId, ruleNodeId, callback); + return new TbMsg(this.id, this.ts, this.type, this.originator, this.metaData, this.dataType, this.data, ruleChainId, ruleNodeId, callback); } public TbMsgCallback getCallback() { diff --git a/common/message/src/main/proto/tbmsg.proto b/common/message/src/main/proto/tbmsg.proto index 737d5f6f57..25e006121f 100644 --- a/common/message/src/main/proto/tbmsg.proto +++ b/common/message/src/main/proto/tbmsg.proto @@ -44,4 +44,5 @@ message TbMsgProto { int32 dataType = 13; string data = 14; + int64 ts = 15; } \ No newline at end of file diff --git a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/TbMsgTimeseriesNode.java b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/TbMsgTimeseriesNode.java index 14565ac475..912a830dd3 100644 --- a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/TbMsgTimeseriesNode.java +++ b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/TbMsgTimeseriesNode.java @@ -70,7 +70,7 @@ public class TbMsgTimeseriesNode implements TbNode { } catch (NumberFormatException e) { } } else { - ts = System.currentTimeMillis(); + ts = msg.getTs(); } String src = msg.getData(); Map> tsKvMap = JsonConverter.convertToTelemetry(new JsonParser().parse(src), ts);