diff --git a/application/src/main/data/json/demo/rule_chains/thermostat_alarms.json b/application/src/main/data/json/demo/rule_chains/thermostat_alarms.json index 155d05f119..e3ba2e3192 100644 --- a/application/src/main/data/json/demo/rule_chains/thermostat_alarms.json +++ b/application/src/main/data/json/demo/rule_chains/thermostat_alarms.json @@ -2,7 +2,7 @@ "ruleChain": { "additionalInfo": null, "name": "Thermostat Alarms", - "type": "SYSTEM", + "type": "CORE", "firstRuleNodeId": null, "root": false, "debugMode": false, diff --git a/application/src/main/data/json/tenant/rule_chains/root_rule_chain.json b/application/src/main/data/json/tenant/rule_chains/root_rule_chain.json index e7591a05aa..cec15f3643 100644 --- a/application/src/main/data/json/tenant/rule_chains/root_rule_chain.json +++ b/application/src/main/data/json/tenant/rule_chains/root_rule_chain.json @@ -2,7 +2,7 @@ "ruleChain": { "additionalInfo": null, "name": "Root Rule Chain", - "type": "SYSTEM", + "type": "CORE", "firstRuleNodeId": null, "root": true, "debugMode": false, 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 96f127d9be..79d0569e2e 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 @@ -15,29 +15,25 @@ */ package org.thingsboard.server.service.edge; +import com.fasterxml.jackson.databind.JsonNode; import com.fasterxml.jackson.databind.ObjectMapper; -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 lombok.extern.slf4j.Slf4j; -import org.apache.commons.codec.binary.Base64; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.stereotype.Service; -import org.thingsboard.server.common.data.Dashboard; -import org.thingsboard.server.common.data.DataConstants; -import org.thingsboard.server.common.data.Device; import org.thingsboard.server.common.data.EntityType; -import org.thingsboard.server.common.data.EntityView; -import org.thingsboard.server.common.data.Event; import org.thingsboard.server.common.data.alarm.Alarm; -import org.thingsboard.server.common.data.asset.Asset; 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.DashboardId; import org.thingsboard.server.common.data.id.EdgeId; import org.thingsboard.server.common.data.id.EntityId; +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.page.TimePageData; @@ -46,29 +42,28 @@ import org.thingsboard.server.common.data.relation.EntityRelation; import org.thingsboard.server.common.data.relation.RelationTypeGroup; import org.thingsboard.server.common.data.rule.RuleChain; import org.thingsboard.server.common.data.rule.RuleChainMetaData; -import org.thingsboard.server.common.data.rule.RuleChainType; -import org.thingsboard.server.common.msg.TbMsg; import org.thingsboard.server.common.msg.queue.TbCallback; -import org.thingsboard.server.common.msg.session.SessionMsgType; -import org.thingsboard.server.dao.asset.AssetService; -import org.thingsboard.server.dao.device.DeviceService; +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.entityview.EntityViewService; -import org.thingsboard.server.dao.event.EventService; import org.thingsboard.server.dao.relation.RelationService; import org.thingsboard.server.dao.rule.RuleChainService; import org.thingsboard.server.gen.transport.TransportProtos; import org.thingsboard.server.queue.util.TbCoreComponent; +import org.thingsboard.server.service.executors.DbCallbackExecutorService; -import javax.annotation.Nullable; import javax.annotation.PostConstruct; import javax.annotation.PreDestroy; import java.io.IOException; +import java.util.ArrayList; +import java.util.Collections; +import java.util.HashSet; import java.util.List; +import java.util.Set; import java.util.UUID; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; +import java.util.stream.Collectors; @Service @TbCoreComponent @@ -81,16 +76,10 @@ public class DefaultEdgeNotificationService implements EdgeNotificationService { private EdgeService edgeService; @Autowired - private DeviceService deviceService; - - @Autowired - private AssetService assetService; - - @Autowired - private EntityViewService entityViewService; + private RuleChainService ruleChainService; @Autowired - private RuleChainService ruleChainService; + private AlarmService alarmService; @Autowired private RelationService relationService; @@ -98,6 +87,9 @@ public class DefaultEdgeNotificationService implements EdgeNotificationService { @Autowired private EdgeEventService edgeEventService; + @Autowired + private DbCallbackExecutorService dbCallbackExecutorService; + private ExecutorService tsCallBackExecutor; @PostConstruct @@ -121,315 +113,160 @@ public class DefaultEdgeNotificationService implements EdgeNotificationService { public Edge setEdgeRootRuleChain(TenantId tenantId, Edge edge, RuleChainId ruleChainId) throws IOException { edge.setRootRuleChainId(ruleChainId); Edge savedEdge = edgeService.saveEdge(edge); - RuleChain ruleChain = ruleChainService.findRuleChainById(tenantId, ruleChainId); - saveEventToEdgeQueue(tenantId, edge.getId(), EdgeEventType.RULE_CHAIN, DataConstants.ENTITY_UPDATED, mapper.writeValueAsString(ruleChain), new FutureCallback() { - @Override - public void onSuccess(@Nullable Void aVoid) { - log.debug("Event saved successfully!"); - } - - @Override - public void onFailure(Throwable t) { - log.debug("Failure during event save", t); - } - }); + saveEdgeEvent(tenantId, edge.getId(), EdgeEventType.RULE_CHAIN, ActionType.UPDATED, ruleChainId, null); return savedEdge; } - private void saveEventToEdgeQueue(TenantId tenantId, EdgeId edgeId, EdgeEventType entityType, String type, String data, FutureCallback callback) throws IOException { - log.debug("Pushing single event to edge queue. tenantId [{}], edgeId [{}], entityType [{}], type[{}], data [{}]", tenantId, edgeId, entityType, type, data); - -// EdgeEQueueEntry queueEntry = new EdgeQueueEntry(); -// queueEntry.setEntityType(entityType); -// queueEntry.setType(type); -// queueEntry.setData(data); + private void saveEdgeEvent(TenantId tenantId, + EdgeId edgeId, + EdgeEventType edgeEventType, + ActionType edgeEventAction, + EntityId entityId, + JsonNode entityBody) { + log.debug("Pushing edge event to edge queue. tenantId [{}], edgeId [{}], edgeEventType [{}], edgeEventAction[{}], entityId [{}], entityBody [{}]", + tenantId, edgeId, edgeEventType, edgeEventAction, entityId, entityBody); EdgeEvent edgeEvent = new EdgeEvent(); edgeEvent.setEdgeId(edgeId); edgeEvent.setTenantId(tenantId); -// event.setType(DataConstants.EDGE_QUEUE_EVENT_TYPE); -// event.setBody(mapper.valueToTree(queueEntry)); - ListenableFuture saveFuture = edgeEventService.saveAsync(edgeEvent); - - addMainCallback(saveFuture, callback); - } - - private void addMainCallback(ListenableFuture saveFuture, final FutureCallback callback) { - Futures.addCallback(saveFuture, new FutureCallback() { - @Override - public void onSuccess(@Nullable EdgeEvent result) { - callback.onSuccess(null); - } - - @Override - public void onFailure(Throwable t) { - callback.onFailure(t); - } - }, tsCallBackExecutor); + edgeEvent.setEdgeEventType(edgeEventType); + edgeEvent.setEdgeEventAction(edgeEventAction.name()); + if (entityId != null) { + edgeEvent.setEntityId(entityId.getId()); + } + edgeEvent.setEntityBody(entityBody); + edgeEventService.saveAsync(edgeEvent); } @Override public void pushNotificationToEdge(TransportProtos.EdgeNotificationMsgProto edgeNotificationMsg, TbCallback callback) { -// if (tbMsg.getType().equals(SessionMsgType.POST_TELEMETRY_REQUEST.name()) || -// tbMsg.getType().equals(SessionMsgType.POST_ATTRIBUTES_REQUEST.name()) || -// tbMsg.getType().equals(DataConstants.ATTRIBUTES_UPDATED) || -// tbMsg.getType().equals(DataConstants.ATTRIBUTES_DELETED)) { -// processCustomTbMsg(tenantId, tbMsg, callback); -// } else { -// try { -// switch (tbMsg.getOriginator().getEntityType()) { -// case EDGE: -// processEdge(tenantId, tbMsg, callback); -// break; -// case ASSET: -// processAsset(tenantId, tbMsg, callback); -// break; -// case DEVICE: -// processDevice(tenantId, tbMsg, callback); -// break; -// case DASHBOARD: -// processDashboard(tenantId, tbMsg, callback); -// break; -// case RULE_CHAIN: -// processRuleChain(tenantId, tbMsg, callback); -// break; -// case ENTITY_VIEW: -// processEntityView(tenantId, tbMsg, callback); -// break; -// case ALARM: -// processAlarm(tenantId, tbMsg, callback); -// break; -// default: -// log.debug("Entity type [{}] is not designed to be pushed to edge", tbMsg.getOriginator().getEntityType()); -// } -// } catch (IOException e) { -// log.error("Can't push to edge updates, entity type [{}], data [{}]", tbMsg.getOriginator().getEntityType(), tbMsg.getData(), e); -// } -// } - } - - - private void processCustomTbMsg(TenantId tenantId, TbMsg tbMsg, FutureCallback callback) { - ListenableFuture edgeIdFuture = getEdgeIdByOriginatorId(tenantId, tbMsg.getOriginator()); - Futures.transform(edgeIdFuture, edgeId -> { - EdgeEventType edgeEventType = getEdgeQueueTypeByEntityType(tbMsg.getOriginator().getEntityType()); - if (edgeId != null && edgeEventType != null) { - try { - saveEventToEdgeQueue(tenantId, edgeId, edgeEventType, tbMsg.getType(), Base64.encodeBase64String(TbMsg.toByteArray(tbMsg)), callback); - } catch (IOException e) { - log.error("Error while saving custom tbMsg into Edge Queue", e); - } + try { + EdgeEventType edgeEventType = EdgeEventType.valueOf(edgeNotificationMsg.getEdgeEventType()); + ActionType edgeEventAction = ActionType.valueOf(edgeNotificationMsg.getEdgeEventAction()); + TenantId tenantId = new TenantId(new UUID(edgeNotificationMsg.getTenantIdMSB(), edgeNotificationMsg.getTenantIdLSB())); + switch (edgeEventType) { + // TODO: voba - handle edge updates + // case EDGE: + case ASSET: + case DEVICE: + case ENTITY_VIEW: + case DASHBOARD: + case RULE_CHAIN: + EntityId entityId = EntityIdFactory.getByEdgeEventTypeAndUuid(edgeEventType, new UUID(edgeNotificationMsg.getEntityIdMSB(), edgeNotificationMsg.getEntityIdLSB())); + ListenableFuture> edgeIdsFuture = findRelatedEdgeIdsEntityId(tenantId, entityId); + Futures.transform(edgeIdsFuture, edgeIds -> { + if (edgeIds != null && !edgeIds.isEmpty()) { + for (EdgeId edgeId : edgeIds) { + try { + saveEdgeEvent(tenantId, edgeId, edgeEventType, edgeEventAction, entityId, null); + if (edgeEventType.equals(EdgeEventType.RULE_CHAIN) && + (ActionType.UPDATED.equals(edgeEventAction) || ActionType.ADDED.equals(edgeEventAction))) { + RuleChainMetaData ruleChainMetaData = ruleChainService.loadRuleChainMetaData(tenantId, new RuleChainId(entityId.getId())); + saveEdgeEvent(tenantId, edgeId, EdgeEventType.RULE_CHAIN_METADATA, edgeEventAction, ruleChainMetaData.getRuleChainId(), null); + } + } catch (Exception e) { + log.error("[{}] Failed to push event to edge, edgeId [{}], edgeEventType [{}], edgeEventAction [{}], entityId [{}]", + tenantId, edgeId, edgeEventType, edgeEventAction, entityId, e); + } + } + } + return null; + }, dbCallbackExecutorService); + break; + case ALARM: + EntityId alarmId = EntityIdFactory.getByEdgeEventTypeAndUuid(edgeEventType, new UUID(edgeNotificationMsg.getEntityIdMSB(), edgeNotificationMsg.getEntityIdLSB())); + processAlarm(tenantId, edgeEventAction, alarmId); + break; + case RELATION: + EntityRelation entityRelation = mapper.convertValue(edgeNotificationMsg.getEntityBody(), EntityRelation.class); + processRelation(tenantId, edgeEventAction, entityRelation); + break; + default: + log.debug("Edge event type [{}] is not designed to be pushed to edge", edgeEventType); } - return null; - }, MoreExecutors.directExecutor()); - } - - private void processDevice(TenantId tenantId, TbMsg tbMsg, FutureCallback callback) throws IOException { - switch (tbMsg.getType()) { - case DataConstants.ENTITY_ASSIGNED_TO_EDGE: - case DataConstants.ENTITY_UNASSIGNED_FROM_EDGE: - processAssignedEntity(tenantId, tbMsg, EdgeEventType.DEVICE, callback); - break; - case DataConstants.ENTITY_DELETED: - case DataConstants.ENTITY_CREATED: - case DataConstants.ENTITY_UPDATED: - Device device = mapper.readValue(tbMsg.getData(), Device.class); - pushEventToEdge(tenantId, device.getId(), EdgeEventType.DEVICE, tbMsg, callback); - break; - default: - log.warn("Unsupported msgType [{}], tbMsg [{}]", tbMsg.getType(), tbMsg); - } - } - - private void processEdge(TenantId tenantId, TbMsg tbMsg, FutureCallback callback) throws IOException { - switch (tbMsg.getType()) { - case DataConstants.ENTITY_DELETED: - case DataConstants.ENTITY_CREATED: - case DataConstants.ENTITY_UPDATED: - // TODO: voba - handle properly edge creation - break; - default: - log.warn("Unsupported msgType [{}], tbMsg [{}]", tbMsg.getType(), tbMsg); - } - } - - private void processAsset(TenantId tenantId, TbMsg tbMsg, FutureCallback callback) throws IOException { - switch (tbMsg.getType()) { - case DataConstants.ENTITY_ASSIGNED_TO_EDGE: - case DataConstants.ENTITY_UNASSIGNED_FROM_EDGE: - processAssignedEntity(tenantId, tbMsg, EdgeEventType.ASSET, callback); - break; - case DataConstants.ENTITY_DELETED: - case DataConstants.ENTITY_CREATED: - case DataConstants.ENTITY_UPDATED: - Asset asset = mapper.readValue(tbMsg.getData(), Asset.class); - pushEventToEdge(tenantId, asset.getId(), EdgeEventType.ASSET, tbMsg, callback); - break; - default: - log.warn("Unsupported msgType [{}], tbMsg [{}]", tbMsg.getType(), tbMsg); + } catch (Exception e) { + callback.onFailure(e); + log.error("Can't push to edge updates, edgeNotificationMsg [{}]", edgeNotificationMsg, e); + } finally { + callback.onSuccess(); } } - private void processEntityView(TenantId tenantId, TbMsg tbMsg, FutureCallback callback) throws IOException { - switch (tbMsg.getType()) { - case DataConstants.ENTITY_ASSIGNED_TO_EDGE: - case DataConstants.ENTITY_UNASSIGNED_FROM_EDGE: - processAssignedEntity(tenantId, tbMsg, EdgeEventType.ENTITY_VIEW, callback); - break; - case DataConstants.ENTITY_DELETED: - case DataConstants.ENTITY_CREATED: - case DataConstants.ENTITY_UPDATED: - EntityView entityView = mapper.readValue(tbMsg.getData(), EntityView.class); - pushEventToEdge(tenantId, entityView.getId(), EdgeEventType.ENTITY_VIEW, tbMsg, callback); - break; - default: - log.warn("Unsupported msgType [{}], tbMsg [{}]", tbMsg.getType(), tbMsg); - } - } - - private void processAlarm(TenantId tenantId, TbMsg tbMsg, FutureCallback callback) throws IOException { - switch (tbMsg.getType()) { - case DataConstants.ENTITY_DELETED: - case DataConstants.ENTITY_CREATED: - case DataConstants.ENTITY_UPDATED: - case DataConstants.ALARM_ACK: - case DataConstants.ALARM_CLEAR: - Alarm alarm = mapper.readValue(tbMsg.getData(), Alarm.class); + private void processAlarm(TenantId tenantId, ActionType edgeActionType, EntityId alarmId) { + ListenableFuture alarmFuture = alarmService.findAlarmByIdAsync(tenantId, new AlarmId(alarmId.getId())); + Futures.transform(alarmFuture, alarm -> { + if (alarm != null) { EdgeEventType edgeEventType = getEdgeQueueTypeByEntityType(alarm.getOriginator().getEntityType()); if (edgeEventType != null) { - pushEventToEdge(tenantId, alarm.getOriginator(), EdgeEventType.ALARM, tbMsg, callback); - } - break; - default: - log.warn("Unsupported msgType [{}], tbMsg [{}]", tbMsg.getType(), tbMsg); - } - } - - private void processDashboard(TenantId tenantId, TbMsg tbMsg, FutureCallback callback) throws IOException { - switch (tbMsg.getType()) { - case DataConstants.ENTITY_ASSIGNED_TO_EDGE: - case DataConstants.ENTITY_UNASSIGNED_FROM_EDGE: - processAssignedEntity(tenantId, tbMsg, EdgeEventType.DASHBOARD, callback); - break; - case DataConstants.ENTITY_DELETED: - case DataConstants.ENTITY_CREATED: - case DataConstants.ENTITY_UPDATED: - Dashboard dashboard = mapper.readValue(tbMsg.getData(), Dashboard.class); - ListenableFuture> future = edgeService.findEdgesByTenantIdAndDashboardId(tenantId, dashboard.getId(), new TimePageLink(Integer.MAX_VALUE)); - Futures.transform(future, edges -> { - if (edges != null && edges.getData() != null && !edges.getData().isEmpty()) { - try { - for (Edge edge : edges.getData()) { - pushEventToEdge(tenantId, edge.getId(), EdgeEventType.DASHBOARD, tbMsg, callback); - } - } catch (IOException e) { - log.error("Can't push event to edge", e); - } - } - return null; - }, MoreExecutors.directExecutor()); - break; - default: - log.warn("Unsupported msgType [{}], tbMsg [{}]", tbMsg.getType(), tbMsg); - } - } - - private void processRuleChain(TenantId tenantId, TbMsg tbMsg, FutureCallback callback) throws IOException { - switch (tbMsg.getType()) { - case DataConstants.ENTITY_ASSIGNED_TO_EDGE: - case DataConstants.ENTITY_UNASSIGNED_FROM_EDGE: - processAssignedEntity(tenantId, tbMsg, EdgeEventType.RULE_CHAIN, callback); - break; - case DataConstants.ENTITY_DELETED: - case DataConstants.ENTITY_CREATED: - case DataConstants.ENTITY_UPDATED: - RuleChain ruleChain = mapper.readValue(tbMsg.getData(), RuleChain.class); - if (RuleChainType.EDGE.equals(ruleChain.getType())) { - ListenableFuture> future = edgeService.findEdgesByTenantIdAndRuleChainId(tenantId, ruleChain.getId(), new TimePageLink(Integer.MAX_VALUE)); - Futures.transform(future, edges -> { - if (edges != null && edges.getData() != null && !edges.getData().isEmpty()) { - try { - for (Edge edge : edges.getData()) { - pushEventToEdge(tenantId, edge.getId(), EdgeEventType.RULE_CHAIN, tbMsg, callback); - } - } catch (IOException e) { - log.error("Can't push event to edge", e); + ListenableFuture> relatedEdgeIdsEntityIdFuture = findRelatedEdgeIdsEntityId(tenantId, alarm.getOriginator()); + Futures.transform(relatedEdgeIdsEntityIdFuture, relatedEdgeIdsEntityId -> { + if (relatedEdgeIdsEntityId != null) { + for (EdgeId edgeId : relatedEdgeIdsEntityId) { + saveEdgeEvent(tenantId, edgeId, EdgeEventType.ALARM, edgeActionType, alarmId, null); } } return null; - }, MoreExecutors.directExecutor()); + }, dbCallbackExecutorService); } - break; - default: - log.warn("Unsupported msgType [{}], tbMsg [{}]", tbMsg.getType(), tbMsg); - } - } - - - private void processAssignedEntity(TenantId tenantId, TbMsg tbMsg, EdgeEventType entityType, FutureCallback callback) throws IOException { - EdgeId edgeId; - switch (tbMsg.getType()) { - case DataConstants.ENTITY_ASSIGNED_TO_EDGE: - edgeId = new EdgeId(UUID.fromString(tbMsg.getMetaData().getValue("assignedEdgeId"))); - pushEventToEdge(tenantId, edgeId, entityType, tbMsg, callback); - break; - case DataConstants.ENTITY_UNASSIGNED_FROM_EDGE: - edgeId = new EdgeId(UUID.fromString(tbMsg.getMetaData().getValue("unassignedEdgeId"))); - pushEventToEdge(tenantId, edgeId, entityType, tbMsg, callback); - break; - - } + } + return null; + }, dbCallbackExecutorService); } - private void pushEventToEdge(TenantId tenantId, EntityId originatorId, EdgeEventType edgeEventType, TbMsg tbMsg, FutureCallback callback) { - ListenableFuture edgeIdFuture = getEdgeIdByOriginatorId(tenantId, originatorId); - Futures.transform(edgeIdFuture, edgeId -> { - if (edgeId != null) { - try { - pushEventToEdge(tenantId, edgeId, edgeEventType, tbMsg, callback); - } catch (Exception e) { - log.error("Failed to push event to edge, edgeId [{}], tbMsg [{}]", edgeId, tbMsg, e); - } + private void processRelation(TenantId tenantId, ActionType edgeActionType, EntityRelation entityRelation) { + List>> futures = new ArrayList<>(); + futures.add(findRelatedEdgeIdsEntityId(tenantId, entityRelation.getTo())); + futures.add(findRelatedEdgeIdsEntityId(tenantId, entityRelation.getFrom())); + ListenableFuture>> combinedFuture = Futures.allAsList(futures); + Futures.transform(combinedFuture, listOfListsEdgeIds -> { + Set uniqueEdgeIds = new HashSet<>(); + if (listOfListsEdgeIds != null && !listOfListsEdgeIds.isEmpty()) { + for (List listOfListsEdgeId : listOfListsEdgeIds) { + if (listOfListsEdgeId != null) { + uniqueEdgeIds.addAll(listOfListsEdgeId); } - return null; - }, - MoreExecutors.directExecutor()); - } - - private ListenableFuture getEdgeIdByOriginatorId(TenantId tenantId, EntityId originatorId) { - List originatorEdgeRelations = relationService.findByToAndType(tenantId, originatorId, EntityRelation.CONTAINS_TYPE, RelationTypeGroup.EDGE); - if (originatorEdgeRelations != null && originatorEdgeRelations.size() > 0) { - return Futures.immediateFuture(new EdgeId(originatorEdgeRelations.get(0).getFrom().getId())); - } else { - return Futures.immediateFuture(null); - } - } - - - private void pushEventToEdge(TenantId tenantId, EdgeId edgeId, EdgeEventType entityType, TbMsg tbMsg, FutureCallback callback) throws IOException { - log.debug("Pushing event(s) to edge queue. tenantId [{}], edgeId [{}], entityType [{}], tbMsg [{}]", tenantId, edgeId, entityType, tbMsg); - - saveEventToEdgeQueue(tenantId, edgeId, entityType, tbMsg.getType(), tbMsg.getData(), callback); - - if (entityType.equals(EdgeEventType.RULE_CHAIN)) { - pushRuleChainMetadataToEdge(tenantId, edgeId, tbMsg, callback); - } + } + } + if (!uniqueEdgeIds.isEmpty()) { + for (EdgeId edgeId : uniqueEdgeIds) { + saveEdgeEvent(tenantId, edgeId, EdgeEventType.RELATION, edgeActionType, null, mapper.valueToTree(entityRelation)); + } + } + return null; + }, dbCallbackExecutorService); } - private void pushRuleChainMetadataToEdge(TenantId tenantId, EdgeId edgeId, TbMsg tbMsg, FutureCallback callback) throws IOException { - RuleChain ruleChain = mapper.readValue(tbMsg.getData(), RuleChain.class); - switch (tbMsg.getType()) { - case DataConstants.ENTITY_ASSIGNED_TO_EDGE: - case DataConstants.ENTITY_UNASSIGNED_FROM_EDGE: - case DataConstants.ENTITY_UPDATED: - RuleChainMetaData ruleChainMetaData = ruleChainService.loadRuleChainMetaData(tenantId, ruleChain.getId()); - saveEventToEdgeQueue(tenantId, edgeId, EdgeEventType.RULE_CHAIN_METADATA, tbMsg.getType(), mapper.writeValueAsString(ruleChainMetaData), callback); - break; + private ListenableFuture> findRelatedEdgeIdsEntityId(TenantId tenantId, EntityId entityId) { + switch (entityId.getEntityType()) { + case DEVICE: + case ASSET: + case ENTITY_VIEW: + ListenableFuture> originatorEdgeRelationsFuture = relationService.findByToAndTypeAsync(tenantId, entityId, EntityRelation.CONTAINS_TYPE, RelationTypeGroup.EDGE); + return Futures.transform(originatorEdgeRelationsFuture, originatorEdgeRelations -> { + if (originatorEdgeRelations != null && originatorEdgeRelations.size() > 0) { + return Collections.singletonList(new EdgeId(originatorEdgeRelations.get(0).getFrom().getId())); + } else { + return Collections.emptyList(); + } + }, dbCallbackExecutorService); + case DASHBOARD: + return convertToEdgeIds(edgeService.findEdgesByTenantIdAndDashboardId(tenantId, new DashboardId(entityId.getId()), new TimePageLink(Integer.MAX_VALUE))); + case RULE_CHAIN: + return convertToEdgeIds(edgeService.findEdgesByTenantIdAndRuleChainId(tenantId, new RuleChainId(entityId.getId()), new TimePageLink(Integer.MAX_VALUE))); default: - log.warn("Unsupported msgType [{}], tbMsg [{}]", tbMsg.getType(), tbMsg); + return Futures.immediateFuture(Collections.emptyList()); } } + private ListenableFuture> convertToEdgeIds(ListenableFuture> future) { + return Futures.transform(future, edges -> { + if (edges != null && edges.getData() != null && !edges.getData().isEmpty()) { + return edges.getData().stream().map(IdBased::getId).collect(Collectors.toList()); + } else { + return Collections.emptyList(); + } + }, dbCallbackExecutorService); + } private EdgeEventType getEdgeQueueTypeByEntityType(EntityType entityType) { switch (entityType) { @@ -446,3 +283,4 @@ public class DefaultEdgeNotificationService implements EdgeNotificationService { } } + 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 26e094f9b3..32f38d2621 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 @@ -31,11 +31,14 @@ import org.thingsboard.server.dao.device.DeviceService; import org.thingsboard.server.dao.edge.EdgeService; 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.service.edge.rpc.EdgeEventStorageSettings; import org.thingsboard.server.service.edge.rpc.constructor.AlarmUpdateMsgConstructor; import org.thingsboard.server.service.edge.rpc.constructor.AssetUpdateMsgConstructor; import org.thingsboard.server.service.edge.rpc.constructor.DashboardUpdateMsgConstructor; import org.thingsboard.server.service.edge.rpc.constructor.DeviceUpdateMsgConstructor; +import org.thingsboard.server.service.edge.rpc.constructor.EntityDataMsgConstructor; import org.thingsboard.server.service.edge.rpc.constructor.EntityViewUpdateMsgConstructor; import org.thingsboard.server.service.edge.rpc.constructor.RelationUpdateMsgConstructor; import org.thingsboard.server.service.edge.rpc.constructor.RuleChainUpdateMsgConstructor; @@ -93,6 +96,14 @@ public class EdgeContextComponent { @Autowired private DashboardService dashboardService; + @Lazy + @Autowired + private RuleChainService ruleChainService; + + @Lazy + @Autowired + private UserService userService; + @Lazy @Autowired private ActorService actorService; @@ -141,6 +152,10 @@ public class EdgeContextComponent { @Autowired private RelationUpdateMsgConstructor relationUpdateMsgConstructor; + @Lazy + @Autowired + private EntityDataMsgConstructor entityDataMsgConstructor; + @Lazy @Autowired private EdgeEventStorageSettings edgeEventStorageSettings; 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 aa48f0a370..a3d9bb7593 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 @@ -19,15 +19,17 @@ 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.protobuf.ByteString; +import com.google.gson.Gson; +import com.google.gson.JsonElement; +import com.google.gson.JsonObject; import io.grpc.stub.StreamObserver; import lombok.Data; import lombok.extern.slf4j.Slf4j; -import org.apache.commons.codec.binary.Base64; -import org.thingsboard.server.common.data.Customer; +import org.checkerframework.checker.nullness.qual.Nullable; import org.thingsboard.server.common.data.Dashboard; import org.thingsboard.server.common.data.DataConstants; import org.thingsboard.server.common.data.Device; @@ -41,12 +43,16 @@ import org.thingsboard.server.common.data.asset.Asset; 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.id.AlarmId; import org.thingsboard.server.common.data.id.AssetId; +import org.thingsboard.server.common.data.id.DashboardId; import org.thingsboard.server.common.data.id.DeviceId; import org.thingsboard.server.common.data.id.EdgeId; 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.AttributeKvEntry; import org.thingsboard.server.common.data.kv.BaseAttributeKvEntry; import org.thingsboard.server.common.data.kv.DataType; @@ -60,14 +66,13 @@ import org.thingsboard.server.common.data.rule.RuleChainMetaData; import org.thingsboard.server.common.data.security.DeviceCredentials; import org.thingsboard.server.common.data.security.DeviceCredentialsType; import org.thingsboard.server.common.msg.TbMsg; -import org.thingsboard.server.common.msg.TbMsgDataType; import org.thingsboard.server.common.msg.TbMsgMetaData; -import org.thingsboard.server.common.msg.queue.TbMsgCallback; +import org.thingsboard.server.common.msg.session.SessionMsgType; +import org.thingsboard.server.common.transport.util.JsonUtils; import org.thingsboard.server.gen.edge.AlarmUpdateMsg; import org.thingsboard.server.gen.edge.ConnectRequestMsg; import org.thingsboard.server.gen.edge.ConnectResponseCode; import org.thingsboard.server.gen.edge.ConnectResponseMsg; -import org.thingsboard.server.gen.edge.CustomerUpdateMsg; import org.thingsboard.server.gen.edge.DeviceUpdateMsg; import org.thingsboard.server.gen.edge.DownlinkMsg; import org.thingsboard.server.gen.edge.EdgeConfiguration; @@ -81,7 +86,7 @@ import org.thingsboard.server.gen.edge.RuleChainMetadataUpdateMsg; import org.thingsboard.server.gen.edge.UpdateMsgType; import org.thingsboard.server.gen.edge.UplinkMsg; import org.thingsboard.server.gen.edge.UplinkResponseMsg; -import org.thingsboard.server.gen.edge.UserUpdateMsg; +import org.thingsboard.server.gen.transport.TransportProtos; import org.thingsboard.server.service.edge.EdgeContextComponent; import java.io.Closeable; @@ -103,6 +108,8 @@ public final class EdgeGrpcSession implements Closeable { private static final ReentrantLock deviceCreationLock = new ReentrantLock(); + private final Gson gson = new Gson(); + private static final String QUEUE_START_TS_ATTR_KEY = "queueStartTs"; private final UUID sessionId; @@ -182,9 +189,9 @@ public final class EdgeGrpcSession implements Closeable { processTelemetryMessage(edgeEvent); } else { processEntityCRUDMessage(edgeEvent, msgType); - } - if (ENTITY_CREATED_RPC_MESSAGE.equals(msgType)) { - pushEntityAttributesToEdge(edgeEvent); + if (ENTITY_CREATED_RPC_MESSAGE.equals(msgType)) { + pushEntityAttributesToEdge(edgeEvent); + } } } catch (Exception e) { log.error("Exception during processing records from queue", e); @@ -213,63 +220,66 @@ public final class EdgeGrpcSession implements Closeable { } } + private ListenableFuture getQueueStartTs() { + ListenableFuture> future = + ctx.getAttributesService().find(edge.getTenantId(), edge.getId(), DataConstants.SERVER_SCOPE, QUEUE_START_TS_ATTR_KEY); + return Futures.transform(future, attributeKvEntryOpt -> { + if (attributeKvEntryOpt != null && attributeKvEntryOpt.isPresent()) { + AttributeKvEntry attributeKvEntry = attributeKvEntryOpt.get(); + return attributeKvEntry.getLongValue().isPresent() ? attributeKvEntry.getLongValue().get() : 0L; + } else { + return 0L; + } + }, MoreExecutors.directExecutor()); + } + + private void updateQueueStartTs(Long newStartTs) { + newStartTs = ++newStartTs; // increments ts by 1 - next edge event search starts from current offset + 1 + List attributes = Collections.singletonList(new BaseAttributeKvEntry(new LongDataEntry(QUEUE_START_TS_ATTR_KEY, newStartTs), System.currentTimeMillis())); + ctx.getAttributesService().save(edge.getTenantId(), edge.getId(), DataConstants.SERVER_SCOPE, attributes); + } + private void pushEntityAttributesToEdge(EdgeEvent edgeEvent) throws IOException { EntityId entityId = null; - String entityName = null; switch (edgeEvent.getEdgeEventType()) { case EDGE: - Edge edge = objectMapper.readValue(entry.getData(), Edge.class); entityId = edge.getId(); - entityName = edge.getName(); break; case DEVICE: - Device device = objectMapper.readValue(entry.getData(), Device.class); - entityId = device.getId(); - entityName = device.getName(); + entityId = new DeviceId(edgeEvent.getEntityId()); break; case ASSET: - Asset asset = objectMapper.readValue(entry.getData(), Asset.class); - entityId = asset.getId(); - entityName = asset.getName(); + entityId = new AssetId(edgeEvent.getEntityId()); break; case ENTITY_VIEW: - EntityView entityView = objectMapper.readValue(entry.getData(), EntityView.class); - entityId = entityView.getId(); - entityName = entityView.getName(); + entityId = new EntityViewId(edgeEvent.getEntityId()); break; case DASHBOARD: - Dashboard dashboard = objectMapper.readValue(entry.getData(), Dashboard.class); - entityId = dashboard.getId(); - entityName = dashboard.getName(); + entityId = new DashboardId(edgeEvent.getEntityId()); break; } if (entityId != null) { final EntityId finalEntityId = entityId; - final String finalEntityName = entityName; ListenableFuture> ssAttrFuture = ctx.getAttributesService().findAll(edge.getTenantId(), entityId, DataConstants.SERVER_SCOPE); Futures.transform(ssAttrFuture, ssAttributes -> { if (ssAttributes != null && !ssAttributes.isEmpty()) { try { - TbMsgMetaData metaData = new TbMsgMetaData(); ObjectNode entityNode = objectMapper.createObjectNode(); - metaData.putValue("scope", DataConstants.SERVER_SCOPE); for (AttributeKvEntry attr : ssAttributes) { - if (attr.getDataType() == DataType.BOOLEAN) { + if (attr.getDataType() == DataType.BOOLEAN && attr.getBooleanValue().isPresent()) { entityNode.put(attr.getKey(), attr.getBooleanValue().get()); - } else if (attr.getDataType() == DataType.DOUBLE) { + } else if (attr.getDataType() == DataType.DOUBLE && attr.getDoubleValue().isPresent()) { entityNode.put(attr.getKey(), attr.getDoubleValue().get()); - } else if (attr.getDataType() == DataType.LONG) { + } else if (attr.getDataType() == DataType.LONG && attr.getLongValue().isPresent()) { entityNode.put(attr.getKey(), attr.getLongValue().get()); } else { entityNode.put(attr.getKey(), attr.getValueAsString()); } } - TbMsg tbMsg = TbMsg.newMsg(DataConstants.ATTRIBUTES_UPDATED, finalEntityId, metaData, TbMsgDataType.JSON - , objectMapper.writeValueAsString(entityNode)); - log.debug("Sending donwlink entity data msg, entityName [{}], tbMsg [{}]", finalEntityName, tbMsg); + log.debug("Sending attributes data msg, entityId [{}], attributes [{}]", finalEntityId, entityNode); + DownlinkMsg value = constructEntityDataProtoMsg(finalEntityId, ActionType.ATTRIBUTES_UPDATED, JsonUtils.parse(objectMapper.writeValueAsString(entityNode))); outputStream.onNext(ResponseMsg.newBuilder() - .setDownlinkMsg(constructEntityDataProtoMsg(finalEntityName, finalEntityId, tbMsg)) - .build()); + .setDownlinkMsg(value).build()); } catch (Exception e) { log.error("[{}] Failed to send attribute updates to the edge", edge.getName(), e); } @@ -283,188 +293,276 @@ public final class EdgeGrpcSession implements Closeable { private void processTelemetryMessage(EdgeEvent edgeEvent) throws IOException { log.trace("Executing processTelemetryMessage, edgeEvent [{}]", edgeEvent); - TbMsg tbMsg = TbMsg.fromBytes(Base64.decodeBase64(entry.getData()), TbMsgCallback.EMPTY); - String entityName = null; EntityId entityId = null; switch (edgeEvent.getEdgeEventType()) { case DEVICE: - Device device = ctx.getDeviceService().findDeviceById(edge.getTenantId(), new DeviceId(tbMsg.getOriginator().getId())); - entityName = device.getName(); - entityId = device.getId(); + entityId = new DeviceId(edgeEvent.getEntityId()); break; case ASSET: - Asset asset = ctx.getAssetService().findAssetById(edge.getTenantId(), new AssetId(tbMsg.getOriginator().getId())); - entityName = asset.getName(); - entityId = asset.getId(); + entityId = new AssetId(edgeEvent.getEntityId()); break; case ENTITY_VIEW: - EntityView entityView = ctx.getEntityViewService().findEntityViewById(edge.getTenantId(), new EntityViewId(tbMsg.getOriginator().getId())); - entityName = entityView.getName(); - entityId = entityView.getId(); + entityId = new EntityViewId(edgeEvent.getEntityId()); break; } - if (entityName != null && entityId != null) { - log.debug("Sending downlink entity data msg, entityName [{}], tbMsg [{}]", entityName, tbMsg); - outputStream.onNext(ResponseMsg.newBuilder() - .setDownlinkMsg(constructEntityDataProtoMsg(entityName, entityId, tbMsg)) - .build()); + if (entityId != null) { + log.debug("Sending telemetry data msg, entityId [{}], body [{}]", edgeEvent.getEntityId(), edgeEvent.getEntityBody()); + DownlinkMsg downlinkMsg; + try { + ActionType actionType = ActionType.valueOf(edgeEvent.getEdgeEventAction()); + downlinkMsg = constructEntityDataProtoMsg(entityId, actionType, JsonUtils.parse(objectMapper.writeValueAsString(edgeEvent.getEntityBody()))); + outputStream.onNext(ResponseMsg.newBuilder() + .setDownlinkMsg(downlinkMsg) + .build()); + } catch (Exception e) { + log.warn("Can't send telemetry data msg, entityId [{}], body [{}]", edgeEvent.getEntityId(), edgeEvent.getEntityBody(), e); + } + } } - private void processEntityCRUDMessage(EdgeEvent edgeEvent, UpdateMsgType msgType) throws java.io.IOException { + private void processEntityCRUDMessage(EdgeEvent edgeEvent, UpdateMsgType msgType) { log.trace("Executing processEntityCRUDMessage, edgeEvent [{}], msgType [{}]", edgeEvent, msgType); switch (edgeEvent.getEdgeEventType()) { case EDGE: - Edge edge = objectMapper.readValue(entry.getData(), Edge.class); - onEdgeUpdated(msgType, edge); + // TODO: voba - add edge update logic break; case DEVICE: - Device device = objectMapper.readValue(entry.getData(), Device.class); - onDeviceUpdated(msgType, device); + processDeviceCRUD(edgeEvent, msgType); break; case ASSET: - Asset asset = objectMapper.readValue(entry.getData(), Asset.class); - onAssetUpdated(msgType, asset); + processAssetCRUD(edgeEvent, msgType); break; case ENTITY_VIEW: - EntityView entityView = objectMapper.readValue(entry.getData(), EntityView.class); - onEntityViewUpdated(msgType, entityView); + processEntityViewCRUD(edgeEvent, msgType); break; case DASHBOARD: - Dashboard dashboard = objectMapper.readValue(entry.getData(), Dashboard.class); - onDashboardUpdated(msgType, dashboard); + processDashboardCRUD(edgeEvent, msgType); break; case RULE_CHAIN: - RuleChain ruleChain = objectMapper.readValue(entry.getData(), RuleChain.class); - onRuleChainUpdated(msgType, ruleChain); + processRuleChainCRUD(edgeEvent, msgType); break; case RULE_CHAIN_METADATA: - RuleChainMetaData ruleChainMetaData = objectMapper.readValue(entry.getData(), RuleChainMetaData.class); - onRuleChainMetadataUpdated(msgType, ruleChainMetaData); + processRuleChainMetadataCRUD(edgeEvent, msgType); break; case ALARM: - Alarm alarm = objectMapper.readValue(entry.getData(), Alarm.class); - onAlarmUpdated(msgType, alarm); + processAlarmCRUD(edgeEvent, msgType); break; case USER: - User user = objectMapper.readValue(entry.getData(), User.class); - onUserUpdated(msgType, user); + processUserCRUD(edgeEvent, msgType); break; case RELATION: - EntityRelation entityRelation = objectMapper.readValue(entry.getData(), EntityRelation.class); - onEntityRelationUpdated(msgType, entityRelation); + processRelationCRUD(edgeEvent, msgType); break; } } - private void updateQueueStartTs(Long newStartTs) { - newStartTs = ++newStartTs; // increments ts by 1 - next edge event search starts from current offset + 1 - List attributes = Collections.singletonList(new BaseAttributeKvEntry(new LongDataEntry(QUEUE_START_TS_ATTR_KEY, newStartTs), System.currentTimeMillis())); - ctx.getAttributesService().save(edge.getTenantId(), edge.getId(), DataConstants.SERVER_SCOPE, attributes); - } - - private ListenableFuture getQueueStartTs() { - ListenableFuture> future = - ctx.getAttributesService().find(edge.getTenantId(), edge.getId(), DataConstants.SERVER_SCOPE, QUEUE_START_TS_ATTR_KEY); - return Futures.transform(future, attributeKvEntryOpt -> { - if (attributeKvEntryOpt != null && attributeKvEntryOpt.isPresent()) { - AttributeKvEntry attributeKvEntry = attributeKvEntryOpt.get(); - return attributeKvEntry.getLongValue().isPresent() ? attributeKvEntry.getLongValue().get() : 0L; - } else { - return 0L; - } - }, MoreExecutors.directExecutor()); - } + private void processDeviceCRUD(EdgeEvent edgeEvent, UpdateMsgType msgType) { + DeviceId deviceId = new DeviceId(edgeEvent.getEntityId()); + ListenableFuture deviceFuture = ctx.getDeviceService().findDeviceByIdAsync(edgeEvent.getTenantId(), deviceId); + Futures.addCallback(deviceFuture, + new FutureCallback() { + @Override + public void onSuccess(@Nullable Device device) { + if (device != null) { + EntityUpdateMsg entityUpdateMsg = EntityUpdateMsg.newBuilder() + .setDeviceUpdateMsg(ctx.getDeviceUpdateMsgConstructor().constructDeviceUpdatedMsg(msgType, device)) + .build(); + outputStream.onNext(ResponseMsg.newBuilder() + .setEntityUpdateMsg(entityUpdateMsg) + .build()); + } + } - private void onEdgeUpdated(UpdateMsgType msgType, Edge edge) { - // TODO: voba add configuration update to edge - this.edge = edge; - } + @Override + public void onFailure(Throwable t) { + log.warn("Can't processDeviceCRUD, edgeEvent [{}]", edgeEvent, t); + } + }, ctx.getDbCallbackExecutor()); + } + + private void processAssetCRUD(EdgeEvent edgeEvent, UpdateMsgType msgType) { + AssetId assetId = new AssetId(edgeEvent.getEntityId()); + ListenableFuture assetFuture = ctx.getAssetService().findAssetByIdAsync(edgeEvent.getTenantId(), assetId); + Futures.addCallback(assetFuture, + new FutureCallback() { + @Override + public void onSuccess(@Nullable Asset asset) { + if (asset != null) { + EntityUpdateMsg entityUpdateMsg = EntityUpdateMsg.newBuilder() + .setAssetUpdateMsg(ctx.getAssetUpdateMsgConstructor().constructAssetUpdatedMsg(msgType, asset)) + .build(); + outputStream.onNext(ResponseMsg.newBuilder() + .setEntityUpdateMsg(entityUpdateMsg) + .build()); + } + } - private void onDeviceUpdated(UpdateMsgType msgType, Device device) { - EntityUpdateMsg entityUpdateMsg = EntityUpdateMsg.newBuilder() - .setDeviceUpdateMsg(ctx.getDeviceUpdateMsgConstructor().constructDeviceUpdatedMsg(msgType, device)) - .build(); - outputStream.onNext(ResponseMsg.newBuilder() - .setEntityUpdateMsg(entityUpdateMsg) - .build()); - } + @Override + public void onFailure(Throwable t) { + log.warn("Can't processAssetCRUD, edgeEvent [{}]", edgeEvent, t); + } + }, ctx.getDbCallbackExecutor()); + } + + private void processEntityViewCRUD(EdgeEvent edgeEvent, UpdateMsgType msgType) { + EntityViewId entityViewId = new EntityViewId(edgeEvent.getEntityId()); + ListenableFuture entityViewFuture = ctx.getEntityViewService().findEntityViewByIdAsync(edgeEvent.getTenantId(), entityViewId); + Futures.addCallback(entityViewFuture, + new FutureCallback() { + @Override + public void onSuccess(@Nullable EntityView entityView) { + if (entityView != null) { + EntityUpdateMsg entityUpdateMsg = EntityUpdateMsg.newBuilder() + .setEntityViewUpdateMsg(ctx.getEntityViewUpdateMsgConstructor().constructEntityViewUpdatedMsg(msgType, entityView)) + .build(); + outputStream.onNext(ResponseMsg.newBuilder() + .setEntityUpdateMsg(entityUpdateMsg) + .build()); + } + } - private void onAssetUpdated(UpdateMsgType msgType, Asset asset) { - EntityUpdateMsg entityUpdateMsg = EntityUpdateMsg.newBuilder() - .setAssetUpdateMsg(ctx.getAssetUpdateMsgConstructor().constructAssetUpdatedMsg(msgType, asset)) - .build(); - outputStream.onNext(ResponseMsg.newBuilder() - .setEntityUpdateMsg(entityUpdateMsg) - .build()); - } + @Override + public void onFailure(Throwable t) { + log.warn("Can't processEntityViewCRUD, edgeEvent [{}]", edgeEvent, t); + } + }, ctx.getDbCallbackExecutor()); + } + + private void processDashboardCRUD(EdgeEvent edgeEvent, UpdateMsgType msgType) { + DashboardId dashboardId = new DashboardId(edgeEvent.getEntityId()); + ListenableFuture dashboardFuture = ctx.getDashboardService().findDashboardByIdAsync(edgeEvent.getTenantId(), dashboardId); + Futures.addCallback(dashboardFuture, + new FutureCallback() { + @Override + public void onSuccess(@Nullable Dashboard dashboard) { + if (dashboard != null) { + EntityUpdateMsg entityUpdateMsg = EntityUpdateMsg.newBuilder() + .setDashboardUpdateMsg(ctx.getDashboardUpdateMsgConstructor().constructDashboardUpdatedMsg(msgType, dashboard)) + .build(); + outputStream.onNext(ResponseMsg.newBuilder() + .setEntityUpdateMsg(entityUpdateMsg) + .build()); + } + } - private void onEntityViewUpdated(UpdateMsgType msgType, EntityView entityView) { - EntityUpdateMsg entityUpdateMsg = EntityUpdateMsg.newBuilder() - .setEntityViewUpdateMsg(ctx.getEntityViewUpdateMsgConstructor().constructEntityViewUpdatedMsg(msgType, entityView)) - .build(); - outputStream.onNext(ResponseMsg.newBuilder() - .setEntityUpdateMsg(entityUpdateMsg) - .build()); - } + @Override + public void onFailure(Throwable t) { + log.warn("Can't processDashboardCRUD, edgeEvent [{}]", edgeEvent, t); + } + }, ctx.getDbCallbackExecutor()); + } + + private void processRuleChainCRUD(EdgeEvent edgeEvent, UpdateMsgType msgType) { + RuleChainId ruleChainId = new RuleChainId(edgeEvent.getEntityId()); + ListenableFuture ruleChainFuture = ctx.getRuleChainService().findRuleChainByIdAsync(edgeEvent.getTenantId(), ruleChainId); + Futures.addCallback(ruleChainFuture, + new FutureCallback() { + @Override + public void onSuccess(@Nullable RuleChain ruleChain) { + if (ruleChain != null) { + EntityUpdateMsg entityUpdateMsg = EntityUpdateMsg.newBuilder() + .setRuleChainUpdateMsg(ctx.getRuleChainUpdateMsgConstructor().constructRuleChainUpdatedMsg(edge.getRootRuleChainId(), msgType, ruleChain)) + .build(); + outputStream.onNext(ResponseMsg.newBuilder() + .setEntityUpdateMsg(entityUpdateMsg) + .build()); + } + } - private void onRuleChainUpdated(UpdateMsgType msgType, RuleChain ruleChain) { - EntityUpdateMsg entityUpdateMsg = EntityUpdateMsg.newBuilder() - .setRuleChainUpdateMsg(ctx.getRuleChainUpdateMsgConstructor().constructRuleChainUpdatedMsg(edge.getRootRuleChainId(), msgType, ruleChain)) - .build(); - outputStream.onNext(ResponseMsg.newBuilder() - .setEntityUpdateMsg(entityUpdateMsg) - .build()); - } + @Override + public void onFailure(Throwable t) { + log.warn("Can't processRuleChainCRUD, edgeEvent [{}]", edgeEvent, t); + } + }, ctx.getDbCallbackExecutor()); + } + + private void processRuleChainMetadataCRUD(EdgeEvent edgeEvent, UpdateMsgType msgType) { + RuleChainId ruleChainId = new RuleChainId(edgeEvent.getEntityId()); + ListenableFuture ruleChainFuture = ctx.getRuleChainService().findRuleChainByIdAsync(edgeEvent.getTenantId(), ruleChainId); + Futures.addCallback(ruleChainFuture, + new FutureCallback() { + @Override + public void onSuccess(@Nullable RuleChain ruleChain) { + if (ruleChain != null) { + RuleChainMetaData ruleChainMetaData = ctx.getRuleChainService().loadRuleChainMetaData(edgeEvent.getTenantId(), ruleChainId); + RuleChainMetadataUpdateMsg ruleChainMetadataUpdateMsg = + ctx.getRuleChainUpdateMsgConstructor().constructRuleChainMetadataUpdatedMsg(msgType, ruleChainMetaData); + if (ruleChainMetadataUpdateMsg != null) { + EntityUpdateMsg entityUpdateMsg = EntityUpdateMsg.newBuilder() + .setRuleChainMetadataUpdateMsg(ruleChainMetadataUpdateMsg) + .build(); + outputStream.onNext(ResponseMsg.newBuilder() + .setEntityUpdateMsg(entityUpdateMsg) + .build()); + } + } + } - private void onRuleChainMetadataUpdated(UpdateMsgType msgType, RuleChainMetaData ruleChainMetaData) { - RuleChainMetadataUpdateMsg ruleChainMetadataUpdateMsg = - ctx.getRuleChainUpdateMsgConstructor().constructRuleChainMetadataUpdatedMsg(msgType, ruleChainMetaData); - if (ruleChainMetadataUpdateMsg != null) { - EntityUpdateMsg entityUpdateMsg = EntityUpdateMsg.newBuilder() - .setRuleChainMetadataUpdateMsg(ruleChainMetadataUpdateMsg) - .build(); - outputStream.onNext(ResponseMsg.newBuilder() - .setEntityUpdateMsg(entityUpdateMsg) - .build()); - } - } + @Override + public void onFailure(Throwable t) { + log.warn("Can't processRuleChainMetadataCRUD, edgeEvent [{}]", edgeEvent, t); + } + }, ctx.getDbCallbackExecutor()); + } + + private void processUserCRUD(EdgeEvent edgeEvent, UpdateMsgType msgType) { + UserId userId = new UserId(edgeEvent.getEntityId()); + ListenableFuture userFuture = ctx.getUserService().findUserByIdAsync(edgeEvent.getTenantId(), userId); + Futures.addCallback(userFuture, + new FutureCallback() { + @Override + public void onSuccess(@Nullable User user) { + if (user != null) { + EntityUpdateMsg entityUpdateMsg = EntityUpdateMsg.newBuilder() + .setUserUpdateMsg(ctx.getUserUpdateMsgConstructor().constructUserUpdatedMsg(msgType, user)) + .build(); + outputStream.onNext(ResponseMsg.newBuilder() + .setEntityUpdateMsg(entityUpdateMsg) + .build()); + } + } - private void onDashboardUpdated(UpdateMsgType msgType, Dashboard dashboard) { - EntityUpdateMsg entityUpdateMsg = EntityUpdateMsg.newBuilder() - .setDashboardUpdateMsg(ctx.getDashboardUpdateMsgConstructor().constructDashboardUpdatedMsg(msgType, dashboard)) - .build(); - outputStream.onNext(ResponseMsg.newBuilder() - .setEntityUpdateMsg(entityUpdateMsg) - .build()); + @Override + public void onFailure(Throwable t) { + log.warn("Can't processUserCRUD, edgeEvent [{}]", edgeEvent, t); + } + }, ctx.getDbCallbackExecutor()); } - private void onAlarmUpdated(UpdateMsgType msgType, Alarm alarm) { + private void processRelationCRUD(EdgeEvent edgeEvent, UpdateMsgType msgType) { + EntityRelation entityRelation = objectMapper.convertValue(edgeEvent.getEntityBody(), EntityRelation.class); EntityUpdateMsg entityUpdateMsg = EntityUpdateMsg.newBuilder() - .setAlarmUpdateMsg(ctx.getAlarmUpdateMsgConstructor().constructAlarmUpdatedMsg(edge.getTenantId(), msgType, alarm)) + .setRelationUpdateMsg(ctx.getRelationUpdateMsgConstructor().constructRelationUpdatedMsg(msgType, entityRelation)) .build(); outputStream.onNext(ResponseMsg.newBuilder() .setEntityUpdateMsg(entityUpdateMsg) .build()); } - private void onUserUpdated(UpdateMsgType msgType, User user) { - EntityUpdateMsg entityUpdateMsg = EntityUpdateMsg.newBuilder() - .setUserUpdateMsg(ctx.getUserUpdateMsgConstructor().constructUserUpdatedMsg(msgType, user)) - .build(); - outputStream.onNext(ResponseMsg.newBuilder() - .setEntityUpdateMsg(entityUpdateMsg) - .build()); - } + private void processAlarmCRUD(EdgeEvent edgeEvent, UpdateMsgType msgType) { + AlarmId alarmId = new AlarmId(edgeEvent.getEntityId()); + ListenableFuture alarmFuture = ctx.getAlarmService().findAlarmByIdAsync(edgeEvent.getTenantId(), alarmId); + Futures.addCallback(alarmFuture, + new FutureCallback() { + @Override + public void onSuccess(@Nullable Alarm alarm) { + if (alarm != null) { + EntityUpdateMsg entityUpdateMsg = EntityUpdateMsg.newBuilder() + .setAlarmUpdateMsg(ctx.getAlarmUpdateMsgConstructor().constructAlarmUpdatedMsg(edge.getTenantId(), msgType, alarm)) + .build(); + outputStream.onNext(ResponseMsg.newBuilder() + .setEntityUpdateMsg(entityUpdateMsg) + .build()); + } + } - private void onEntityRelationUpdated(UpdateMsgType msgType, EntityRelation entityRelation) { - EntityUpdateMsg entityUpdateMsg = EntityUpdateMsg.newBuilder() - .setRelationUpdateMsg(ctx.getRelationUpdateMsgConstructor().constructRelationUpdatedMsg(msgType, entityRelation)) - .build(); - outputStream.onNext(ResponseMsg.newBuilder() - .setEntityUpdateMsg(entityUpdateMsg) - .build()); + @Override + public void onFailure(Throwable t) { + log.warn("Can't processAlarmCRUD, edgeEvent [{}]", edgeEvent, t); + } + }, ctx.getDbCallbackExecutor()); } private UpdateMsgType getResponseMsgType(ActionType actionType) { @@ -490,29 +588,10 @@ public final class EdgeGrpcSession implements Closeable { } } - private DownlinkMsg constructEntityDataProtoMsg(String entityName, EntityId entityId, TbMsg tbMsg) { - EntityDataProto entityData = EntityDataProto.newBuilder() - .setEntityName(entityName) - .setTbMsg(ByteString.copyFrom(TbMsg.toByteArray(tbMsg))) - .setEntityIdMSB(entityId.getId().getMostSignificantBits()) - .setEntityIdLSB(entityId.getId().getLeastSignificantBits()) - .build(); - + private DownlinkMsg constructEntityDataProtoMsg(EntityId entityId, ActionType actionType, JsonElement entityData) { + EntityDataProto entityDataProto = ctx.getEntityDataMsgConstructor().constructEntityDataMsg(entityId, actionType, entityData); DownlinkMsg.Builder builder = DownlinkMsg.newBuilder() - .addAllEntityData(Collections.singletonList(entityData)); - - return builder.build(); - } - - private CustomerUpdateMsg constructCustomerUpdatedMsg(UpdateMsgType msgType, Customer customer) { - CustomerUpdateMsg.Builder builder = CustomerUpdateMsg.newBuilder() - .setMsgType(msgType); - return builder.build(); - } - - private UserUpdateMsg constructUserUpdatedMsg(UpdateMsgType msgType, User user) { - UserUpdateMsg.Builder builder = UserUpdateMsg.newBuilder() - .setMsgType(msgType); + .addAllEntityData(Collections.singletonList(entityDataProto)); return builder.build(); } @@ -520,20 +599,21 @@ public final class EdgeGrpcSession implements Closeable { try { if (uplinkMsg.getEntityDataList() != null && !uplinkMsg.getEntityDataList().isEmpty()) { for (EntityDataProto entityData : uplinkMsg.getEntityDataList()) { - TbMsg tbMsg = null; - TbMsg originalTbMsg = TbMsg.fromBytes(entityData.getTbMsg().toByteArray(), TbMsgCallback.EMPTY); - if (originalTbMsg.getOriginator().getEntityType() == EntityType.DEVICE) { - String deviceName = entityData.getEntityName(); - Device device = ctx.getDeviceService().findDeviceByTenantIdAndName(edge.getTenantId(), deviceName); - if (device != null) { - tbMsg = TbMsg.newMsg(originalTbMsg.getType(), device.getId(), originalTbMsg.getMetaData().copy(), - originalTbMsg.getDataType(), originalTbMsg.getData()); - } - } else { - tbMsg = originalTbMsg; - } - if (tbMsg != null) { - ctx.getTbClusterService().pushMsgToRuleEngine(edge.getTenantId(), tbMsg.getOriginator(), tbMsg, null); + EntityId entityId = constructEntityId(entityData); + if ((entityData.hasPostAttributesMsg() || entityData.hasPostTelemetryMsg()) && entityId != null) { + ListenableFuture metaDataFuture = constructBaseMsgMetadata(entityId); + Futures.transform(metaDataFuture, metaData -> { + if (metaData != null) { + metaData.putValue(DataConstants.MSG_SOURCE_KEY, DataConstants.EDGE_MSG_SOURCE); + if (entityData.hasPostAttributesMsg()) { + processPostAttributes(entityId, entityData.getPostAttributesMsg(), metaData); + } + if (entityData.hasPostTelemetryMsg()) { + processPostTelemetry(entityId, entityData.getPostTelemetryMsg(), metaData); + } + } + return null; + }, ctx.getDbCallbackExecutor()); } } } @@ -559,6 +639,78 @@ public final class EdgeGrpcSession implements Closeable { return UplinkResponseMsg.newBuilder().setSuccess(true).build(); } + private ListenableFuture constructBaseMsgMetadata(EntityId entityId) { + switch (entityId.getEntityType()) { + case DEVICE: + ListenableFuture deviceFuture = ctx.getDeviceService().findDeviceByIdAsync(edge.getTenantId(), new DeviceId(entityId.getId())); + return Futures.transform(deviceFuture, device -> { + TbMsgMetaData metaData = new TbMsgMetaData(); + if (device != null) { + metaData.putValue("deviceName", device.getName()); + metaData.putValue("deviceType", device.getType()); + } + return metaData; + }, ctx.getDbCallbackExecutor()); + case ASSET: + ListenableFuture assetFuture = ctx.getAssetService().findAssetByIdAsync(edge.getTenantId(), new AssetId(entityId.getId())); + return Futures.transform(assetFuture, asset -> { + TbMsgMetaData metaData = new TbMsgMetaData(); + if (asset != null) { + metaData.putValue("assetName", asset.getName()); + metaData.putValue("assetType", asset.getType()); + } + return metaData; + }, ctx.getDbCallbackExecutor()); + case ENTITY_VIEW: + ListenableFuture entityViewFuture = ctx.getEntityViewService().findEntityViewByIdAsync(edge.getTenantId(), new EntityViewId(entityId.getId())); + return Futures.transform(entityViewFuture, entityView -> { + TbMsgMetaData metaData = new TbMsgMetaData(); + if (entityView != null) { + metaData.putValue("entityViewName", entityView.getName()); + metaData.putValue("entityViewType", entityView.getType()); + } + return metaData; + }, ctx.getDbCallbackExecutor()); + default: + log.debug("Constructing empty metadata for entityId [{}]", entityId); + return Futures.immediateFuture(new TbMsgMetaData()); + } + } + + private EntityId constructEntityId(EntityDataProto entityData) { + EntityType entityType = EntityType.valueOf(entityData.getEntityType()); + switch (entityType) { + case DEVICE: + return new DeviceId(new UUID(entityData.getEntityIdMSB(), entityData.getEntityIdLSB())); + case ASSET: + return new AssetId(new UUID(entityData.getEntityIdMSB(), entityData.getEntityIdLSB())); + case ENTITY_VIEW: + return new EntityViewId(new UUID(entityData.getEntityIdMSB(), entityData.getEntityIdLSB())); + case DASHBOARD: + return new DashboardId(new UUID(entityData.getEntityIdMSB(), entityData.getEntityIdLSB())); + default: + log.warn("Unsupported entity type [{}] during construct of entity id. EntityDataProto [{}]", entityData.getEntityType(), entityData); + return null; + } + } + + private void processPostTelemetry(EntityId entityId, TransportProtos.PostTelemetryMsg msg, TbMsgMetaData metaData) { + 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)); + // TODO: voba - verify that null callback is OK + ctx.getTbClusterService().pushMsgToRuleEngine(edge.getTenantId(), tbMsg.getOriginator(), tbMsg, null); + } + } + + private void processPostAttributes(EntityId entityId, TransportProtos.PostAttributeMsg msg, TbMsgMetaData metaData) { + JsonObject json = JsonUtils.getJsonObject(msg.getKvList()); + TbMsg tbMsg = TbMsg.newMsg(SessionMsgType.POST_ATTRIBUTES_REQUEST.name(), entityId, metaData, gson.toJson(json)); + // TODO: voba - verify that null callback is OK + ctx.getTbClusterService().pushMsgToRuleEngine(edge.getTenantId(), tbMsg.getOriginator(), tbMsg, null); + } + private void onDeviceUpdate(DeviceUpdateMsg deviceUpdateMsg) { log.info("onDeviceUpdate {}", deviceUpdateMsg); DeviceId edgeDeviceId = new DeviceId(new UUID(deviceUpdateMsg.getIdMSB(), deviceUpdateMsg.getIdLSB())); @@ -759,7 +911,7 @@ public final class EdgeGrpcSession implements Closeable { .setTenantIdLSB(edge.getTenantId().getId().getLeastSignificantBits()) .setName(edge.getName()) .setRoutingKey(edge.getRoutingKey()) - .setType(edge.getType().toString()) + .setType(edge.getType()) .build(); } diff --git a/application/src/main/java/org/thingsboard/server/service/edge/rpc/constructor/EntityDataMsgConstructor.java b/application/src/main/java/org/thingsboard/server/service/edge/rpc/constructor/EntityDataMsgConstructor.java new file mode 100644 index 0000000000..d9a06655c5 --- /dev/null +++ b/application/src/main/java/org/thingsboard/server/service/edge/rpc/constructor/EntityDataMsgConstructor.java @@ -0,0 +1,56 @@ +/** + * Copyright © 2016-2020 The Thingsboard Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.thingsboard.server.service.edge.rpc.constructor; + +import com.google.gson.JsonElement; +import lombok.extern.slf4j.Slf4j; +import org.springframework.stereotype.Component; +import org.thingsboard.server.common.data.audit.ActionType; +import org.thingsboard.server.common.data.id.EntityId; +import org.thingsboard.server.common.transport.adaptor.JsonConverter; +import org.thingsboard.server.gen.edge.EntityDataProto; + +@Component +@Slf4j +public class EntityDataMsgConstructor { + + public EntityDataProto constructEntityDataMsg(EntityId entityId, ActionType actionType, JsonElement entityData) { + EntityDataProto.Builder builder = EntityDataProto.newBuilder() + .setEntityIdMSB(entityId.getId().getMostSignificantBits()) + .setEntityIdLSB(entityId.getId().getLeastSignificantBits()) + .setEntityType(entityId.getEntityType().name()); + switch (actionType) { + case TIMESERIES_UPDATED: + try { + builder.setPostTelemetryMsg(JsonConverter.convertToTelemetryProto(entityData)); + } catch (Exception e) { + log.warn("Can't convert to telemetry proto, entityData [{}]", entityData, e); + } + break; + case ATTRIBUTES_UPDATED: + try { + builder.setPostAttributesMsg(JsonConverter.convertToAttributesProto(entityData)); + } catch (Exception e) { + log.warn("Can't convert to attributes proto, entityData [{}]", entityData, e); + } + break; + // TODO: voba - add support for attribute delete + // case ATTRIBUTES_DELETED: + } + return builder.build(); + } + +} diff --git a/common/data/src/main/java/org/thingsboard/server/common/data/DataConstants.java b/common/data/src/main/java/org/thingsboard/server/common/data/DataConstants.java index 01b9e86d25..ccb3b8765f 100644 --- a/common/data/src/main/java/org/thingsboard/server/common/data/DataConstants.java +++ b/common/data/src/main/java/org/thingsboard/server/common/data/DataConstants.java @@ -65,4 +65,7 @@ public class DataConstants { public static final String DEFAULT_SECRET_KEY = ""; public static final String SECRET_KEY_FIELD_NAME = "secretKey"; public static final String DURATION_MS_FIELD_NAME = "durationMs"; + + public static final String EDGE_MSG_SOURCE = "edge"; + public static final String MSG_SOURCE_KEY = "source"; } diff --git a/common/data/src/main/java/org/thingsboard/server/common/data/id/EntityIdFactory.java b/common/data/src/main/java/org/thingsboard/server/common/data/id/EntityIdFactory.java index 583b3b8e67..17d86e26ec 100644 --- a/common/data/src/main/java/org/thingsboard/server/common/data/id/EntityIdFactory.java +++ b/common/data/src/main/java/org/thingsboard/server/common/data/id/EntityIdFactory.java @@ -16,6 +16,7 @@ package org.thingsboard.server.common.data.id; import org.thingsboard.server.common.data.EntityType; +import org.thingsboard.server.common.data.edge.EdgeEventType; import java.util.UUID; @@ -68,4 +69,28 @@ public class EntityIdFactory { throw new IllegalArgumentException("EntityType " + type + " is not supported!"); } + public static EntityId getByEdgeEventTypeAndUuid(EdgeEventType edgeEventType, UUID uuid) { + switch (edgeEventType) { + case CUSTOMER: + return new CustomerId(uuid); + case USER: + return new UserId(uuid); + case DASHBOARD: + return new DashboardId(uuid); + case DEVICE: + return new DeviceId(uuid); + case ASSET: + return new AssetId(uuid); + case ALARM: + return new AlarmId(uuid); + case RULE_CHAIN: + return new RuleChainId(uuid); + case ENTITY_VIEW: + return new EntityViewId(uuid); + case EDGE: + return new EdgeId(uuid); + } + throw new IllegalArgumentException("EdgeEventType " + edgeEventType + " is not supported!"); + } + } diff --git a/common/edge-api/src/main/proto/edge.proto b/common/edge-api/src/main/proto/edge.proto index 787de743be..38cbe7092c 100644 --- a/common/edge-api/src/main/proto/edge.proto +++ b/common/edge-api/src/main/proto/edge.proto @@ -100,9 +100,9 @@ enum UpdateMsgType { } message EntityDataProto { - string entityName = 1; - int64 entityIdMSB = 2; - int64 entityIdLSB = 3; + int64 entityIdMSB = 1; + int64 entityIdLSB = 2; + string entityType = 3; transport.PostTelemetryMsg postTelemetryMsg = 4; transport.PostAttributeMsg postAttributesMsg = 5; // transport.ToDeviceRpcRequestMsg ??? diff --git a/dao/src/main/java/org/thingsboard/server/dao/edge/BaseEdgeEventService.java b/dao/src/main/java/org/thingsboard/server/dao/edge/BaseEdgeEventService.java index 134ffe3b06..fb8f3a03a4 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/edge/BaseEdgeEventService.java +++ b/dao/src/main/java/org/thingsboard/server/dao/edge/BaseEdgeEventService.java @@ -52,6 +52,8 @@ public class BaseEdgeEventService implements EdgeEventService { return EdgeEventType.DASHBOARD; case USER: return EdgeEventType.USER; + case ALARM: + return EdgeEventType.ALARM; default: log.warn("Failed to push notification to edge service. Unsupported entity type [{}]", entityType); return null; diff --git a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/edge/TbMsgPushToEdgeNode.java b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/edge/TbMsgPushToEdgeNode.java index 1e42224c70..ad757220b3 100644 --- a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/edge/TbMsgPushToEdgeNode.java +++ b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/edge/TbMsgPushToEdgeNode.java @@ -70,34 +70,23 @@ public class TbMsgPushToEdgeNode implements TbNode { @Override public void onMsg(TbContext ctx, TbMsg msg) { - if (EntityType.DEVICE.equals(msg.getOriginator().getEntityType()) || - EntityType.DEVICE.equals(msg.getOriginator().getEntityType()) || - EntityType.DEVICE.equals(msg.getOriginator().getEntityType()) || - EntityType.DEVICE.equals(msg.getOriginator().getEntityType())) { - if (SessionMsgType.POST_TELEMETRY_REQUEST.name().equals(msg.getType()) || - SessionMsgType.POST_ATTRIBUTES_REQUEST.name().equals(msg.getType()) || - DataConstants.ATTRIBUTES_UPDATED.equals(msg.getType()) || - DataConstants.ATTRIBUTES_DELETED.equals(msg.getType())) { + if (DataConstants.EDGE_MSG_SOURCE.equalsIgnoreCase(msg.getMetaData().getValue(DataConstants.MSG_SOURCE_KEY))) { + log.debug("Ignoring msg from the cloud, msg [{}]", msg); + return; + } + if (isSupportedOriginator(msg.getOriginator().getEntityType())) { + if (isSupportedMsgType(msg.getType())) { ListenableFuture getEdgeIdFuture = getEdgeIdByOriginatorId(ctx, ctx.getTenantId(), msg.getOriginator()); Futures.transform(getEdgeIdFuture, edgeId -> { EdgeEventType edgeEventTypeByEntityType = ctx.getEdgeEventService().getEdgeEventTypeByEntityType(msg.getOriginator().getEntityType()); if (edgeEventTypeByEntityType == null) { log.debug("Edge event type is null. Entity Type {}", msg.getOriginator().getEntityType()); - ctx.tellFailure(msg, new RuntimeException("Edge event type is null. Entity Type '"+ msg.getOriginator().getEntityType() + "'")); - } - ActionType actionType; - if (SessionMsgType.POST_TELEMETRY_REQUEST.name().equals(msg.getType())) { - actionType = ActionType.TIMESERIES_UPDATED; - } else if (SessionMsgType.POST_ATTRIBUTES_REQUEST.name().equals(msg.getType()) || - DataConstants.ATTRIBUTES_UPDATED.equals(msg.getType())) { - actionType = ActionType.ATTRIBUTES_UPDATED; - } else { - actionType = ActionType.ATTRIBUTES_DELETED; + ctx.tellFailure(msg, new RuntimeException("Edge event type is null. Entity Type '" + msg.getOriginator().getEntityType() + "'")); } EdgeEvent edgeEvent = new EdgeEvent(); edgeEvent.setTenantId(ctx.getTenantId()); edgeEvent.setEdgeId(edgeId); - edgeEvent.setEdgeEventAction(actionType.name()); + edgeEvent.setEdgeEventAction(getActionTypeByMsgType(msg.getType()).name()); edgeEvent.setEntityId(msg.getOriginator().getId()); edgeEvent.setEdgeEventType(edgeEventTypeByEntityType); edgeEvent.setEntityBody(json.valueToTree(msg.getData())); @@ -126,6 +115,42 @@ public class TbMsgPushToEdgeNode implements TbNode { } } + private ActionType getActionTypeByMsgType(String msgType) { + ActionType actionType; + if (SessionMsgType.POST_TELEMETRY_REQUEST.name().equals(msgType)) { + actionType = ActionType.TIMESERIES_UPDATED; + } else if (SessionMsgType.POST_ATTRIBUTES_REQUEST.name().equals(msgType) + || DataConstants.ATTRIBUTES_UPDATED.equals(msgType)) { + actionType = ActionType.ATTRIBUTES_UPDATED; + } else { + actionType = ActionType.ATTRIBUTES_DELETED; + } + return actionType; + } + + private boolean isSupportedOriginator(EntityType entityType) { + switch (entityType) { + case DEVICE: + case ASSET: + case ENTITY_VIEW: + case DASHBOARD: + return true; + default: + return false; + } + } + + private boolean isSupportedMsgType(String msgType) { + if (SessionMsgType.POST_TELEMETRY_REQUEST.name().equals(msgType) + || SessionMsgType.POST_ATTRIBUTES_REQUEST.name().equals(msgType) + || DataConstants.ATTRIBUTES_UPDATED.equals(msgType) + || DataConstants.ATTRIBUTES_DELETED.equals(msgType)) { + return true; + } else { + return false; + } + } + private ListenableFuture getEdgeIdByOriginatorId(TbContext ctx, TenantId tenantId, EntityId originatorId) { ListenableFuture> future = ctx.getRelationService().findByToAndTypeAsync(tenantId, originatorId, EntityRelation.CONTAINS_TYPE, RelationTypeGroup.EDGE); return Futures.transform(future, relations -> { diff --git a/ui/src/app/rulechain/rulechain.routes.js b/ui/src/app/rulechain/rulechain.routes.js index 2991000e58..08dc2f522e 100644 --- a/ui/src/app/rulechain/rulechain.routes.js +++ b/ui/src/app/rulechain/rulechain.routes.js @@ -37,7 +37,7 @@ export default function RuleChainRoutes($stateProvider, NodeTemplatePathProvider } }) .state('home.ruleChains.core', { - url: '/ruleChains/system', + url: '/ruleChains/core', params: {'topIndex': 0}, module: 'private', auth: ['SYS_ADMIN', 'TENANT_ADMIN'],