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

Refactoring for TbMsgProto usage

pull/13323/head
ViacheslavKlimov 1 year ago
parent
0adfaf2b01
commit
1d83a2c9d3
7 changed files with 55 additions and 49 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. 8
      application/src/main/java/org/thingsboard/server/service/queue/processing/SequentialByOriginatorIdTbRuleEngineSubmitStrategy.java
  2. 8
      application/src/main/java/org/thingsboard/server/service/queue/processing/TbRuleEngineProcessingStrategyFactory.java
  3. 15
      application/src/main/java/org/thingsboard/server/service/queue/ruleengine/TbRuleEngineQueueConsumerManager.java
  4. 45
      common/message/src/main/java/org/thingsboard/server/common/msg/TbMsg.java
  5. 4
      common/message/src/main/proto/tbmsg.proto
  6. 22
      common/proto/src/main/java/org/thingsboard/server/common/util/ProtoUtils.java
  7. 2
      common/proto/src/main/proto/queue.proto

8
application/src/main/java/org/thingsboard/server/service/queue/processing/SequentialByOriginatorIdTbRuleEngineSubmitStrategy.java
View File

@ -20,6 +20,7 @@ import lombok.extern.slf4j.Slf4j;
import org.thingsboard.server.common.data.id.EntityId;
import org.thingsboard.server.common.data.id.EntityIdFactory;
import org.thingsboard.server.common.msg.gen.MsgProtos;
import org.thingsboard.server.common.util.ProtoUtils;
import org.thingsboard.server.gen.transport.TransportProtos;
import java.util.UUID;
@ -34,12 +35,7 @@ public class SequentialByOriginatorIdTbRuleEngineSubmitStrategy extends Sequenti
@Override
protected EntityId getEntityId(TransportProtos.ToRuleEngineMsg msg) {
try {
MsgProtos.TbMsgProto proto;
if (msg.getTbMsg().isEmpty()) {
proto = msg.getTbMsgProto();
} else {
proto = MsgProtos.TbMsgProto.parseFrom(msg.getTbMsg());
}
MsgProtos.TbMsgProto proto = ProtoUtils.getTbMsgProto(msg);
return EntityIdFactory.getByTypeAndUuid(proto.getEntityType(), new UUID(proto.getEntityIdMSB(), proto.getEntityIdLSB()));
} catch (InvalidProtocolBufferException e) {
log.warn("[{}] Failed to parse TbMsg: {}", queueName, msg);

8
application/src/main/java/org/thingsboard/server/service/queue/processing/TbRuleEngineProcessingStrategyFactory.java
View File

@ -19,8 +19,8 @@ import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;
import org.springframework.util.CollectionUtils;
import org.thingsboard.server.common.data.queue.ProcessingStrategy;
import org.thingsboard.server.common.msg.TbMsg;
import org.thingsboard.server.common.msg.queue.TbMsgCallback;
import org.thingsboard.server.common.util.ProtoUtils;
import org.thingsboard.server.gen.transport.TransportProtos;
import org.thingsboard.server.queue.common.TbProtoQueueMsg;
@ -125,7 +125,7 @@ public class TbRuleEngineProcessingStrategyFactory {
}
log.debug("[{}] Going to reprocess {} messages", queueName, toReprocess.size());
if (log.isTraceEnabled()) {
toReprocess.forEach((id, msg) -> log.trace("Going to reprocess [{}]: {}", id, TbMsg.fromProto(result.getQueueName(), msg.getValue().getTbMsgProto(), msg.getValue().getTbMsg(), TbMsgCallback.EMPTY)));
toReprocess.forEach((id, msg) -> log.trace("Going to reprocess [{}]: {}", id, ProtoUtils.fromTbMsgProto(result.getQueueName(), msg.getValue(), TbMsgCallback.EMPTY)));
}
if (pauseBetweenRetries > 0) {
try {
@ -164,10 +164,10 @@ public class TbRuleEngineProcessingStrategyFactory {
log.debug("[{}] Reprocessing skipped for {} failed and {} timeout messages", queueName, result.getFailedMap().size(), result.getPendingMap().size());
}
if (log.isTraceEnabled()) {
result.getFailedMap().forEach((id, msg) -> log.trace("Failed messages [{}]: {}", id, TbMsg.fromProto(result.getQueueName(), msg.getValue().getTbMsgProto(), msg.getValue().getTbMsg(), TbMsgCallback.EMPTY)));
result.getFailedMap().forEach((id, msg) -> log.trace("Failed messages [{}]: {}", id, ProtoUtils.fromTbMsgProto(result.getQueueName(), msg.getValue(), TbMsgCallback.EMPTY)));
}
if (log.isTraceEnabled()) {
result.getPendingMap().forEach((id, msg) -> log.trace("Timeout messages [{}]: {}", id, TbMsg.fromProto(result.getQueueName(), msg.getValue().getTbMsgProto(), msg.getValue().getTbMsg(), TbMsgCallback.EMPTY)));
result.getPendingMap().forEach((id, msg) -> log.trace("Timeout messages [{}]: {}", id, ProtoUtils.fromTbMsgProto(result.getQueueName(), msg.getValue(), TbMsgCallback.EMPTY)));
}
return new TbRuleEngineProcessingDecision(true, null);
}

15
application/src/main/java/org/thingsboard/server/service/queue/ruleengine/TbRuleEngineQueueConsumerManager.java
View File

@ -30,6 +30,7 @@ import org.thingsboard.server.common.msg.queue.RuleNodeInfo;
import org.thingsboard.server.common.msg.queue.ServiceType;
import org.thingsboard.server.common.msg.queue.TbMsgCallback;
import org.thingsboard.server.common.msg.queue.TopicPartitionInfo;
import org.thingsboard.server.common.util.ProtoUtils;
import org.thingsboard.server.gen.transport.TransportProtos.ToRuleEngineMsg;
import org.thingsboard.server.queue.TbQueueConsumer;
import org.thingsboard.server.queue.common.TbProtoQueueMsg;
@ -178,7 +179,7 @@ public class TbRuleEngineQueueConsumerManager extends MainQueueConsumerManager<T
new TbMsgPackCallback(id, tenantId, packCtx, stats.getTimer(tenantId, SUCCESSFUL_STATUS), stats.getTimer(tenantId, FAILED_STATUS)) :
new TbMsgPackCallback(id, tenantId, packCtx);
try {
if (!toRuleEngineMsg.getTbMsg().isEmpty() || toRuleEngineMsg.getTbMsgProto().isInitialized()) {
if (!toRuleEngineMsg.getTbMsg().isEmpty() || toRuleEngineMsg.hasTbMsgProto()) {
forwardToRuleEngineActor(config.getName(), tenantId, toRuleEngineMsg, callback);
} else {
callback.onSuccess();
@ -189,7 +190,7 @@ public class TbRuleEngineQueueConsumerManager extends MainQueueConsumerManager<T
}
private void forwardToRuleEngineActor(String queueName, TenantId tenantId, ToRuleEngineMsg toRuleEngineMsg, TbMsgCallback callback) {
TbMsg tbMsg = TbMsg.fromProto(queueName, toRuleEngineMsg.getTbMsgProto(), toRuleEngineMsg.getTbMsg(), callback);
TbMsg tbMsg = ProtoUtils.fromTbMsgProto(queueName, toRuleEngineMsg, callback);
QueueToRuleEngineMsg msg;
ProtocolStringList relationTypesList = toRuleEngineMsg.getRelationTypesList();
Set<String> relationTypes;
@ -207,7 +208,7 @@ public class TbRuleEngineQueueConsumerManager extends MainQueueConsumerManager<T
log.info("[{}] {} to process [{}] messages", queueKey, prefix, map.size());
for (Map.Entry<UUID, TbProtoQueueMsg<ToRuleEngineMsg>> pending : map.entrySet()) {
ToRuleEngineMsg tmp = pending.getValue().getValue();
TbMsg tmpMsg = TbMsg.fromProto(config.getName(), tmp.getTbMsgProto(), tmp.getTbMsg(), TbMsgCallback.EMPTY);
TbMsg tmpMsg = ProtoUtils.fromTbMsgProto(config.getName(), tmp, TbMsgCallback.EMPTY);
RuleNodeInfo ruleNodeInfo = ctx.getLastVisitedRuleNode(pending.getKey());
if (printAll) {
log.trace("[{}][{}] {} to process message: {}, Last Rule Node: {}", queueKey, TenantId.fromUUID(new UUID(tmp.getTenantIdMSB(), tmp.getTenantIdLSB())), prefix, tmpMsg, ruleNodeInfo);
@ -236,13 +237,7 @@ public class TbRuleEngineQueueConsumerManager extends MainQueueConsumerManager<T
}
for (TbProtoQueueMsg<ToRuleEngineMsg> msg : msgs) {
try {
MsgProtos.TbMsgProto tbMsgProto;
if (msg.getValue().getTbMsg().isEmpty()) {
tbMsgProto = msg.getValue().getTbMsgProto();
} else {
tbMsgProto = MsgProtos.TbMsgProto.parseFrom(msg.getValue().getTbMsg());
}
MsgProtos.TbMsgProto tbMsgProto = ProtoUtils.getTbMsgProto(msg.getValue());
EntityId originator = EntityIdFactory.getByTypeAndUuid(tbMsgProto.getEntityType(), new UUID(tbMsgProto.getEntityIdMSB(), tbMsgProto.getEntityIdLSB()));
TopicPartitionInfo tpi = ctx.getPartitionService().resolve(ServiceType.TB_RULE_ENGINE, config.getName(), TenantId.SYS_TENANT_ID, originator);

45
common/message/src/main/java/org/thingsboard/server/common/msg/TbMsg.java
View File

@ -32,6 +32,7 @@ import org.thingsboard.server.common.data.id.RuleChainId;
import org.thingsboard.server.common.data.id.RuleNodeId;
import org.thingsboard.server.common.data.msg.TbMsgType;
import org.thingsboard.server.common.msg.gen.MsgProtos;
import org.thingsboard.server.common.msg.gen.MsgProtos.TbMsgProto;
import org.thingsboard.server.common.msg.queue.TbMsgCallback;
import java.io.Serializable;
@ -151,8 +152,8 @@ public final class TbMsg implements Serializable {
this.callback = Objects.requireNonNullElse(callback, TbMsgCallback.EMPTY);
}
public static MsgProtos.TbMsgProto toProto(TbMsg msg) {
MsgProtos.TbMsgProto.Builder builder = MsgProtos.TbMsgProto.newBuilder();
public static TbMsgProto toProto(TbMsg msg) {
TbMsgProto.Builder builder = TbMsgProto.newBuilder();
builder.setId(msg.getId().toString());
builder.setTs(msg.getTs());
builder.setType(msg.getType());
@ -204,12 +205,11 @@ public final class TbMsg implements Serializable {
return builder.build();
}
//TODO: added for processing old messages from queue, should be removed after release
@Deprecated(forRemoval = true)
public static TbMsg fromProto(String queueName, MsgProtos.TbMsgProto proto, ByteString data, TbMsgCallback callback) {
@Deprecated(forRemoval = true, since = "4.1") // to be removed in 4.2
public static TbMsg fromProto(String queueName, TbMsgProto proto, ByteString data, TbMsgCallback callback) {
try {
if (!data.isEmpty()) {
proto = MsgProtos.TbMsgProto.parseFrom(data);
proto = TbMsgProto.parseFrom(data);
}
} catch (InvalidProtocolBufferException e) {
throw new IllegalStateException("Could not parse protobuf for TbMsg", e);
@ -217,7 +217,7 @@ public final class TbMsg implements Serializable {
return fromProto(queueName, proto, callback);
}
public static TbMsg fromProto(String queueName, MsgProtos.TbMsgProto proto, TbMsgCallback callback) {
public static TbMsg fromProto(String queueName, TbMsgProto proto, TbMsgCallback callback) {
TbMsgMetaData metaData = new TbMsgMetaData(proto.getMetaData().getDataMap());
EntityId entityId = EntityIdFactory.getByTypeAndUuid(proto.getEntityType(), new UUID(proto.getEntityIdMSB(), proto.getEntityIdLSB()));
CustomerId customerId = null;
@ -225,7 +225,8 @@ public final class TbMsg implements Serializable {
RuleNodeId ruleNodeId = null;
UUID correlationId = null;
Integer partition = null;
List<CalculatedFieldId> calculatedFieldIds = new CopyOnWriteArrayList<>();if (proto.getCustomerIdMSB() != 0L && proto.getCustomerIdLSB() != 0L) {
List<CalculatedFieldId> calculatedFieldIds = new CopyOnWriteArrayList<>();
if (proto.getCustomerIdMSB() != 0L && proto.getCustomerIdLSB() != 0L) {
customerId = new CustomerId(new UUID(proto.getCustomerIdMSB(), proto.getCustomerIdLSB()));
}
if (proto.getRuleChainIdMSB() != 0L && proto.getRuleChainIdLSB() != 0L) {
@ -240,19 +241,13 @@ public final class TbMsg implements Serializable {
}
for (MsgProtos.CalculatedFieldIdProto cfIdProto : proto.getCalculatedFieldsList()) {
CalculatedFieldId calculatedFieldId = new CalculatedFieldId(new UUID(
cfIdProto.getCalculatedFieldIdMSB(),
cfIdProto.getCalculatedFieldIdLSB()
));
calculatedFieldIds.add(calculatedFieldId);
}TbMsgProcessingCtx ctx;
if (proto.hasCtx()) {
ctx = TbMsgProcessingCtx.fromProto(proto.getCtx());
} else {
// Backward compatibility with unprocessed messages fetched from queue after update.
ctx = new TbMsgProcessingCtx(proto.getRuleNodeExecCounter());
CalculatedFieldId calculatedFieldId = new CalculatedFieldId(new UUID(
cfIdProto.getCalculatedFieldIdMSB(),
cfIdProto.getCalculatedFieldIdLSB()
));
calculatedFieldIds.add(calculatedFieldId);
}
TbMsgProcessingCtx ctx = TbMsgProcessingCtx.fromProto(proto.getCtx());
TbMsgDataType dataType = TbMsgDataType.values()[proto.getDataType()];
return new TbMsg(queueName, UUID.fromString(proto.getId()), proto.getTs(), null, proto.getType(), entityId, customerId,
metaData, dataType, proto.getData(), ruleChainId, ruleNodeId, correlationId, partition, calculatedFieldIds, ctx, callback);
@ -505,11 +500,11 @@ public final class TbMsg implements Serializable {
public String toString() {
return "TbMsg.TbMsgBuilder(queueName=" + this.queueName + ", id=" + this.id + ", ts=" + this.ts +
", type=" + this.type + ", internalType=" + this.internalType + ", originator=" + this.originator +
", customerId=" + this.customerId + ", metaData=" + this.metaData + ", dataType=" + this.dataType +
", data=" + this.data + ", ruleChainId=" + this.ruleChainId + ", ruleNodeId=" + this.ruleNodeId +
", correlationId=" + this.correlationId + ", partition=" + this.partition + ", previousCalculatedFields=" + this.previousCalculatedFieldIds +
", ctx=" + this.ctx + ", callback=" + this.callback + ")";
", type=" + this.type + ", internalType=" + this.internalType + ", originator=" + this.originator +
", customerId=" + this.customerId + ", metaData=" + this.metaData + ", dataType=" + this.dataType +
", data=" + this.data + ", ruleChainId=" + this.ruleChainId + ", ruleNodeId=" + this.ruleNodeId +
", correlationId=" + this.correlationId + ", partition=" + this.partition + ", previousCalculatedFields=" + this.previousCalculatedFieldIds +
", ctx=" + this.ctx + ", callback=" + this.callback + ")";
}
}

4
common/message/src/main/proto/tbmsg.proto
View File

@ -59,8 +59,8 @@ message TbMsgProto {
string data = 14;
int64 ts = 15;
// Will be removed in 3.4. Moved to processing context
int32 ruleNodeExecCounter = 16;
// ruleNodeExecCounter (16) was removed in 4.1
int64 customerIdMSB = 17;
int64 customerIdLSB = 18;

22
common/proto/src/main/java/org/thingsboard/server/common/util/ProtoUtils.java
View File

@ -18,6 +18,8 @@ package org.thingsboard.server.common.util;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.JsonNode;
import com.google.protobuf.ByteString;
import com.google.protobuf.InvalidProtocolBufferException;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.thingsboard.common.util.JacksonUtil;
import org.thingsboard.server.common.data.ApiUsageRecordKey;
@ -76,12 +78,15 @@ import org.thingsboard.server.common.data.security.DeviceCredentials;
import org.thingsboard.server.common.data.security.DeviceCredentialsType;
import org.thingsboard.server.common.data.sync.vc.RepositoryAuthMethod;
import org.thingsboard.server.common.data.sync.vc.RepositorySettings;
import org.thingsboard.server.common.msg.TbMsg;
import org.thingsboard.server.common.msg.ToDeviceActorNotificationMsg;
import org.thingsboard.server.common.msg.edge.EdgeEventUpdateMsg;
import org.thingsboard.server.common.msg.edge.EdgeHighPriorityMsg;
import org.thingsboard.server.common.msg.edge.FromEdgeSyncResponse;
import org.thingsboard.server.common.msg.edge.ToEdgeSyncRequest;
import org.thingsboard.server.common.msg.gen.MsgProtos;
import org.thingsboard.server.common.msg.plugin.ComponentLifecycleMsg;
import org.thingsboard.server.common.msg.queue.TbMsgCallback;
import org.thingsboard.server.common.msg.rpc.FromDeviceRpcResponse;
import org.thingsboard.server.common.msg.rpc.FromDeviceRpcResponseActorMsg;
import org.thingsboard.server.common.msg.rpc.RemoveRpcActorMsg;
@ -93,8 +98,8 @@ import org.thingsboard.server.common.msg.rule.engine.DeviceDeleteMsg;
import org.thingsboard.server.common.msg.rule.engine.DeviceEdgeUpdateMsg;
import org.thingsboard.server.common.msg.rule.engine.DeviceNameOrTypeUpdateMsg;
import org.thingsboard.server.gen.transport.TransportProtos;
import org.thingsboard.server.gen.transport.TransportProtos.KeyValueProto;
import org.thingsboard.server.gen.transport.TransportProtos.ApiUsageRecordKeyProto;
import org.thingsboard.server.gen.transport.TransportProtos.KeyValueProto;
import java.util.ArrayList;
import java.util.Arrays;
@ -1344,6 +1349,21 @@ public class ProtoUtils {
return builder.build();
}
@Deprecated(forRemoval = true, since = "4.1")
public static MsgProtos.TbMsgProto getTbMsgProto(TransportProtos.ToRuleEngineMsg ruleEngineMsg) throws InvalidProtocolBufferException {
if (ruleEngineMsg.getTbMsg().isEmpty()) {
return ruleEngineMsg.getTbMsgProto();
} else {
return MsgProtos.TbMsgProto.parseFrom(ruleEngineMsg.getTbMsg());
}
}
@SneakyThrows
@Deprecated(forRemoval = true, since = "4.1") // inline to TbMsg.fromProto(queueName, ruleEngineMsg.getTbMsgProto(), callback)
public static TbMsg fromTbMsgProto(String queueName, TransportProtos.ToRuleEngineMsg ruleEngineMsg, TbMsgCallback callback) {
return TbMsg.fromProto(queueName, getTbMsgProto(ruleEngineMsg), callback);
}
private static boolean isNotNull(Object obj) {
return obj != null;
}

2
common/proto/src/main/proto/queue.proto
View File

@ -1703,7 +1703,7 @@ message ToCalculatedFieldNotificationMsg {
message ToRuleEngineMsg {
int64 tenantIdMSB = 1;
int64 tenantIdLSB = 2;
bytes tbMsg = 3 [deprecated = true];
bytes tbMsg = 3 [deprecated = true]; // for removal in 4.2
repeated string relationTypes = 4;
string failureMessage = 5;
msgqueue.TbMsgProto tbMsgProto = 6;

Write Preview
Loading…
Cancel
Save