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

Push edge entities to edge on edge connect

pull/2818/head
Volodymyr Babak 6 years ago
parent
69ba71be27
commit
58cdc63abe
10 changed files with 462 additions and 114 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Split View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 28
      application/src/main/java/org/thingsboard/server/service/edge/EdgeContextComponent.java
  2. 110
      application/src/main/java/org/thingsboard/server/service/edge/rpc/EdgeGrpcSession.java
  3. 4
      application/src/main/java/org/thingsboard/server/service/edge/rpc/constructor/AlarmUpdateMsgConstructor.java
  4. 36
      application/src/main/java/org/thingsboard/server/service/edge/rpc/constructor/AssetUpdateMsgConstructor.java
  5. 45
      application/src/main/java/org/thingsboard/server/service/edge/rpc/constructor/DashboardUpdateMsgConstructor.java
  6. 35
      application/src/main/java/org/thingsboard/server/service/edge/rpc/constructor/DeviceUpdateMsgConstructor.java
  7. 67
      application/src/main/java/org/thingsboard/server/service/edge/rpc/constructor/EntityViewUpdateMsgConstructor.java
  8. 5
      application/src/main/java/org/thingsboard/server/service/edge/rpc/constructor/RuleChainUpdateMsgConstructor.java
  9. 228
      application/src/main/java/org/thingsboard/server/service/edge/rpc/init/DefaultInitEdgeService.java
  10. 18
      application/src/main/java/org/thingsboard/server/service/edge/rpc/init/InitEdgeService.java

28
application/src/main/java/org/thingsboard/server/service/edge/EdgeContextComponent.java
View File

@ -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;
}

110
application/src/main/java/org/thingsboard/server/service/edge/rpc/EdgeGrpcSession.java
View File

@ -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<Event> 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<Optional<AttributeKvEntry>> 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();
}

4
application/src/main/java/org/thingsboard/server/service/edge/rpc/alarm/AlarmMetadataConstructor.java → application/src/main/java/org/thingsboard/server/service/edge/rpc/constructor/AlarmUpdateMsgConstructor.java
View File

@ -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;

36
application/src/main/java/org/thingsboard/server/service/edge/rpc/constructor/AssetUpdateMsgConstructor.java
View File

@ -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();
}
}

45
application/src/main/java/org/thingsboard/server/service/edge/rpc/constructor/DashboardUpdateMsgConstructor.java
View File

@ -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();
}
}

35
application/src/main/java/org/thingsboard/server/service/edge/rpc/constructor/DeviceUpdateMsgConstructor.java
View File

@ -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();
}
}

67
application/src/main/java/org/thingsboard/server/service/edge/rpc/constructor/EntityViewUpdateMsgConstructor.java
View File

@ -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();
}
}

5
application/src/main/java/org/thingsboard/server/service/edge/rpc/ruleChain/RuleChainMetadataConstructor.java → application/src/main/java/org/thingsboard/server/service/edge/rpc/constructor/RuleChainUpdateMsgConstructor.java
View File

@ -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())

228
application/src/main/java/org/thingsboard/server/service/edge/rpc/init/DefaultInitEdgeService.java
View File

@ -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<ResponseMsg> outputStream) {
initRuleChains(edge, outputStream);
initDevices(edge, outputStream);
initAssets(edge, outputStream);
initEntityViews(edge, outputStream);
initDashboards(edge, outputStream);
}
private void initDevices(Edge edge, StreamObserver<ResponseMsg> outputStream) {
try {
TextPageLink pageLink = new TextPageLink(100);
TextPageData<Device> 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<ResponseMsg> outputStream) {
try {
TextPageLink pageLink = new TextPageLink(100);
TextPageData<Asset> 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<ResponseMsg> outputStream) {
try {
TextPageLink pageLink = new TextPageLink(100);
TextPageData<EntityView> 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<ResponseMsg> outputStream) {
try {
TimePageLink pageLink = new TimePageLink(100);
TimePageData<DashboardInfo> 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<ResponseMsg> 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<ResponseMsg> 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());
}
}
}
}

18
application/src/main/java/org/thingsboard/server/service/edge/rpc/init/InitEdgeService.java
View File

@ -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<ResponseMsg> outputStream);
void initRuleChainMetadata(Edge edge, RuleChainMetadataRequestMsg ruleChainMetadataRequestMsg, StreamObserver<ResponseMsg> outputStream);
}

Write Preview
Loading…
Cancel
Save