From a52b88802bf466fd045348067f92ca23d66e4629 Mon Sep 17 00:00:00 2001 From: Volodymyr Babak Date: Thu, 17 Dec 2020 18:57:26 +0200 Subject: [PATCH] Use default rule chain and queue from device profile in edge session --- .../service/edge/rpc/EdgeGrpcSession.java | 28 +++++------ .../DeviceProfileMsgConstructor.java | 15 +++--- .../edge/rpc/processor/BaseProcessor.java | 5 ++ .../rpc/processor/TelemetryProcessor.java | 48 +++++++++++++++++-- 4 files changed, 68 insertions(+), 28 deletions(-) diff --git a/application/src/main/java/org/thingsboard/server/service/edge/rpc/EdgeGrpcSession.java b/application/src/main/java/org/thingsboard/server/service/edge/rpc/EdgeGrpcSession.java index da98fea8c4..9484e91d68 100644 --- a/application/src/main/java/org/thingsboard/server/service/edge/rpc/EdgeGrpcSession.java +++ b/application/src/main/java/org/thingsboard/server/service/edge/rpc/EdgeGrpcSession.java @@ -552,12 +552,6 @@ public final class EdgeGrpcSession implements Closeable { case UPDATED: DeviceProfile deviceProfile = ctx.getDeviceProfileService().findDeviceProfileById(edgeEvent.getTenantId(), deviceProfileId); if (deviceProfile != null) { - - // TODO: voba HACK -// PageData ruleChainsByTenantIdAndEdgeId = ctx.getRuleChainService().findRuleChainsByTenantIdAndEdgeId(edgeEvent.getTenantId(), edgeEvent.getEdgeId(), new TimePageLink(100)); -// RuleChain ruleChain = ruleChainsByTenantIdAndEdgeId.getData().get(0); -// deviceProfile.setDefaultRuleChainId(ruleChain.getId()); - DeviceProfileUpdateMsg deviceProfileUpdateMsg = ctx.getDeviceProfileMsgConstructor().constructDeviceProfileUpdatedMsg(msgType, deviceProfile); downlinkMsg = DownlinkMsg.newBuilder() @@ -916,57 +910,57 @@ public final class EdgeGrpcSession implements Closeable { private ListenableFuture> processUplinkMsg(UplinkMsg uplinkMsg) { List> result = new ArrayList<>(); try { - if (uplinkMsg.getEntityDataList() != null && !uplinkMsg.getEntityDataList().isEmpty()) { + if (uplinkMsg.getEntityDataCount() > 0) { for (EntityDataProto entityData : uplinkMsg.getEntityDataList()) { result.addAll(ctx.getTelemetryProcessor().onTelemetryUpdate(edge.getTenantId(), entityData)); } } - if (uplinkMsg.getDeviceUpdateMsgList() != null && !uplinkMsg.getDeviceUpdateMsgList().isEmpty()) { + if (uplinkMsg.getDeviceUpdateMsgCount() > 0) { for (DeviceUpdateMsg deviceUpdateMsg : uplinkMsg.getDeviceUpdateMsgList()) { result.add(ctx.getDeviceProcessor().onDeviceUpdate(edge.getTenantId(), edge, deviceUpdateMsg)); } } - if (uplinkMsg.getDeviceCredentialsUpdateMsgList() != null && !uplinkMsg.getDeviceCredentialsUpdateMsgList().isEmpty()) { + if (uplinkMsg.getDeviceCredentialsUpdateMsgCount() > 0) { for (DeviceCredentialsUpdateMsg deviceCredentialsUpdateMsg : uplinkMsg.getDeviceCredentialsUpdateMsgList()) { result.add(ctx.getDeviceProcessor().onDeviceCredentialsUpdate(edge.getTenantId(), deviceCredentialsUpdateMsg)); } } - if (uplinkMsg.getAlarmUpdateMsgList() != null && !uplinkMsg.getAlarmUpdateMsgList().isEmpty()) { + if (uplinkMsg.getAlarmUpdateMsgCount() > 0) { for (AlarmUpdateMsg alarmUpdateMsg : uplinkMsg.getAlarmUpdateMsgList()) { result.add(ctx.getAlarmProcessor().onAlarmUpdate(edge.getTenantId(), alarmUpdateMsg)); } } - if (uplinkMsg.getRelationUpdateMsgList() != null && !uplinkMsg.getRelationUpdateMsgList().isEmpty()) { + if (uplinkMsg.getRelationUpdateMsgCount() > 0) { for (RelationUpdateMsg relationUpdateMsg : uplinkMsg.getRelationUpdateMsgList()) { result.add(ctx.getRelationProcessor().onRelationUpdate(edge.getTenantId(), relationUpdateMsg)); } } - if (uplinkMsg.getRuleChainMetadataRequestMsgList() != null && !uplinkMsg.getRuleChainMetadataRequestMsgList().isEmpty()) { + if (uplinkMsg.getRuleChainMetadataRequestMsgCount() > 0) { for (RuleChainMetadataRequestMsg ruleChainMetadataRequestMsg : uplinkMsg.getRuleChainMetadataRequestMsgList()) { result.add(ctx.getSyncEdgeService().processRuleChainMetadataRequestMsg(edge, ruleChainMetadataRequestMsg)); } } - if (uplinkMsg.getAttributesRequestMsgList() != null && !uplinkMsg.getAttributesRequestMsgList().isEmpty()) { + if (uplinkMsg.getAttributesRequestMsgCount() > 0) { for (AttributesRequestMsg attributesRequestMsg : uplinkMsg.getAttributesRequestMsgList()) { result.add(ctx.getSyncEdgeService().processAttributesRequestMsg(edge, attributesRequestMsg)); } } - if (uplinkMsg.getRelationRequestMsgList() != null && !uplinkMsg.getRelationRequestMsgList().isEmpty()) { + if (uplinkMsg.getRelationRequestMsgCount() > 0) { for (RelationRequestMsg relationRequestMsg : uplinkMsg.getRelationRequestMsgList()) { result.add(ctx.getSyncEdgeService().processRelationRequestMsg(edge, relationRequestMsg)); } } - if (uplinkMsg.getUserCredentialsRequestMsgList() != null && !uplinkMsg.getUserCredentialsRequestMsgList().isEmpty()) { + if (uplinkMsg.getUserCredentialsRequestMsgCount() > 0) { for (UserCredentialsRequestMsg userCredentialsRequestMsg : uplinkMsg.getUserCredentialsRequestMsgList()) { result.add(ctx.getSyncEdgeService().processUserCredentialsRequestMsg(edge, userCredentialsRequestMsg)); } } - if (uplinkMsg.getDeviceCredentialsRequestMsgList() != null && !uplinkMsg.getDeviceCredentialsRequestMsgList().isEmpty()) { + if (uplinkMsg.getDeviceCredentialsRequestMsgCount() > 0) { for (DeviceCredentialsRequestMsg deviceCredentialsRequestMsg : uplinkMsg.getDeviceCredentialsRequestMsgList()) { result.add(ctx.getSyncEdgeService().processDeviceCredentialsRequestMsg(edge, deviceCredentialsRequestMsg)); } } - if (uplinkMsg.getDeviceRpcCallMsgList() != null && !uplinkMsg.getDeviceRpcCallMsgList().isEmpty()) { + if (uplinkMsg.getDeviceRpcCallMsgCount() > 0) { for (DeviceRpcCallMsg deviceRpcCallMsg : uplinkMsg.getDeviceRpcCallMsgList()) { result.add(ctx.getDeviceProcessor().processDeviceRpcCallResponseMsg(edge.getTenantId(), deviceRpcCallMsg)); } diff --git a/application/src/main/java/org/thingsboard/server/service/edge/rpc/constructor/DeviceProfileMsgConstructor.java b/application/src/main/java/org/thingsboard/server/service/edge/rpc/constructor/DeviceProfileMsgConstructor.java index 603fb37b68..b56e1447cf 100644 --- a/application/src/main/java/org/thingsboard/server/service/edge/rpc/constructor/DeviceProfileMsgConstructor.java +++ b/application/src/main/java/org/thingsboard/server/service/edge/rpc/constructor/DeviceProfileMsgConstructor.java @@ -44,13 +44,14 @@ public class DeviceProfileMsgConstructor { .setTransportType(deviceProfile.getTransportType().name()) .setProvisionType(deviceProfile.getProvisionType().name()) .setProfileDataBytes(ByteString.copyFrom(dataDecodingEncodingService.encode(deviceProfile.getProfileData()))); - if (deviceProfile.getDefaultRuleChainId() != null) { - builder.setDefaultRuleChainIdMSB(deviceProfile.getDefaultRuleChainId().getId().getMostSignificantBits()) - .setDefaultRuleChainIdLSB(deviceProfile.getDefaultRuleChainId().getId().getLeastSignificantBits()); - } - if (deviceProfile.getDefaultQueueName() != null) { - builder.setDefaultQueueName(deviceProfile.getDefaultQueueName()); - } + // TODO: voba - should this be always null at the moment?? +// if (deviceProfile.getDefaultRuleChainId() != null) { +// builder.setDefaultRuleChainIdMSB(deviceProfile.getDefaultRuleChainId().getId().getMostSignificantBits()) +// .setDefaultRuleChainIdLSB(deviceProfile.getDefaultRuleChainId().getId().getLeastSignificantBits()); +// } +// if (deviceProfile.getDefaultQueueName() != null) { +// builder.setDefaultQueueName(deviceProfile.getDefaultQueueName()); +// } if (deviceProfile.getProvisionDeviceKey() != null) { builder.setProvisionDeviceKey(deviceProfile.getProvisionDeviceKey()); } diff --git a/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/BaseProcessor.java b/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/BaseProcessor.java index d1236f610d..1a00e77388 100644 --- a/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/BaseProcessor.java +++ b/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/BaseProcessor.java @@ -42,6 +42,8 @@ import org.thingsboard.server.dao.entityview.EntityViewService; import org.thingsboard.server.dao.relation.RelationService; import org.thingsboard.server.dao.user.UserService; import org.thingsboard.server.service.executors.DbCallbackExecutorService; +import org.thingsboard.server.service.profile.DefaultTbDeviceProfileCache; +import org.thingsboard.server.service.profile.TbDeviceProfileCache; import org.thingsboard.server.service.queue.TbClusterService; import org.thingsboard.server.service.state.DeviceStateService; @@ -56,6 +58,9 @@ public abstract class BaseProcessor { @Autowired protected DeviceService deviceService; + @Autowired + protected TbDeviceProfileCache deviceProfileCache; + @Autowired protected DashboardService dashboardService; 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 7fa76d7d67..fc32c089d1 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 @@ -21,11 +21,15 @@ import com.google.common.util.concurrent.ListenableFuture; import com.google.common.util.concurrent.SettableFuture; import com.google.gson.Gson; import com.google.gson.JsonObject; +import groovy.lang.Tuple; import lombok.extern.slf4j.Slf4j; +import org.javatuples.Pair; +import org.passay.Rule; import org.springframework.stereotype.Component; import org.thingsboard.rule.engine.api.msg.DeviceAttributesEventNotificationMsg; import org.thingsboard.server.common.data.DataConstants; import org.thingsboard.server.common.data.Device; +import org.thingsboard.server.common.data.DeviceProfile; import org.thingsboard.server.common.data.EntityType; import org.thingsboard.server.common.data.EntityView; import org.thingsboard.server.common.data.asset.Asset; @@ -33,14 +37,18 @@ 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.DeviceProfileId; 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.data.rule.RuleChain; import org.thingsboard.server.common.msg.TbMsg; import org.thingsboard.server.common.msg.TbMsgMetaData; +import org.thingsboard.server.common.msg.queue.ServiceQueue; import org.thingsboard.server.common.msg.session.SessionMsgType; import org.thingsboard.server.common.transport.adaptor.JsonConverter; import org.thingsboard.server.common.transport.util.JsonUtils; @@ -70,7 +78,9 @@ public class TelemetryProcessor extends BaseProcessor { List> result = new ArrayList<>(); EntityId entityId = constructEntityId(entityData); if ((entityData.hasPostAttributesMsg() || entityData.hasPostTelemetryMsg() || entityData.hasAttributesUpdatedMsg()) && entityId != null) { - TbMsgMetaData metaData = constructBaseMsgMetadata(tenantId, entityId); + // TODO: voba - in terms of performance we should not fetch device from DB by id + // TbMsgMetaData metaData = constructBaseMsgMetadata(tenantId, entityId); + TbMsgMetaData metaData = new TbMsgMetaData(); metaData.putValue(DataConstants.MSG_SOURCE_KEY, DataConstants.EDGE_MSG_SOURCE); if (entityData.hasPostAttributesMsg()) { result.add(processPostAttributes(tenantId, entityId, entityData.getPostAttributesMsg(), metaData)); @@ -120,12 +130,36 @@ public class TelemetryProcessor extends BaseProcessor { return metaData; } + private Pair getDefaultQueueNameAndRuleChainId(TenantId tenantId, EntityId entityId) { + if (EntityType.DEVICE.equals(entityId.getEntityType())) { + DeviceProfile deviceProfile = deviceProfileCache.get(tenantId, new DeviceId(entityId.getId())); + RuleChainId ruleChainId; + String queueName; + + if (deviceProfile == null) { + log.warn("[{}] Device profile is null!", entityId); + ruleChainId = null; + queueName = ServiceQueue.MAIN; + } else { + ruleChainId = deviceProfile.getDefaultRuleChainId(); + String defaultQueueName = deviceProfile.getDefaultQueueName(); + queueName = defaultQueueName != null ? defaultQueueName : ServiceQueue.MAIN; + } + return new Pair<>(queueName, ruleChainId); + } else { + return new Pair<>(ServiceQueue.MAIN, null); + } + } + private ListenableFuture processPostTelemetry(TenantId tenantId, EntityId entityId, TransportProtos.PostTelemetryMsg msg, TbMsgMetaData metaData) { SettableFuture futureToSet = SettableFuture.create(); for (TransportProtos.TsKvListProto tsKv : msg.getTsKvListList()) { JsonObject json = JsonUtils.getJsonObject(tsKv.getKvList()); metaData.putValue("ts", tsKv.getTs() + ""); - TbMsg tbMsg = TbMsg.newMsg(SessionMsgType.POST_TELEMETRY_REQUEST.name(), entityId, metaData, gson.toJson(json)); + Pair defaultQueueAndRuleChain = getDefaultQueueNameAndRuleChainId(tenantId, entityId); + String queueName = defaultQueueAndRuleChain.getValue0(); + RuleChainId ruleChainId = defaultQueueAndRuleChain.getValue1(); + TbMsg tbMsg = TbMsg.newMsg(queueName, SessionMsgType.POST_TELEMETRY_REQUEST.name(), entityId, metaData, gson.toJson(json), ruleChainId, null); tbClusterService.pushMsgToRuleEngine(tenantId, tbMsg.getOriginator(), tbMsg, new TbQueueCallback() { @Override public void onSuccess(TbQueueMsgMetadata metadata) { @@ -145,7 +179,10 @@ 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)); + Pair defaultQueueAndRuleChain = getDefaultQueueNameAndRuleChainId(tenantId, entityId); + String queueName = defaultQueueAndRuleChain.getValue0(); + RuleChainId ruleChainId = defaultQueueAndRuleChain.getValue1(); + TbMsg tbMsg = TbMsg.newMsg(queueName, SessionMsgType.POST_ATTRIBUTES_REQUEST.name(), entityId, metaData, gson.toJson(json), ruleChainId, null); tbClusterService.pushMsgToRuleEngine(tenantId, tbMsg.getOriginator(), tbMsg, new TbQueueCallback() { @Override public void onSuccess(TbQueueMsgMetadata metadata) { @@ -169,7 +206,10 @@ public class TelemetryProcessor extends BaseProcessor { Futures.addCallback(future, new FutureCallback>() { @Override public void onSuccess(@Nullable List voids) { - TbMsg tbMsg = TbMsg.newMsg(DataConstants.ATTRIBUTES_UPDATED, entityId, metaData, gson.toJson(json)); + Pair defaultQueueAndRuleChain = getDefaultQueueNameAndRuleChainId(tenantId, entityId); + String queueName = defaultQueueAndRuleChain.getValue0(); + RuleChainId ruleChainId = defaultQueueAndRuleChain.getValue1(); + TbMsg tbMsg = TbMsg.newMsg(queueName, DataConstants.ATTRIBUTES_UPDATED, entityId, metaData, gson.toJson(json), ruleChainId, null); tbClusterService.pushMsgToRuleEngine(tenantId, tbMsg.getOriginator(), tbMsg, new TbQueueCallback() { @Override public void onSuccess(TbQueueMsgMetadata metadata) {