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 1e5121638b..57a5caa5bb 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 @@ -29,9 +29,13 @@ 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.service.edge.rpc.alarm.AlarmMetadataConstructor; +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.EntityViewUpdateMsgConstructor; import org.thingsboard.server.service.edge.rpc.init.InitEdgeService; -import org.thingsboard.server.service.edge.rpc.ruleChain.RuleChainMetadataConstructor; +import org.thingsboard.server.service.edge.rpc.constructor.RuleChainUpdateMsgConstructor; @Component @Data @@ -83,9 +87,25 @@ public class EdgeContextComponent { @Lazy @Autowired - private RuleChainMetadataConstructor ruleChainMetadataConstructor; + private RuleChainUpdateMsgConstructor ruleChainUpdateMsgConstructor; @Lazy @Autowired - private AlarmMetadataConstructor alarmMetadataConstructor; + private AlarmUpdateMsgConstructor alarmUpdateMsgConstructor; + + @Lazy + @Autowired + private DeviceUpdateMsgConstructor deviceUpdateMsgConstructor; + + @Lazy + @Autowired + private AssetUpdateMsgConstructor assetUpdateMsgConstructor; + + @Lazy + @Autowired + private EntityViewUpdateMsgConstructor entityViewUpdateMsgConstructor; + + @Lazy + @Autowired + private DashboardUpdateMsgConstructor dashboardUpdateMsgConstructor; } 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 5398495409..b542b55f08 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 @@ -29,7 +29,6 @@ import org.thingsboard.server.common.data.Customer; 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.User; @@ -37,7 +36,6 @@ import org.thingsboard.server.common.data.alarm.Alarm; import org.thingsboard.server.common.data.alarm.AlarmSeverity; import org.thingsboard.server.common.data.alarm.AlarmStatus; 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.EdgeQueueEntry; import org.thingsboard.server.common.data.id.AssetId; @@ -45,7 +43,6 @@ import org.thingsboard.server.common.data.id.CustomerId; 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.EntityIdFactory; import org.thingsboard.server.common.data.id.EntityViewId; import org.thingsboard.server.common.data.id.TenantId; import org.thingsboard.server.common.data.kv.AttributeKvEntry; @@ -56,48 +53,36 @@ import org.thingsboard.server.common.data.page.TimePageData; import org.thingsboard.server.common.data.page.TimePageLink; import org.thingsboard.server.common.data.relation.EntityRelation; import org.thingsboard.server.common.data.relation.RelationTypeGroup; -import org.thingsboard.server.common.data.rule.NodeConnectionInfo; import org.thingsboard.server.common.data.rule.RuleChain; -import org.thingsboard.server.common.data.rule.RuleChainConnectionInfo; import org.thingsboard.server.common.data.rule.RuleChainMetaData; -import org.thingsboard.server.common.data.rule.RuleNode; 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.cluster.SendToClusterMsg; import org.thingsboard.server.common.msg.session.SessionMsgType; import org.thingsboard.server.common.msg.system.ServiceToRuleEngineMsg; -import org.thingsboard.server.dao.util.mapping.JacksonUtil; import org.thingsboard.server.gen.edge.AlarmUpdateMsg; -import org.thingsboard.server.gen.edge.AssetUpdateMsg; 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.DashboardUpdateMsg; import org.thingsboard.server.gen.edge.DeviceUpdateMsg; import org.thingsboard.server.gen.edge.DownlinkMsg; import org.thingsboard.server.gen.edge.EdgeConfiguration; import org.thingsboard.server.gen.edge.EntityDataProto; import org.thingsboard.server.gen.edge.EntityUpdateMsg; -import org.thingsboard.server.gen.edge.EntityViewUpdateMsg; -import org.thingsboard.server.gen.edge.NodeConnectionInfoProto; import org.thingsboard.server.gen.edge.RequestMsg; import org.thingsboard.server.gen.edge.RequestMsgType; import org.thingsboard.server.gen.edge.ResponseMsg; -import org.thingsboard.server.gen.edge.RuleChainConnectionInfoProto; +import org.thingsboard.server.gen.edge.RuleChainMetadataRequestMsg; import org.thingsboard.server.gen.edge.RuleChainMetadataUpdateMsg; -import org.thingsboard.server.gen.edge.RuleChainUpdateMsg; -import org.thingsboard.server.gen.edge.RuleNodeProto; 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.service.edge.EdgeContextComponent; -import javax.swing.text.html.parser.Entity; import java.io.IOException; -import java.util.ArrayList; import java.util.Collections; import java.util.List; import java.util.Optional; @@ -108,7 +93,6 @@ import java.util.function.BiConsumer; import java.util.function.Consumer; import static org.thingsboard.server.gen.edge.UpdateMsgType.ENTITY_CREATED_RPC_MESSAGE; -import static org.thingsboard.server.gen.edge.UpdateMsgType.ENTITY_UPDATED_RPC_MESSAGE; @Slf4j @Data @@ -176,7 +160,6 @@ public final class EdgeGrpcSession implements Cloneable { }; } - void processHandleMessages() throws ExecutionException, InterruptedException { Long queueStartTs = getQueueStartTs().get(); // TODO: this 100 value must be changed properly @@ -184,7 +167,7 @@ public final class EdgeGrpcSession implements Cloneable { TimePageData pageData; UUID ifOffset = null; do { - pageData = ctx.getEdgeService().findQueueEvents(edge.getTenantId(), edge.getId(), pageLink); + pageData = ctx.getEdgeService().findQueueEvents(edge.getTenantId(), edge.getId(), pageLink); if (!pageData.getData().isEmpty()) { log.trace("[{}] [{}] event(s) are going to be processed.", this.sessionId, pageData.getData().size()); for (Event event : pageData.getData()) { @@ -363,13 +346,13 @@ public final class EdgeGrpcSession implements Cloneable { ListenableFuture> future = ctx.getAttributesService().find(edge.getTenantId(), edge.getId(), DataConstants.SERVER_SCOPE, "queueStartTs"); 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; - } - } ); + if (attributeKvEntryOpt != null && attributeKvEntryOpt.isPresent()) { + AttributeKvEntry attributeKvEntry = attributeKvEntryOpt.get(); + return attributeKvEntry.getLongValue().isPresent() ? attributeKvEntry.getLongValue().get() : 0L; + } else { + return 0L; + } + }); } private void onEdgeUpdated(UpdateMsgType msgType, Edge edge) { @@ -379,7 +362,7 @@ public final class EdgeGrpcSession implements Cloneable { private void onDeviceUpdated(UpdateMsgType msgType, Device device) { EntityUpdateMsg entityUpdateMsg = EntityUpdateMsg.newBuilder() - .setDeviceUpdateMsg(constructDeviceUpdatedMsg(msgType, device)) + .setDeviceUpdateMsg(ctx.getDeviceUpdateMsgConstructor().constructDeviceUpdatedMsg(msgType, device)) .build(); outputStream.onNext(ResponseMsg.newBuilder() .setEntityUpdateMsg(entityUpdateMsg) @@ -388,7 +371,7 @@ public final class EdgeGrpcSession implements Cloneable { private void onAssetUpdated(UpdateMsgType msgType, Asset asset) { EntityUpdateMsg entityUpdateMsg = EntityUpdateMsg.newBuilder() - .setAssetUpdateMsg(constructAssetUpdatedMsg(msgType, asset)) + .setAssetUpdateMsg(ctx.getAssetUpdateMsgConstructor().constructAssetUpdatedMsg(msgType, asset)) .build(); outputStream.onNext(ResponseMsg.newBuilder() .setEntityUpdateMsg(entityUpdateMsg) @@ -397,7 +380,7 @@ public final class EdgeGrpcSession implements Cloneable { private void onEntityViewUpdated(UpdateMsgType msgType, EntityView entityView) { EntityUpdateMsg entityUpdateMsg = EntityUpdateMsg.newBuilder() - .setEntityViewUpdateMsg(constructEntityViewUpdatedMsg(msgType, entityView)) + .setEntityViewUpdateMsg(ctx.getEntityViewUpdateMsgConstructor().constructEntityViewUpdatedMsg(msgType, entityView)) .build(); outputStream.onNext(ResponseMsg.newBuilder() .setEntityUpdateMsg(entityUpdateMsg) @@ -406,7 +389,7 @@ public final class EdgeGrpcSession implements Cloneable { private void onRuleChainUpdated(UpdateMsgType msgType, RuleChain ruleChain) { EntityUpdateMsg entityUpdateMsg = EntityUpdateMsg.newBuilder() - .setRuleChainUpdateMsg(ctx.getRuleChainMetadataConstructor().constructRuleChainUpdatedMsg(edge, msgType, ruleChain)) + .setRuleChainUpdateMsg(ctx.getRuleChainUpdateMsgConstructor().constructRuleChainUpdatedMsg(edge, msgType, ruleChain)) .build(); outputStream.onNext(ResponseMsg.newBuilder() .setEntityUpdateMsg(entityUpdateMsg) @@ -415,7 +398,7 @@ public final class EdgeGrpcSession implements Cloneable { private void onRuleChainMetadataUpdated(UpdateMsgType msgType, RuleChainMetaData ruleChainMetaData) { RuleChainMetadataUpdateMsg ruleChainMetadataUpdateMsg = - ctx.getRuleChainMetadataConstructor().constructRuleChainMetadataUpdatedMsg(msgType, ruleChainMetaData); + ctx.getRuleChainUpdateMsgConstructor().constructRuleChainMetadataUpdatedMsg(msgType, ruleChainMetaData); if (ruleChainMetadataUpdateMsg != null) { EntityUpdateMsg entityUpdateMsg = EntityUpdateMsg.newBuilder() .setRuleChainMetadataUpdateMsg(ruleChainMetadataUpdateMsg) @@ -428,7 +411,7 @@ public final class EdgeGrpcSession implements Cloneable { private void onDashboardUpdated(UpdateMsgType msgType, Dashboard dashboard) { EntityUpdateMsg entityUpdateMsg = EntityUpdateMsg.newBuilder() - .setDashboardUpdateMsg(constructDashboardUpdatedMsg(msgType, dashboard)) + .setDashboardUpdateMsg(ctx.getDashboardUpdateMsgConstructor().constructDashboardUpdatedMsg(msgType, dashboard)) .build(); outputStream.onNext(ResponseMsg.newBuilder() .setEntityUpdateMsg(entityUpdateMsg) @@ -437,7 +420,7 @@ public final class EdgeGrpcSession implements Cloneable { private void onAlarmUpdated(UpdateMsgType msgType, Alarm alarm) { EntityUpdateMsg entityUpdateMsg = EntityUpdateMsg.newBuilder() - .setAlarmUpdateMsg(ctx.getAlarmMetadataConstructor().constructAlarmUpdatedMsg(edge.getTenantId(), msgType, alarm)) + .setAlarmUpdateMsg(ctx.getAlarmUpdateMsgConstructor().constructAlarmUpdatedMsg(edge.getTenantId(), msgType, alarm)) .build(); outputStream.onNext(ResponseMsg.newBuilder() .setEntityUpdateMsg(entityUpdateMsg) @@ -481,17 +464,6 @@ public final class EdgeGrpcSession implements Cloneable { return builder.build(); } - private DashboardUpdateMsg constructDashboardUpdatedMsg(UpdateMsgType msgType, Dashboard dashboard) { - dashboard = ctx.getDashboardService().findDashboardById(edge.getTenantId(), dashboard.getId()); - DashboardUpdateMsg.Builder builder = DashboardUpdateMsg.newBuilder() - .setMsgType(msgType) - .setIdMSB(dashboard.getId().getId().getMostSignificantBits()) - .setIdLSB(dashboard.getId().getId().getLeastSignificantBits()) - .setTitle(dashboard.getTitle()) - .setConfiguration(JacksonUtil.toString(dashboard.getConfiguration())); - return builder.build(); - } - private CustomerUpdateMsg constructCustomerUpdatedMsg(UpdateMsgType msgType, Customer customer) { CustomerUpdateMsg.Builder builder = CustomerUpdateMsg.newBuilder() .setMsgType(msgType); @@ -504,47 +476,6 @@ public final class EdgeGrpcSession implements Cloneable { return builder.build(); } - private DeviceUpdateMsg constructDeviceUpdatedMsg(UpdateMsgType msgType, Device device) { - DeviceUpdateMsg.Builder builder = DeviceUpdateMsg.newBuilder() - .setMsgType(msgType) - .setName(device.getName()) - .setType(device.getType()); - return builder.build(); - } - - private AssetUpdateMsg constructAssetUpdatedMsg(UpdateMsgType msgType, Asset asset) { - AssetUpdateMsg.Builder builder = AssetUpdateMsg.newBuilder() - .setMsgType(msgType) - .setName(asset.getName()) - .setType(asset.getType()); - return builder.build(); - } - - private EntityViewUpdateMsg constructEntityViewUpdatedMsg(UpdateMsgType msgType, EntityView entityView) { - String relatedName; - String relatedType; - org.thingsboard.server.gen.edge.EntityType relatedEntityType; - if (entityView.getEntityId().getEntityType().equals(EntityType.DEVICE)) { - Device device = ctx.getDeviceService().findDeviceById(entityView.getTenantId(), new DeviceId(entityView.getEntityId().getId())); - relatedName = device.getName(); - relatedType = device.getType(); - relatedEntityType = org.thingsboard.server.gen.edge.EntityType.DEVICE; - } else { - Asset asset = ctx.getAssetService().findAssetById(entityView.getTenantId(), new AssetId(entityView.getEntityId().getId())); - relatedName = asset.getName(); - relatedType = asset.getType(); - relatedEntityType = org.thingsboard.server.gen.edge.EntityType.ASSET; - } - EntityViewUpdateMsg.Builder builder = EntityViewUpdateMsg.newBuilder() - .setMsgType(msgType) - .setName(entityView.getName()) - .setType(entityView.getType()) - .setRelatedName(relatedName) - .setRelatedType(relatedType) - .setRelatedEntityType(relatedEntityType); - return builder.build(); - } - private UplinkResponseMsg processUplinkMsg(UplinkMsg uplinkMsg) { try { if (uplinkMsg.getEntityDataList() != null && !uplinkMsg.getEntityDataList().isEmpty()) { @@ -594,11 +525,16 @@ public final class EdgeGrpcSession implements Cloneable { } } } - if (uplinkMsg.getAlarmUpdatemsgList() != null && !uplinkMsg.getAlarmUpdatemsgList().isEmpty()) { - for (AlarmUpdateMsg alarmUpdateMsg : uplinkMsg.getAlarmUpdatemsgList()) { + if (uplinkMsg.getAlarmUpdateMsgList() != null && !uplinkMsg.getAlarmUpdateMsgList().isEmpty()) { + for (AlarmUpdateMsg alarmUpdateMsg : uplinkMsg.getAlarmUpdateMsgList()) { onAlarmUpdate(alarmUpdateMsg); } } + if (uplinkMsg.getRuleChainMetadataRequestMsgList() != null && !uplinkMsg.getRuleChainMetadataRequestMsgList().isEmpty()) { + for (RuleChainMetadataRequestMsg ruleChainMetadataRequestMsg : uplinkMsg.getRuleChainMetadataRequestMsgList()) { + ctx.getInitEdgeService().initRuleChainMetadata(edge, ruleChainMetadataRequestMsg, outputStream); + } + } } catch (Exception e) { return UplinkResponseMsg.newBuilder().setSuccess(false).setErrorMsg(e.getMessage()).build(); } diff --git a/application/src/main/java/org/thingsboard/server/service/edge/rpc/alarm/AlarmMetadataConstructor.java b/application/src/main/java/org/thingsboard/server/service/edge/rpc/constructor/AlarmUpdateMsgConstructor.java similarity index 96% rename from application/src/main/java/org/thingsboard/server/service/edge/rpc/alarm/AlarmMetadataConstructor.java rename to application/src/main/java/org/thingsboard/server/service/edge/rpc/constructor/AlarmUpdateMsgConstructor.java index 51aa34e97f..a97e921734 100644 --- a/application/src/main/java/org/thingsboard/server/service/edge/rpc/alarm/AlarmMetadataConstructor.java +++ b/application/src/main/java/org/thingsboard/server/service/edge/rpc/constructor/AlarmUpdateMsgConstructor.java @@ -1,4 +1,4 @@ -package org.thingsboard.server.service.edge.rpc.alarm; +package org.thingsboard.server.service.edge.rpc.constructor; import lombok.extern.slf4j.Slf4j; import org.bouncycastle.jcajce.provider.symmetric.DES; @@ -19,7 +19,7 @@ import org.thingsboard.server.gen.edge.UpdateMsgType; @Component @Slf4j -public class AlarmMetadataConstructor { +public class AlarmUpdateMsgConstructor { @Autowired private DeviceService deviceService; diff --git a/application/src/main/java/org/thingsboard/server/service/edge/rpc/constructor/AssetUpdateMsgConstructor.java b/application/src/main/java/org/thingsboard/server/service/edge/rpc/constructor/AssetUpdateMsgConstructor.java new file mode 100644 index 0000000000..32b1e55e63 --- /dev/null +++ b/application/src/main/java/org/thingsboard/server/service/edge/rpc/constructor/AssetUpdateMsgConstructor.java @@ -0,0 +1,36 @@ +/** + * Copyright © 2016-2019 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 lombok.extern.slf4j.Slf4j; +import org.springframework.stereotype.Component; +import org.thingsboard.server.common.data.asset.Asset; +import org.thingsboard.server.gen.edge.AssetUpdateMsg; +import org.thingsboard.server.gen.edge.UpdateMsgType; + +@Component +@Slf4j +public class AssetUpdateMsgConstructor { + + public AssetUpdateMsg constructAssetUpdatedMsg(UpdateMsgType msgType, Asset asset) { + AssetUpdateMsg.Builder builder = AssetUpdateMsg.newBuilder() + .setMsgType(msgType) + .setName(asset.getName()) + .setType(asset.getType()); + return builder.build(); + } + +} diff --git a/application/src/main/java/org/thingsboard/server/service/edge/rpc/constructor/DashboardUpdateMsgConstructor.java b/application/src/main/java/org/thingsboard/server/service/edge/rpc/constructor/DashboardUpdateMsgConstructor.java new file mode 100644 index 0000000000..5dfd74412f --- /dev/null +++ b/application/src/main/java/org/thingsboard/server/service/edge/rpc/constructor/DashboardUpdateMsgConstructor.java @@ -0,0 +1,45 @@ +/** + * Copyright © 2016-2019 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 lombok.extern.slf4j.Slf4j; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.stereotype.Component; +import org.thingsboard.server.common.data.Dashboard; +import org.thingsboard.server.dao.dashboard.DashboardService; +import org.thingsboard.server.dao.util.mapping.JacksonUtil; +import org.thingsboard.server.gen.edge.DashboardUpdateMsg; +import org.thingsboard.server.gen.edge.UpdateMsgType; + +@Component +@Slf4j +public class DashboardUpdateMsgConstructor { + + @Autowired + private DashboardService dashboardService; + + public DashboardUpdateMsg constructDashboardUpdatedMsg(UpdateMsgType msgType, Dashboard dashboard) { + dashboard = dashboardService.findDashboardById(dashboard.getTenantId(), dashboard.getId()); + DashboardUpdateMsg.Builder builder = DashboardUpdateMsg.newBuilder() + .setMsgType(msgType) + .setIdMSB(dashboard.getId().getId().getMostSignificantBits()) + .setIdLSB(dashboard.getId().getId().getLeastSignificantBits()) + .setTitle(dashboard.getTitle()) + .setConfiguration(JacksonUtil.toString(dashboard.getConfiguration())); + return builder.build(); + } + +} diff --git a/application/src/main/java/org/thingsboard/server/service/edge/rpc/constructor/DeviceUpdateMsgConstructor.java b/application/src/main/java/org/thingsboard/server/service/edge/rpc/constructor/DeviceUpdateMsgConstructor.java new file mode 100644 index 0000000000..4d39cb9b76 --- /dev/null +++ b/application/src/main/java/org/thingsboard/server/service/edge/rpc/constructor/DeviceUpdateMsgConstructor.java @@ -0,0 +1,35 @@ +/** + * Copyright © 2016-2019 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 lombok.extern.slf4j.Slf4j; +import org.springframework.stereotype.Component; +import org.thingsboard.server.common.data.Device; +import org.thingsboard.server.gen.edge.DeviceUpdateMsg; +import org.thingsboard.server.gen.edge.UpdateMsgType; + +@Component +@Slf4j +public class DeviceUpdateMsgConstructor { + + public DeviceUpdateMsg constructDeviceUpdatedMsg(UpdateMsgType msgType, Device device) { + DeviceUpdateMsg.Builder builder = DeviceUpdateMsg.newBuilder() + .setMsgType(msgType) + .setName(device.getName()) + .setType(device.getType()); + return builder.build(); + } +} diff --git a/application/src/main/java/org/thingsboard/server/service/edge/rpc/constructor/EntityViewUpdateMsgConstructor.java b/application/src/main/java/org/thingsboard/server/service/edge/rpc/constructor/EntityViewUpdateMsgConstructor.java new file mode 100644 index 0000000000..44f13e4985 --- /dev/null +++ b/application/src/main/java/org/thingsboard/server/service/edge/rpc/constructor/EntityViewUpdateMsgConstructor.java @@ -0,0 +1,67 @@ +/** + * Copyright © 2016-2019 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 lombok.extern.slf4j.Slf4j; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.stereotype.Component; +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.asset.Asset; +import org.thingsboard.server.common.data.id.AssetId; +import org.thingsboard.server.common.data.id.DeviceId; +import org.thingsboard.server.dao.asset.AssetService; +import org.thingsboard.server.dao.device.DeviceService; +import org.thingsboard.server.gen.edge.EntityViewUpdateMsg; +import org.thingsboard.server.gen.edge.UpdateMsgType; + +@Component +@Slf4j +public class EntityViewUpdateMsgConstructor { + + @Autowired + private DeviceService deviceService; + + @Autowired + private AssetService assetService; + + public EntityViewUpdateMsg constructEntityViewUpdatedMsg(UpdateMsgType msgType, EntityView entityView) { + String relatedName; + String relatedType; + org.thingsboard.server.gen.edge.EntityType relatedEntityType; + if (entityView.getEntityId().getEntityType().equals(EntityType.DEVICE)) { + Device device = deviceService.findDeviceById(entityView.getTenantId(), new DeviceId(entityView.getEntityId().getId())); + relatedName = device.getName(); + relatedType = device.getType(); + relatedEntityType = org.thingsboard.server.gen.edge.EntityType.DEVICE; + } else { + Asset asset = assetService.findAssetById(entityView.getTenantId(), new AssetId(entityView.getEntityId().getId())); + relatedName = asset.getName(); + relatedType = asset.getType(); + relatedEntityType = org.thingsboard.server.gen.edge.EntityType.ASSET; + } + EntityViewUpdateMsg.Builder builder = EntityViewUpdateMsg.newBuilder() + .setMsgType(msgType) + .setName(entityView.getName()) + .setType(entityView.getType()) + .setRelatedName(relatedName) + .setRelatedType(relatedType) + .setRelatedEntityType(relatedEntityType); + return builder.build(); + } + +} diff --git a/application/src/main/java/org/thingsboard/server/service/edge/rpc/ruleChain/RuleChainMetadataConstructor.java b/application/src/main/java/org/thingsboard/server/service/edge/rpc/constructor/RuleChainUpdateMsgConstructor.java similarity index 98% rename from application/src/main/java/org/thingsboard/server/service/edge/rpc/ruleChain/RuleChainMetadataConstructor.java rename to application/src/main/java/org/thingsboard/server/service/edge/rpc/constructor/RuleChainUpdateMsgConstructor.java index 1259d6fb41..ca3c579f60 100644 --- a/application/src/main/java/org/thingsboard/server/service/edge/rpc/ruleChain/RuleChainMetadataConstructor.java +++ b/application/src/main/java/org/thingsboard/server/service/edge/rpc/constructor/RuleChainUpdateMsgConstructor.java @@ -1,4 +1,4 @@ -package org.thingsboard.server.service.edge.rpc.ruleChain; +package org.thingsboard.server.service.edge.rpc.constructor; import com.fasterxml.jackson.core.JsonProcessingException; import com.fasterxml.jackson.databind.ObjectMapper; @@ -23,7 +23,7 @@ import java.util.List; @Component @Slf4j -public class RuleChainMetadataConstructor { +public class RuleChainUpdateMsgConstructor { private static final ObjectMapper objectMapper = new ObjectMapper(); @@ -111,7 +111,6 @@ public class RuleChainMetadataConstructor { } - private RuleNodeProto constructNode(RuleNode node) throws JsonProcessingException { return RuleNodeProto.newBuilder() .setIdMSB(node.getId().getId().getMostSignificantBits()) diff --git a/application/src/main/java/org/thingsboard/server/service/edge/rpc/init/DefaultInitEdgeService.java b/application/src/main/java/org/thingsboard/server/service/edge/rpc/init/DefaultInitEdgeService.java index aca568a78a..61a8cc1245 100644 --- a/application/src/main/java/org/thingsboard/server/service/edge/rpc/init/DefaultInitEdgeService.java +++ b/application/src/main/java/org/thingsboard/server/service/edge/rpc/init/DefaultInitEdgeService.java @@ -1,21 +1,59 @@ +/** + * Copyright © 2016-2019 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.init; import io.grpc.stub.StreamObserver; import lombok.extern.slf4j.Slf4j; 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.DashboardInfo; +import org.thingsboard.server.common.data.Device; +import org.thingsboard.server.common.data.EntityView; +import org.thingsboard.server.common.data.asset.Asset; import org.thingsboard.server.common.data.edge.Edge; +import org.thingsboard.server.common.data.id.RuleChainId; +import org.thingsboard.server.common.data.page.TextPageData; +import org.thingsboard.server.common.data.page.TextPageLink; import org.thingsboard.server.common.data.page.TimePageData; import org.thingsboard.server.common.data.page.TimePageLink; import org.thingsboard.server.common.data.rule.RuleChain; import org.thingsboard.server.common.data.rule.RuleChainMetaData; +import org.thingsboard.server.dao.asset.AssetService; +import org.thingsboard.server.dao.dashboard.DashboardService; +import org.thingsboard.server.dao.device.DeviceService; +import org.thingsboard.server.dao.entityview.EntityViewService; import org.thingsboard.server.dao.rule.RuleChainService; +import org.thingsboard.server.gen.edge.AssetUpdateMsg; +import org.thingsboard.server.gen.edge.DashboardUpdateMsg; +import org.thingsboard.server.gen.edge.DeviceUpdateMsg; import org.thingsboard.server.gen.edge.EntityUpdateMsg; +import org.thingsboard.server.gen.edge.EntityViewUpdateMsg; import org.thingsboard.server.gen.edge.ResponseMsg; +import org.thingsboard.server.gen.edge.RuleChainMetadataRequestMsg; import org.thingsboard.server.gen.edge.RuleChainMetadataUpdateMsg; import org.thingsboard.server.gen.edge.RuleChainUpdateMsg; import org.thingsboard.server.gen.edge.UpdateMsgType; -import org.thingsboard.server.service.edge.rpc.ruleChain.RuleChainMetadataConstructor; +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.EntityViewUpdateMsgConstructor; +import org.thingsboard.server.service.edge.rpc.constructor.RuleChainUpdateMsgConstructor; + +import java.util.UUID; @Service @Slf4j @@ -25,11 +63,160 @@ public class DefaultInitEdgeService implements InitEdgeService { private RuleChainService ruleChainService; @Autowired - private RuleChainMetadataConstructor ruleChainMetadataConstructor; + private DeviceService deviceService; + + @Autowired + private AssetService assetService; + + @Autowired + private EntityViewService entityViewService; + + @Autowired + private DashboardService dashboardService; + + @Autowired + private RuleChainUpdateMsgConstructor ruleChainUpdateMsgConstructor; + + @Autowired + private DeviceUpdateMsgConstructor deviceUpdateMsgConstructor; + + @Autowired + private AssetUpdateMsgConstructor assetUpdateMsgConstructor; + + @Autowired + private EntityViewUpdateMsgConstructor entityViewUpdateMsgConstructor; + + @Autowired + private DashboardUpdateMsgConstructor dashboardUpdateMsgConstructor; @Override public void init(Edge edge, StreamObserver outputStream) { initRuleChains(edge, outputStream); + initDevices(edge, outputStream); + initAssets(edge, outputStream); + initEntityViews(edge, outputStream); + initDashboards(edge, outputStream); + } + + private void initDevices(Edge edge, StreamObserver outputStream) { + try { + TextPageLink pageLink = new TextPageLink(100); + TextPageData pageData; + do { + pageData = deviceService.findDevicesByTenantIdAndEdgeId(edge.getTenantId(), edge.getId(), pageLink); + if (!pageData.getData().isEmpty()) { + log.trace("[{}] [{}] device(s) are going to be pushed to edge.", edge.getId(), pageData.getData().size()); + for (Device device : pageData.getData()) { + DeviceUpdateMsg deviceUpdateMsg = + deviceUpdateMsgConstructor.constructDeviceUpdatedMsg( + UpdateMsgType.ENTITY_CREATED_RPC_MESSAGE, + device); + EntityUpdateMsg entityUpdateMsg = EntityUpdateMsg.newBuilder() + .setDeviceUpdateMsg(deviceUpdateMsg) + .build(); + outputStream.onNext(ResponseMsg.newBuilder() + .setEntityUpdateMsg(entityUpdateMsg) + .build()); + } + } + if (pageData.hasNext()) { + pageLink = pageData.getNextPageLink(); + } + } while (pageData.hasNext()); + } catch (Exception e) { + log.error("Exception during loading edge device(s) on init!"); + } + } + + private void initAssets(Edge edge, StreamObserver outputStream) { + try { + TextPageLink pageLink = new TextPageLink(100); + TextPageData pageData; + do { + pageData = assetService.findAssetsByTenantIdAndEdgeId(edge.getTenantId(), edge.getId(), pageLink); + if (!pageData.getData().isEmpty()) { + log.trace("[{}] [{}] asset(s) are going to be pushed to edge.", edge.getId(), pageData.getData().size()); + for (Asset asset : pageData.getData()) { + AssetUpdateMsg assetUpdateMsg = + assetUpdateMsgConstructor.constructAssetUpdatedMsg( + UpdateMsgType.ENTITY_CREATED_RPC_MESSAGE, + asset); + EntityUpdateMsg entityUpdateMsg = EntityUpdateMsg.newBuilder() + .setAssetUpdateMsg(assetUpdateMsg) + .build(); + outputStream.onNext(ResponseMsg.newBuilder() + .setEntityUpdateMsg(entityUpdateMsg) + .build()); + } + } + if (pageData.hasNext()) { + pageLink = pageData.getNextPageLink(); + } + } while (pageData.hasNext()); + } catch (Exception e) { + log.error("Exception during loading edge asset(s) on init!"); + } + } + + private void initEntityViews(Edge edge, StreamObserver outputStream) { + try { + TextPageLink pageLink = new TextPageLink(100); + TextPageData pageData; + do { + pageData = entityViewService.findEntityViewsByTenantIdAndEdgeId(edge.getTenantId(), edge.getId(), pageLink); + if (!pageData.getData().isEmpty()) { + log.trace("[{}] [{}] entity view(s) are going to be pushed to edge.", edge.getId(), pageData.getData().size()); + for (EntityView entityView : pageData.getData()) { + EntityViewUpdateMsg entityViewUpdateMsg = + entityViewUpdateMsgConstructor.constructEntityViewUpdatedMsg( + UpdateMsgType.ENTITY_CREATED_RPC_MESSAGE, + entityView); + EntityUpdateMsg entityUpdateMsg = EntityUpdateMsg.newBuilder() + .setEntityViewUpdateMsg(entityViewUpdateMsg) + .build(); + outputStream.onNext(ResponseMsg.newBuilder() + .setEntityUpdateMsg(entityUpdateMsg) + .build()); + } + } + if (pageData.hasNext()) { + pageLink = pageData.getNextPageLink(); + } + } while (pageData.hasNext()); + } catch (Exception e) { + log.error("Exception during loading edge entity view(s) on init!"); + } + } + + private void initDashboards(Edge edge, StreamObserver outputStream) { + try { + TimePageLink pageLink = new TimePageLink(100); + TimePageData pageData; + do { + pageData = dashboardService.findDashboardsByTenantIdAndEdgeId(edge.getTenantId(), edge.getId(), pageLink).get(); + if (!pageData.getData().isEmpty()) { + log.trace("[{}] [{}] dashboard(s) are going to be pushed to edge.", edge.getId(), pageData.getData().size()); + for (DashboardInfo dashboardInfo : pageData.getData()) { + Dashboard dashboard = dashboardService.findDashboardById(edge.getTenantId(), dashboardInfo.getId()); + DashboardUpdateMsg dashboardUpdateMsg = + dashboardUpdateMsgConstructor.constructDashboardUpdatedMsg( + UpdateMsgType.ENTITY_CREATED_RPC_MESSAGE, + dashboard); + EntityUpdateMsg entityUpdateMsg = EntityUpdateMsg.newBuilder() + .setDashboardUpdateMsg(dashboardUpdateMsg) + .build(); + outputStream.onNext(ResponseMsg.newBuilder() + .setEntityUpdateMsg(entityUpdateMsg) + .build()); + } + } + if (pageData.hasNext()) { + pageLink = pageData.getNextPageLink(); + } + } while (pageData.hasNext()); + } catch (Exception e) { + log.error("Exception during loading edge dashboard(s) on init!"); + } } private void initRuleChains(Edge edge, StreamObserver outputStream) { @@ -42,7 +229,7 @@ public class DefaultInitEdgeService implements InitEdgeService { log.trace("[{}] [{}] rule chains(s) are going to be pushed to edge.", edge.getId(), pageData.getData().size()); for (RuleChain ruleChain : pageData.getData()) { RuleChainUpdateMsg ruleChainUpdateMsg = - ruleChainMetadataConstructor.constructRuleChainUpdatedMsg( + ruleChainUpdateMsgConstructor.constructRuleChainUpdatedMsg( edge, UpdateMsgType.ENTITY_UPDATED_RPC_MESSAGE, ruleChain); @@ -52,20 +239,6 @@ public class DefaultInitEdgeService implements InitEdgeService { outputStream.onNext(ResponseMsg.newBuilder() .setEntityUpdateMsg(entityUpdateMsg) .build()); - - RuleChainMetaData ruleChainMetaData = ruleChainService.loadRuleChainMetaData(edge.getTenantId(), ruleChain.getId()); - RuleChainMetadataUpdateMsg ruleChainMetadataUpdateMsg = - ruleChainMetadataConstructor.constructRuleChainMetadataUpdatedMsg( - UpdateMsgType.ENTITY_UPDATED_RPC_MESSAGE, - ruleChainMetaData); - if (ruleChainMetadataUpdateMsg != null) { - entityUpdateMsg = EntityUpdateMsg.newBuilder() - .setRuleChainMetadataUpdateMsg(ruleChainMetadataUpdateMsg) - .build(); - outputStream.onNext(ResponseMsg.newBuilder() - .setEntityUpdateMsg(entityUpdateMsg) - .build()); - } } } if (pageData.hasNext()) { @@ -73,8 +246,27 @@ public class DefaultInitEdgeService implements InitEdgeService { } } while (pageData.hasNext()); } catch (Exception e) { - log.error("Exception during loading edge rule chains on init!"); + log.error("Exception during loading edge rule chain(s) on init!"); } + } + @Override + public void initRuleChainMetadata(Edge edge, RuleChainMetadataRequestMsg ruleChainMetadataRequestMsg, StreamObserver outputStream) { + if (ruleChainMetadataRequestMsg.getRuleChainIdMSB() != 0 && ruleChainMetadataRequestMsg.getRuleChainIdLSB() != 0) { + RuleChainId ruleChainId = new RuleChainId(new UUID(ruleChainMetadataRequestMsg.getRuleChainIdMSB(), ruleChainMetadataRequestMsg.getRuleChainIdLSB())); + RuleChainMetaData ruleChainMetaData = ruleChainService.loadRuleChainMetaData(edge.getTenantId(), ruleChainId); + RuleChainMetadataUpdateMsg ruleChainMetadataUpdateMsg = + ruleChainUpdateMsgConstructor.constructRuleChainMetadataUpdatedMsg( + UpdateMsgType.ENTITY_UPDATED_RPC_MESSAGE, + ruleChainMetaData); + if (ruleChainMetadataUpdateMsg != null) { + EntityUpdateMsg entityUpdateMsg = EntityUpdateMsg.newBuilder() + .setRuleChainMetadataUpdateMsg(ruleChainMetadataUpdateMsg) + .build(); + outputStream.onNext(ResponseMsg.newBuilder() + .setEntityUpdateMsg(entityUpdateMsg) + .build()); + } + } } } diff --git a/application/src/main/java/org/thingsboard/server/service/edge/rpc/init/InitEdgeService.java b/application/src/main/java/org/thingsboard/server/service/edge/rpc/init/InitEdgeService.java index be303b947a..03ab84a06a 100644 --- a/application/src/main/java/org/thingsboard/server/service/edge/rpc/init/InitEdgeService.java +++ b/application/src/main/java/org/thingsboard/server/service/edge/rpc/init/InitEdgeService.java @@ -1,10 +1,28 @@ +/** + * Copyright © 2016-2019 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.init; import io.grpc.stub.StreamObserver; import org.thingsboard.server.common.data.edge.Edge; import org.thingsboard.server.gen.edge.ResponseMsg; +import org.thingsboard.server.gen.edge.RuleChainMetadataRequestMsg; public interface InitEdgeService { void init(Edge edge, StreamObserver outputStream); + + void initRuleChainMetadata(Edge edge, RuleChainMetadataRequestMsg ruleChainMetadataRequestMsg, StreamObserver outputStream); }