From bf5f90d2712b2d01553a3829150abfb2079cd3cb Mon Sep 17 00:00:00 2001 From: Volodymyr Babak Date: Wed, 1 Jul 2020 09:27:16 +0300 Subject: [PATCH] Added support for user update. Push notification to rule engine in case device was created --- .../edge/DefaultEdgeNotificationService.java | 18 +++++ .../service/edge/EdgeContextComponent.java | 11 +++ .../service/edge/rpc/EdgeGrpcSession.java | 67 +++++++++++++++++-- 3 files changed, 89 insertions(+), 7 deletions(-) diff --git a/application/src/main/java/org/thingsboard/server/service/edge/DefaultEdgeNotificationService.java b/application/src/main/java/org/thingsboard/server/service/edge/DefaultEdgeNotificationService.java index d16082173c..eb5fcdad9e 100644 --- a/application/src/main/java/org/thingsboard/server/service/edge/DefaultEdgeNotificationService.java +++ b/application/src/main/java/org/thingsboard/server/service/edge/DefaultEdgeNotificationService.java @@ -23,12 +23,14 @@ import lombok.extern.slf4j.Slf4j; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.stereotype.Service; import org.thingsboard.server.common.data.EntityType; +import org.thingsboard.server.common.data.User; import org.thingsboard.server.common.data.alarm.Alarm; import org.thingsboard.server.common.data.audit.ActionType; import org.thingsboard.server.common.data.edge.Edge; import org.thingsboard.server.common.data.edge.EdgeEvent; import org.thingsboard.server.common.data.edge.EdgeEventType; import org.thingsboard.server.common.data.id.AlarmId; +import org.thingsboard.server.common.data.id.CustomerId; import org.thingsboard.server.common.data.id.DashboardId; import org.thingsboard.server.common.data.id.EdgeId; import org.thingsboard.server.common.data.id.EntityId; @@ -36,6 +38,7 @@ import org.thingsboard.server.common.data.id.EntityIdFactory; import org.thingsboard.server.common.data.id.IdBased; 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.page.TextPageData; import org.thingsboard.server.common.data.page.TextPageLink; import org.thingsboard.server.common.data.page.TimePageData; @@ -46,7 +49,9 @@ import org.thingsboard.server.common.msg.queue.TbCallback; import org.thingsboard.server.dao.alarm.AlarmService; import org.thingsboard.server.dao.edge.EdgeEventService; import org.thingsboard.server.dao.edge.EdgeService; +import org.thingsboard.server.dao.model.ModelConstants; import org.thingsboard.server.dao.relation.RelationService; +import org.thingsboard.server.dao.user.UserService; import org.thingsboard.server.gen.transport.TransportProtos; import org.thingsboard.server.queue.util.TbCoreComponent; import org.thingsboard.server.service.executors.DbCallbackExecutorService; @@ -77,6 +82,9 @@ public class DefaultEdgeNotificationService implements EdgeNotificationService { @Autowired private AlarmService alarmService; + @Autowired + private UserService userService; + @Autowired private RelationService relationService; @@ -142,6 +150,7 @@ public class DefaultEdgeNotificationService implements EdgeNotificationService { switch (edgeEventType) { // TODO: voba - handle edge updates // case EDGE: + case USER: case ASSET: case DEVICE: case ENTITY_VIEW: @@ -286,6 +295,15 @@ public class DefaultEdgeNotificationService implements EdgeNotificationService { return convertToEdgeIds(edgeService.findEdgesByTenantIdAndDashboardId(tenantId, new DashboardId(entityId.getId()))); case RULE_CHAIN: return convertToEdgeIds(edgeService.findEdgesByTenantIdAndRuleChainId(tenantId, new RuleChainId(entityId.getId()))); + case USER: + User userById = userService.findUserById(tenantId, new UserId(entityId.getId())); + TextPageData edges; + if (userById.getCustomerId() == null || userById.getCustomerId().getId().equals(ModelConstants.NULL_UUID)) { + edges = edgeService.findEdgesByTenantId(tenantId, new TextPageLink(Integer.MAX_VALUE)); + } else { + edges = edgeService.findEdgesByTenantIdAndCustomerId(tenantId, new CustomerId(entityId.getId()), new TextPageLink(Integer.MAX_VALUE)); + } + return convertToEdgeIds(Futures.immediateFuture(edges.getData())); default: return Futures.immediateFuture(Collections.emptyList()); } diff --git a/application/src/main/java/org/thingsboard/server/service/edge/EdgeContextComponent.java b/application/src/main/java/org/thingsboard/server/service/edge/EdgeContextComponent.java index 32f38d2621..c21c1ebc9c 100644 --- a/application/src/main/java/org/thingsboard/server/service/edge/EdgeContextComponent.java +++ b/application/src/main/java/org/thingsboard/server/service/edge/EdgeContextComponent.java @@ -33,6 +33,9 @@ import org.thingsboard.server.dao.entityview.EntityViewService; import org.thingsboard.server.dao.relation.RelationService; import org.thingsboard.server.dao.rule.RuleChainService; import org.thingsboard.server.dao.user.UserService; +import org.thingsboard.server.queue.discovery.PartitionService; +import org.thingsboard.server.queue.provider.TbQueueProducerProvider; +import org.thingsboard.server.queue.util.TbCoreComponent; import org.thingsboard.server.service.edge.rpc.EdgeEventStorageSettings; import org.thingsboard.server.service.edge.rpc.constructor.AlarmUpdateMsgConstructor; import org.thingsboard.server.service.edge.rpc.constructor.AssetUpdateMsgConstructor; @@ -49,6 +52,7 @@ import org.thingsboard.server.service.queue.TbClusterService; import org.thingsboard.server.service.state.DeviceStateService; @Component +@TbCoreComponent @Data public class EdgeContextComponent { @@ -56,6 +60,13 @@ public class EdgeContextComponent { @Autowired private EdgeService edgeService; + @Autowired + private PartitionService partitionService; + + @Autowired + @Lazy + private TbQueueProducerProvider producerProvider; + @Lazy @Autowired private EdgeNotificationService edgeNotificationService; 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 5e077aea67..cca61aa28e 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 @@ -18,10 +18,10 @@ package org.thingsboard.server.service.edge.rpc; import com.datastax.driver.core.utils.UUIDs; import com.fasterxml.jackson.core.JsonProcessingException; import com.fasterxml.jackson.databind.ObjectMapper; +import com.fasterxml.jackson.databind.node.ObjectNode; import com.google.common.util.concurrent.FutureCallback; import com.google.common.util.concurrent.Futures; import com.google.common.util.concurrent.ListenableFuture; -import com.google.common.util.concurrent.MoreExecutors; import com.google.gson.Gson; import com.google.gson.JsonElement; import com.google.gson.JsonObject; @@ -44,6 +44,7 @@ import org.thingsboard.server.common.data.edge.Edge; import org.thingsboard.server.common.data.edge.EdgeEvent; import org.thingsboard.server.common.data.id.AlarmId; 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.EdgeId; @@ -66,6 +67,8 @@ import org.thingsboard.server.common.data.security.DeviceCredentialsType; import org.thingsboard.server.common.data.security.UserCredentials; import org.thingsboard.server.common.msg.TbMsg; import org.thingsboard.server.common.msg.TbMsgMetaData; +import org.thingsboard.server.common.msg.queue.ServiceType; +import org.thingsboard.server.common.msg.queue.TopicPartitionInfo; import org.thingsboard.server.common.msg.session.SessionMsgType; import org.thingsboard.server.common.transport.util.JsonUtils; import org.thingsboard.server.gen.edge.AlarmUpdateMsg; @@ -96,6 +99,10 @@ import org.thingsboard.server.gen.edge.UplinkResponseMsg; import org.thingsboard.server.gen.edge.UserCredentialsRequestMsg; import org.thingsboard.server.gen.edge.UserCredentialsUpdateMsg; import org.thingsboard.server.gen.transport.TransportProtos; +import org.thingsboard.server.queue.TbQueueCallback; +import org.thingsboard.server.queue.TbQueueMsgMetadata; +import org.thingsboard.server.queue.TbQueueProducer; +import org.thingsboard.server.queue.common.TbProtoQueueMsg; import org.thingsboard.server.service.edge.EdgeContextComponent; import java.io.Closeable; @@ -132,6 +139,8 @@ public final class EdgeGrpcSession implements Closeable { private StreamObserver outputStream; private boolean connected; + private TbQueueProducer> ruleEngineMsgProducer; + EdgeGrpcSession(EdgeContextComponent ctx, StreamObserver outputStream, BiConsumer sessionOpenListener, Consumer sessionCloseListener, ObjectMapper mapper) { this.sessionId = UUID.randomUUID(); @@ -140,6 +149,7 @@ public final class EdgeGrpcSession implements Closeable { this.sessionOpenListener = sessionOpenListener; this.sessionCloseListener = sessionCloseListener; this.mapper = mapper; + this.ruleEngineMsgProducer = ctx.getProducerProvider().getRuleEngineMsgProducer(); initInputStream(); } @@ -767,17 +777,19 @@ public final class EdgeGrpcSession implements Closeable { Device deviceById = ctx.getDeviceService().findDeviceById(edge.getTenantId(), edgeDeviceId); if (deviceById != null) { // this ID already used by other device - create new device and update ID on the edge - Device savedDevice = createDevice(deviceUpdateMsg); + device = createDevice(deviceUpdateMsg); EntityUpdateMsg entityUpdateMsg = EntityUpdateMsg.newBuilder() - .setDeviceUpdateMsg(ctx.getDeviceUpdateMsgConstructor().constructDeviceUpdatedMsg(UpdateMsgType.DEVICE_CONFLICT_RPC_MESSAGE, savedDevice)) + .setDeviceUpdateMsg(ctx.getDeviceUpdateMsgConstructor().constructDeviceUpdatedMsg(UpdateMsgType.DEVICE_CONFLICT_RPC_MESSAGE, device)) .build(); outputStream.onNext(ResponseMsg.newBuilder() .setEntityUpdateMsg(entityUpdateMsg) .build()); } else { - createDevice(deviceUpdateMsg); + device = createDevice(deviceUpdateMsg); } } + // TODO: voba - assign device only in case device is not assigned yet. Missing functionality to check this relation prior assignment + ctx.getDeviceService().assignDeviceToEdge(edge.getTenantId(), device.getId(), edge.getId()); break; case ENTITY_UPDATED_RPC_MESSAGE: updateDevice(deviceUpdateMsg); @@ -866,11 +878,9 @@ public final class EdgeGrpcSession implements Closeable { device.setType(deviceUpdateMsg.getType()); device.setLabel(deviceUpdateMsg.getLabel()); device = ctx.getDeviceService().saveDevice(device); - device = ctx.getDeviceService().assignDeviceToEdge(edge.getTenantId(), device.getId(), edge.getId()); createRelationFromEdge(device.getId()); - ctx.getRelationService().saveRelationAsync(TenantId.SYS_TENANT_ID, new EntityRelation(edge.getId(), device.getId(), "Created")); ctx.getDeviceStateService().onDeviceAdded(device); - + pushDeviceCreatedEventToRuleEngine(device); requestDeviceCredentialsFromEdge(device); } finally { deviceCreationLock.unlock(); @@ -878,6 +888,49 @@ public final class EdgeGrpcSession implements Closeable { return device; } + private void pushDeviceCreatedEventToRuleEngine(Device device) { + try { + ObjectNode entityNode = mapper.valueToTree(device); + TbMsg tbMsg = TbMsg.newMsg(DataConstants.ENTITY_CREATED, device.getId(), getActionTbMsgMetaData(device.getCustomerId()), mapper.writeValueAsString(entityNode)); + sendToRuleEngine(edge.getTenantId(), tbMsg, new TbQueueCallback() { + @Override + public void onSuccess(TbQueueMsgMetadata metadata) { + // TODO: voba - handle success + } + + @Override + public void onFailure(Throwable t) { + // TODO: voba - handle failure + } + }); + } catch (JsonProcessingException | IllegalArgumentException e) { + log.warn("[{}] Failed to push device action to rule engine: {}", device.getId(), DataConstants.ENTITY_CREATED, e); + } + } + + protected void sendToRuleEngine(TenantId tenantId, TbMsg tbMsg, TbQueueCallback callback) { + TopicPartitionInfo tpi = ctx.getPartitionService().resolve(ServiceType.TB_RULE_ENGINE, tenantId, tbMsg.getOriginator()); + TransportProtos.ToRuleEngineMsg msg = TransportProtos.ToRuleEngineMsg.newBuilder().setTbMsg(TbMsg.toByteString(tbMsg)) + .setTenantIdMSB(tenantId.getId().getMostSignificantBits()) + .setTenantIdLSB(tenantId.getId().getLeastSignificantBits()).build(); + ruleEngineMsgProducer.send(tpi, new TbProtoQueueMsg<>(tbMsg.getId(), msg), callback); + } + + private TbMsgMetaData getActionTbMsgMetaData(CustomerId customerId) { + TbMsgMetaData metaData = getTbMsgMetaData(edge); + if (customerId != null && !customerId.isNullUid()) { + metaData.putValue("customerId", customerId.toString()); + } + return metaData; + } + + private TbMsgMetaData getTbMsgMetaData(Edge edge) { + TbMsgMetaData metaData = new TbMsgMetaData(); + metaData.putValue("edgeId", edge.getId().toString()); + metaData.putValue("edgeName", edge.getName()); + return metaData; + } + private EntityId getAlarmOriginator(String entityName, org.thingsboard.server.common.data.EntityType entityType) { switch (entityType) { case DEVICE: