简柏林 / thingskit · Commits

Commit acfaf68042db6332d040fa60dfb84c8bce7f215d

Authored by Volodymyr Babak
1 parent c99cf51c

Added ALARM entity support

Inline Side-by-side
Showing 8 changed files with 227 additions and 136 deletions
... ... @@ -350,6 +350,8 @@ public class RuleChainActorMessageProcessor extends ComponentMsgProcessor<RuleCh
350 350 case DataConstants.ENTITY_DELETED:
351 351 case DataConstants.ENTITY_ASSIGNED_TO_EDGE:
352 352 case DataConstants.ENTITY_UNASSIGNED_FROM_EDGE:
  353 + case DataConstants.ALARM_ACK:
  354 + case DataConstants.ALARM_CLEAR:
353 355 edgeService.pushEventToEdge(tenantId, msg, new FutureCallback<Void>() {
354 356 @Override
355 357 public void onSuccess(@Nullable Void aVoid) {
... ...
... ... @@ -25,20 +25,9 @@ import org.springframework.beans.factory.annotation.Autowired;
25 25 import org.springframework.beans.factory.annotation.Value;
26 26 import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
27 27 import org.springframework.stereotype.Service;
28   -import org.thingsboard.server.actors.service.ActorService;
29   -import org.thingsboard.server.common.data.Event;
30   -import org.thingsboard.server.common.data.edge.Edge;
31   -import org.thingsboard.server.common.data.id.DeviceId;
32 28 import org.thingsboard.server.common.data.id.EdgeId;
  29 +import org.thingsboard.server.common.data.id.RuleChainId;
33 30 import org.thingsboard.server.common.data.id.TenantId;
34   -import org.thingsboard.server.common.data.page.TimePageData;
35   -import org.thingsboard.server.common.data.page.TimePageLink;
36   -import org.thingsboard.server.dao.asset.AssetService;
37   -import org.thingsboard.server.dao.attributes.AttributesService;
38   -import org.thingsboard.server.dao.device.DeviceService;
39   -import org.thingsboard.server.dao.edge.EdgeService;
40   -import org.thingsboard.server.dao.entityview.EntityViewService;
41   -import org.thingsboard.server.dao.event.EventService;
42 31 import org.thingsboard.server.gen.edge.EdgeRpcServiceGrpc;
43 32 import org.thingsboard.server.gen.edge.RequestMsg;
44 33 import org.thingsboard.server.gen.edge.ResponseMsg;
... ... @@ -52,7 +41,6 @@ import java.util.Map;
52 41 import java.util.concurrent.ConcurrentHashMap;
53 42 import java.util.concurrent.ExecutorService;
54 43 import java.util.concurrent.Executors;
55   -import java.util.concurrent.TimeUnit;
56 44
57 45 @Service
58 46 @Slf4j
... ... @@ -74,24 +62,6 @@ public class EdgeGrpcService extends EdgeRpcServiceGrpc.EdgeRpcServiceImplBase {
74 62 @Autowired
75 63 private EdgeContextComponent ctx;
76 64
77   - @Autowired
78   - private EdgeService edgeService;
79   -
80   - @Autowired
81   - private AssetService assetService;
82   -
83   - @Autowired
84   - private DeviceService deviceService;
85   -
86   - @Autowired
87   - private EntityViewService entityViewService;
88   -
89   - @Autowired
90   - private AttributesService attributesService;
91   -
92   - @Autowired
93   - private ActorService actorService;
94   -
95 65 private Server server;
96 66
97 67 private ExecutorService executor;
... ... @@ -156,4 +126,5 @@ public class EdgeGrpcService extends EdgeRpcServiceGrpc.EdgeRpcServiceImplBase {
156 126 private void onEdgeDisconnect(EdgeId edgeId) {
157 127 sessions.remove(edgeId);
158 128 }
  129 +
159 130 }
... ...
... ... @@ -31,6 +31,7 @@ import org.thingsboard.server.common.data.Device;
31 31 import org.thingsboard.server.common.data.EntityType;
32 32 import org.thingsboard.server.common.data.EntityView;
33 33 import org.thingsboard.server.common.data.Event;
  34 +import org.thingsboard.server.common.data.alarm.Alarm;
34 35 import org.thingsboard.server.common.data.asset.Asset;
35 36 import org.thingsboard.server.common.data.edge.Edge;
36 37 import org.thingsboard.server.common.data.edge.EdgeQueueEntry;
... ... @@ -59,6 +60,7 @@ import org.thingsboard.server.common.msg.cluster.SendToClusterMsg;
59 60 import org.thingsboard.server.common.msg.session.SessionMsgType;
60 61 import org.thingsboard.server.common.msg.system.ServiceToRuleEngineMsg;
61 62 import org.thingsboard.server.dao.util.mapping.JacksonUtil;
  63 +import org.thingsboard.server.gen.edge.AlarmUpdateMsg;
62 64 import org.thingsboard.server.gen.edge.AssetUpdateMsg;
63 65 import org.thingsboard.server.gen.edge.ConnectRequestMsg;
64 66 import org.thingsboard.server.gen.edge.ConnectResponseCode;
... ... @@ -68,6 +70,7 @@ import org.thingsboard.server.gen.edge.DeviceUpdateMsg;
68 70 import org.thingsboard.server.gen.edge.DownlinkMsg;
69 71 import org.thingsboard.server.gen.edge.EdgeConfiguration;
70 72 import org.thingsboard.server.gen.edge.EntityDataProto;
  73 +import org.thingsboard.server.gen.edge.EntityUpdateMsg;
71 74 import org.thingsboard.server.gen.edge.EntityViewUpdateMsg;
72 75 import org.thingsboard.server.gen.edge.NodeConnectionInfoProto;
73 76 import org.thingsboard.server.gen.edge.RequestMsg;
... ... @@ -140,11 +143,6 @@ public final class EdgeGrpcSession implements Cloneable {
140 143 .setUplinkResponseMsg(processUplinkMsg(requestMsg.getUplinkMsg()))
141 144 .build());
142 145 }
143   - if (requestMsg.getMsgType().equals(RequestMsgType.DEVICE_UPDATE_RPC_MESSAGE) && requestMsg.hasDeviceUpdateMsg()) {
144   - outputStream.onNext(ResponseMsg.newBuilder()
145   - .setUplinkResponseMsg(processUplinkMsg(requestMsg.getUplinkMsg()))
146   - .build());
147   - }
148 146 }
149 147 }
150 148
... ... @@ -161,6 +159,7 @@ public final class EdgeGrpcSession implements Cloneable {
161 159 };
162 160 }
163 161
  162 +
164 163 void processHandleMessages() throws ExecutionException, InterruptedException {
165 164 Long queueStartTs = getQueueStartTs().get();
166 165 // TODO: this 100 value must be changed properly
... ... @@ -170,7 +169,6 @@ public final class EdgeGrpcSession implements Cloneable {
170 169 do {
171 170 pageData = ctx.getEdgeService().findQueueEvents(edge.getTenantId(), edge.getId(), pageLink);
172 171 if (!pageData.getData().isEmpty()) {
173   - edge = ctx.getEdgeService().findEdgeById(edge.getTenantId(), edge.getId());
174 172 for (Event event : pageData.getData()) {
175 173 EdgeQueueEntry entry;
176 174 try {
... ... @@ -181,6 +179,8 @@ public final class EdgeGrpcSession implements Cloneable {
181 179 case ENTITY_DELETED_RPC_MESSAGE:
182 180 case ENTITY_UPDATED_RPC_MESSAGE:
183 181 case ENTITY_CREATED_RPC_MESSAGE:
  182 + case ALARM_ACK_RPC_MESSAGE:
  183 + case ALARM_CLEARK_RPC_MESSAGE:
184 184 processEntityCRUDMessage(entry, msgType);
185 185 break;
186 186 case RULE_CHAIN_CUSTOM_MESSAGE:
... ... @@ -239,6 +239,10 @@ public final class EdgeGrpcSession implements Cloneable {
239 239 private void processEntityCRUDMessage(EdgeQueueEntry entry, UpdateMsgType msgType) throws java.io.IOException {
240 240 log.trace("Executing processEntityCRUDMessage, entry [{}], msgType [{}]", entry, msgType);
241 241 switch (entry.getEntityType()) {
  242 + case EDGE:
  243 + Edge edge = objectMapper.readValue(entry.getData(), Edge.class);
  244 + onEdgeUpdated(msgType, edge);
  245 + break;
242 246 case DEVICE:
243 247 Device device = objectMapper.readValue(entry.getData(), Device.class);
244 248 onDeviceUpdated(msgType, device);
... ... @@ -263,6 +267,10 @@ public final class EdgeGrpcSession implements Cloneable {
263 267 RuleChainMetaData ruleChainMetaData = objectMapper.readValue(entry.getData(), RuleChainMetaData.class);
264 268 onRuleChainMetadataUpdated(msgType, ruleChainMetaData);
265 269 break;
  270 + case ALARM:
  271 + Alarm alarm = objectMapper.readValue(entry.getData(), Alarm.class);
  272 + onAlarmUpdated(msgType, alarm);
  273 + break;
266 274 }
267 275 }
268 276
... ... @@ -284,45 +292,107 @@ public final class EdgeGrpcSession implements Cloneable {
284 292 } );
285 293 }
286 294
  295 + private void onEdgeUpdated(UpdateMsgType msgType, Edge edge) {
  296 + // TODO: voba add configuration update to edge
  297 + this.edge = edge;
  298 + }
  299 +
287 300 private void onDeviceUpdated(UpdateMsgType msgType, Device device) {
288   - outputStream.onNext(ResponseMsg.newBuilder()
  301 + EntityUpdateMsg entityUpdateMsg = EntityUpdateMsg.newBuilder()
289 302 .setDeviceUpdateMsg(constructDeviceUpdatedMsg(msgType, device))
  303 + .build();
  304 + outputStream.onNext(ResponseMsg.newBuilder()
  305 + .setEntityUpdateMsg(entityUpdateMsg)
290 306 .build());
291 307 }
292 308
293 309 private void onAssetUpdated(UpdateMsgType msgType, Asset asset) {
294   - outputStream.onNext(ResponseMsg.newBuilder()
  310 + EntityUpdateMsg entityUpdateMsg = EntityUpdateMsg.newBuilder()
295 311 .setAssetUpdateMsg(constructAssetUpdatedMsg(msgType, asset))
  312 + .build();
  313 + outputStream.onNext(ResponseMsg.newBuilder()
  314 + .setEntityUpdateMsg(entityUpdateMsg)
296 315 .build());
297 316 }
298 317
299 318 private void onEntityViewUpdated(UpdateMsgType msgType, EntityView entityView) {
300   - outputStream.onNext(ResponseMsg.newBuilder()
  319 + EntityUpdateMsg entityUpdateMsg = EntityUpdateMsg.newBuilder()
301 320 .setEntityViewUpdateMsg(constructEntityViewUpdatedMsg(msgType, entityView))
  321 + .build();
  322 + outputStream.onNext(ResponseMsg.newBuilder()
  323 + .setEntityUpdateMsg(entityUpdateMsg)
302 324 .build());
303 325 }
304 326
305 327 private void onRuleChainUpdated(UpdateMsgType msgType, RuleChain ruleChain) {
306   - outputStream.onNext(ResponseMsg.newBuilder()
  328 + EntityUpdateMsg entityUpdateMsg = EntityUpdateMsg.newBuilder()
307 329 .setRuleChainUpdateMsg(constructRuleChainUpdatedMsg(msgType, ruleChain))
  330 + .build();
  331 + outputStream.onNext(ResponseMsg.newBuilder()
  332 + .setEntityUpdateMsg(entityUpdateMsg)
308 333 .build());
309 334 }
310 335
311 336 private void onRuleChainMetadataUpdated(UpdateMsgType msgType, RuleChainMetaData ruleChainMetaData) {
312 337 RuleChainMetadataUpdateMsg ruleChainMetadataUpdateMsg = constructRuleChainMetadataUpdatedMsg(msgType, ruleChainMetaData);
313 338 if (ruleChainMetadataUpdateMsg != null) {
314   - outputStream.onNext(ResponseMsg.newBuilder()
  339 + EntityUpdateMsg entityUpdateMsg = EntityUpdateMsg.newBuilder()
315 340 .setRuleChainMetadataUpdateMsg(ruleChainMetadataUpdateMsg)
  341 + .build();
  342 + outputStream.onNext(ResponseMsg.newBuilder()
  343 + .setEntityUpdateMsg(entityUpdateMsg)
316 344 .build());
317 345 }
318 346 }
319 347
320 348 private void onDashboardUpdated(UpdateMsgType msgType, Dashboard dashboard) {
321   - outputStream.onNext(ResponseMsg.newBuilder()
  349 + EntityUpdateMsg entityUpdateMsg = EntityUpdateMsg.newBuilder()
322 350 .setDashboardUpdateMsg(constructDashboardUpdatedMsg(msgType, dashboard))
  351 + .build();
  352 + outputStream.onNext(ResponseMsg.newBuilder()
  353 + .setEntityUpdateMsg(entityUpdateMsg)
  354 + .build());
  355 + }
  356 +
  357 + private void onAlarmUpdated(UpdateMsgType msgType, Alarm alarm) {
  358 + EntityUpdateMsg entityUpdateMsg = EntityUpdateMsg.newBuilder()
  359 + .setAlarmUpdateMsg(constructAlarmUpdatedMsg(msgType, alarm))
  360 + .build();
  361 + outputStream.onNext(ResponseMsg.newBuilder()
  362 + .setEntityUpdateMsg(entityUpdateMsg)
323 363 .build());
324 364 }
325 365
  366 + private AlarmUpdateMsg constructAlarmUpdatedMsg(UpdateMsgType msgType, Alarm alarm) {
  367 + String entityName = null;
  368 + switch (alarm.getOriginator().getEntityType()) {
  369 + case DEVICE:
  370 + entityName = ctx.getDeviceService().findDeviceById(edge.getTenantId(), new DeviceId(alarm.getOriginator().getId())).getName();
  371 + break;
  372 + case ASSET:
  373 + entityName = ctx.getAssetService().findAssetById(edge.getTenantId(), new AssetId(alarm.getOriginator().getId())).getName();
  374 + break;
  375 + case ENTITY_VIEW:
  376 + entityName = ctx.getEntityViewService().findEntityViewById(edge.getTenantId(), new EntityViewId(alarm.getOriginator().getId())).getName();
  377 + break;
  378 + }
  379 + AlarmUpdateMsg.Builder builder = AlarmUpdateMsg.newBuilder()
  380 + .setMsgType(msgType)
  381 + .setName(alarm.getName())
  382 + .setType(alarm.getName())
  383 + .setOriginatorName(entityName)
  384 + .setOriginatorType(alarm.getOriginator().getEntityType().name())
  385 + .setSeverity(alarm.getSeverity().name())
  386 + .setStatus(alarm.getStatus().name())
  387 + .setStartTs(alarm.getStartTs())
  388 + .setEndTs(alarm.getEndTs())
  389 + .setAckTs(alarm.getAckTs())
  390 + .setClearTs(alarm.getClearTs())
  391 + .setDetails(JacksonUtil.toString(alarm.getDetails()))
  392 + .setPropagate(alarm.isPropagate());
  393 + return builder.build();
  394 + }
  395 +
326 396 private UpdateMsgType getResponseMsgType(String msgType) {
327 397 if (msgType.equals(SessionMsgType.POST_TELEMETRY_REQUEST.name()) ||
328 398 msgType.equals(SessionMsgType.POST_ATTRIBUTES_REQUEST.name()) ||
... ... @@ -504,31 +574,31 @@ public final class EdgeGrpcSession implements Cloneable {
504 574 if (uplinkMsg.getEntityDataList() != null && !uplinkMsg.getEntityDataList().isEmpty()) {
505 575 for (EntityDataProto entityData : uplinkMsg.getEntityDataList()) {
506 576 TbMsg tbMsg = null;
507   - TbMsg tmp = TbMsg.fromBytes(entityData.getTbMsg().toByteArray());
508   - switch (tmp.getOriginator().getEntityType()) {
  577 + TbMsg originalTbMsg = TbMsg.fromBytes(entityData.getTbMsg().toByteArray());
  578 + switch (originalTbMsg.getOriginator().getEntityType()) {
509 579 case DEVICE:
510 580 String deviceName = entityData.getEntityName();
511 581 String deviceType = entityData.getEntityType();
512 582 Device device = getOrCreateDevice(deviceName, deviceType);
513 583 if (device != null) {
514   - tbMsg = new TbMsg(UUIDs.timeBased(), tmp.getType(), device.getId(), tmp.getMetaData().copy(),
515   - tmp.getDataType(), tmp.getData(), null, null, 0L);
  584 + tbMsg = new TbMsg(UUIDs.timeBased(), originalTbMsg.getType(), device.getId(), originalTbMsg.getMetaData().copy(),
  585 + originalTbMsg.getDataType(), originalTbMsg.getData(), null, null, 0L);
516 586 }
517 587 break;
518 588 case ASSET:
519 589 String assetName = entityData.getEntityName();
520 590 Asset asset = ctx.getAssetService().findAssetByTenantIdAndName(edge.getTenantId(), assetName);
521 591 if (asset != null) {
522   - tbMsg = new TbMsg(UUIDs.timeBased(), tmp.getType(), asset.getId(), tmp.getMetaData().copy(),
523   - tmp.getDataType(), tmp.getData(), null, null, 0L);
  592 + tbMsg = new TbMsg(UUIDs.timeBased(), originalTbMsg.getType(), asset.getId(), originalTbMsg.getMetaData().copy(),
  593 + originalTbMsg.getDataType(), originalTbMsg.getData(), null, null, 0L);
524 594 }
525 595 break;
526 596 case ENTITY_VIEW:
527 597 String entityViewName = entityData.getEntityName();
528 598 EntityView entityView = ctx.getEntityViewService().findEntityViewByTenantIdAndName(edge.getTenantId(), entityViewName);
529 599 if (entityView != null) {
530   - tbMsg = new TbMsg(UUIDs.timeBased(), tmp.getType(), entityView.getId(), tmp.getMetaData().copy(),
531   - tmp.getDataType(), tmp.getData(), null, null, 0L);
  600 + tbMsg = new TbMsg(UUIDs.timeBased(), originalTbMsg.getType(), entityView.getId(), originalTbMsg.getMetaData().copy(),
  601 + originalTbMsg.getDataType(), originalTbMsg.getData(), null, null, 0L);
532 602 }
533 603 break;
534 604 }
... ... @@ -537,6 +607,17 @@ public final class EdgeGrpcSession implements Cloneable {
537 607 }
538 608 }
539 609 }
  610 + if (uplinkMsg.getDeviceUpdateMsgList() != null && !uplinkMsg.getDeviceUpdateMsgList().isEmpty()) {
  611 + for (DeviceUpdateMsg deviceUpdateMsg : uplinkMsg.getDeviceUpdateMsgList()) {
  612 + String deviceName = deviceUpdateMsg.getName();
  613 + String deviceType = deviceUpdateMsg.getType();
  614 + switch (deviceUpdateMsg.getMsgType()) {
  615 + case ENTITY_CREATED_RPC_MESSAGE:
  616 + getOrCreateDevice(deviceName, deviceType);
  617 + break;
  618 + }
  619 + }
  620 + }
540 621 } catch (Exception e) {
541 622 return UplinkResponseMsg.newBuilder().setSuccess(false).setErrorMsg(e.getMessage()).build();
542 623 }
... ...
... ... @@ -16,5 +16,5 @@
16 16 package org.thingsboard.server.common.data.edge;
17 17
18 18 public enum EdgeQueueEntityType {
19   - DASHBOARD, ASSET, DEVICE, ENTITY_VIEW, ALARM, RULE_CHAIN, RULE_CHAIN_METADATA
  19 + DASHBOARD, ASSET, DEVICE, ENTITY_VIEW, ALARM, RULE_CHAIN, RULE_CHAIN_METADATA, EDGE
20 20 }
... ...
... ... @@ -24,21 +24,16 @@ import lombok.extern.slf4j.Slf4j;
24 24 import org.springframework.beans.factory.annotation.Value;
25 25 import org.springframework.stereotype.Service;
26 26 import org.thingsboard.edge.exception.EdgeConnectionException;
27   -import org.thingsboard.server.gen.edge.AssetUpdateMsg;
28 27 import org.thingsboard.server.gen.edge.ConnectRequestMsg;
29 28 import org.thingsboard.server.gen.edge.ConnectResponseCode;
30 29 import org.thingsboard.server.gen.edge.ConnectResponseMsg;
31   -import org.thingsboard.server.gen.edge.DashboardUpdateMsg;
32   -import org.thingsboard.server.gen.edge.DeviceUpdateMsg;
33 30 import org.thingsboard.server.gen.edge.DownlinkMsg;
34 31 import org.thingsboard.server.gen.edge.EdgeConfiguration;
35 32 import org.thingsboard.server.gen.edge.EdgeRpcServiceGrpc;
36   -import org.thingsboard.server.gen.edge.EntityViewUpdateMsg;
  33 +import org.thingsboard.server.gen.edge.EntityUpdateMsg;
37 34 import org.thingsboard.server.gen.edge.RequestMsg;
38 35 import org.thingsboard.server.gen.edge.RequestMsgType;
39 36 import org.thingsboard.server.gen.edge.ResponseMsg;
40   -import org.thingsboard.server.gen.edge.RuleChainMetadataUpdateMsg;
41   -import org.thingsboard.server.gen.edge.RuleChainUpdateMsg;
42 37 import org.thingsboard.server.gen.edge.UplinkMsg;
43 38 import org.thingsboard.server.gen.edge.UplinkResponseMsg;
44 39
... ... @@ -72,12 +67,7 @@ public class EdgeGrpcClient implements EdgeRpcClient {
72 67 String edgeSecret,
73 68 Consumer<UplinkResponseMsg> onUplinkResponse,
74 69 Consumer<EdgeConfiguration> onEdgeUpdate,
75   - Consumer<DeviceUpdateMsg> onDeviceUpdate,
76   - Consumer<AssetUpdateMsg> onAssetUpdate,
77   - Consumer<EntityViewUpdateMsg> onEntityViewUpdate,
78   - Consumer<RuleChainUpdateMsg> onRuleChainUpdate,
79   - Consumer<RuleChainMetadataUpdateMsg> onRuleChainMetadataUpdate,
80   - Consumer<DashboardUpdateMsg> onDashboardUpdate,
  70 + Consumer<EntityUpdateMsg> onEntityUpdate,
81 71 Consumer<DownlinkMsg> onDownlink,
82 72 Consumer<Exception> onError) {
83 73 NettyChannelBuilder builder = NettyChannelBuilder.forAddress(rpcHost, rpcPort).usePlaintext();
... ... @@ -92,7 +82,7 @@ public class EdgeGrpcClient implements EdgeRpcClient {
92 82 channel = builder.build();
93 83 EdgeRpcServiceGrpc.EdgeRpcServiceStub stub = EdgeRpcServiceGrpc.newStub(channel);
94 84 log.info("[{}] Sending a connect request to the TB!", edgeKey);
95   - this.inputStream = stub.handleMsgs(initOutputStream(edgeKey, onUplinkResponse, onEdgeUpdate, onDeviceUpdate, onAssetUpdate, onEntityViewUpdate, onRuleChainUpdate, onRuleChainMetadataUpdate, onDashboardUpdate, onDownlink, onError));
  85 + this.inputStream = stub.handleMsgs(initOutputStream(edgeKey, onUplinkResponse, onEdgeUpdate, onEntityUpdate, onDownlink, onError));
96 86 this.inputStream.onNext(RequestMsg.newBuilder()
97 87 .setMsgType(RequestMsgType.CONNECT_RPC_MESSAGE)
98 88 .setConnectRequestMsg(ConnectRequestMsg.newBuilder().setEdgeRoutingKey(edgeKey).setEdgeSecret(edgeSecret).build())
... ... @@ -118,12 +108,7 @@ public class EdgeGrpcClient implements EdgeRpcClient {
118 108 private StreamObserver<ResponseMsg> initOutputStream(String edgeKey,
119 109 Consumer<UplinkResponseMsg> onUplinkResponse,
120 110 Consumer<EdgeConfiguration> onEdgeUpdate,
121   - Consumer<DeviceUpdateMsg> onDeviceUpdate,
122   - Consumer<AssetUpdateMsg> onAssetUpdate,
123   - Consumer<EntityViewUpdateMsg> onEntityViewUpdate,
124   - Consumer<RuleChainUpdateMsg> onRuleChainUpdate,
125   - Consumer<RuleChainMetadataUpdateMsg> onRuleChainMetadataUpdate,
126   - Consumer<DashboardUpdateMsg> onDashboardUpdate,
  111 + Consumer<EntityUpdateMsg> onEntityUpdate,
127 112 Consumer<DownlinkMsg> onDownlink,
128 113 Consumer<Exception> onError) {
129 114 return new StreamObserver<ResponseMsg>() {
... ... @@ -141,24 +126,9 @@ public class EdgeGrpcClient implements EdgeRpcClient {
141 126 } else if (responseMsg.hasUplinkResponseMsg()) {
142 127 log.debug("[{}] Uplink response message received {}", edgeKey, responseMsg.getUplinkResponseMsg());
143 128 onUplinkResponse.accept(responseMsg.getUplinkResponseMsg());
144   - } else if (responseMsg.hasDeviceUpdateMsg()) {
145   - log.debug("[{}] Device update message received {}", edgeKey, responseMsg.getDeviceUpdateMsg());
146   - onDeviceUpdate.accept(responseMsg.getDeviceUpdateMsg());
147   - } else if (responseMsg.hasAssetUpdateMsg()) {
148   - log.debug("[{}] Asset update message received {}", edgeKey, responseMsg.getAssetUpdateMsg());
149   - onAssetUpdate.accept(responseMsg.getAssetUpdateMsg());
150   - } else if (responseMsg.hasEntityViewUpdateMsg()) {
151   - log.debug("[{}] EntityView update message received {}", edgeKey, responseMsg.getEntityViewUpdateMsg());
152   - onEntityViewUpdate.accept(responseMsg.getEntityViewUpdateMsg());
153   - } else if (responseMsg.hasRuleChainUpdateMsg()) {
154   - log.debug("[{}] Rule Chain udpate message received {}", edgeKey, responseMsg.getRuleChainUpdateMsg());
155   - onRuleChainUpdate.accept(responseMsg.getRuleChainUpdateMsg());
156   - } else if (responseMsg.hasRuleChainMetadataUpdateMsg()) {
157   - log.debug("[{}] Rule Chain Metadata udpate message received {}", edgeKey, responseMsg.getRuleChainMetadataUpdateMsg());
158   - onRuleChainMetadataUpdate.accept(responseMsg.getRuleChainMetadataUpdateMsg());
159   - } else if (responseMsg.hasDashboardUpdateMsg()) {
160   - log.debug("[{}] Dashboard message received {}", edgeKey, responseMsg.getDashboardUpdateMsg());
161   - onDashboardUpdate.accept(responseMsg.getDashboardUpdateMsg());
  129 + } else if (responseMsg.hasEntityUpdateMsg()) {
  130 + log.debug("[{}] Entity update message received {}", edgeKey, responseMsg.getEntityUpdateMsg());
  131 + onEntityUpdate.accept(responseMsg.getEntityUpdateMsg());
162 132 } else if (responseMsg.hasDownlinkMsg()) {
163 133 log.debug("[{}] Downlink message received for rule chain {}", edgeKey, responseMsg.getDownlinkMsg());
164 134 onDownlink.accept(responseMsg.getDownlinkMsg());
... ...
... ... @@ -15,14 +15,9 @@
15 15 */
16 16 package org.thingsboard.edge.rpc;
17 17
18   -import org.thingsboard.server.gen.edge.AssetUpdateMsg;
19   -import org.thingsboard.server.gen.edge.DashboardUpdateMsg;
20   -import org.thingsboard.server.gen.edge.DeviceUpdateMsg;
21 18 import org.thingsboard.server.gen.edge.DownlinkMsg;
22 19 import org.thingsboard.server.gen.edge.EdgeConfiguration;
23   -import org.thingsboard.server.gen.edge.EntityViewUpdateMsg;
24   -import org.thingsboard.server.gen.edge.RuleChainMetadataUpdateMsg;
25   -import org.thingsboard.server.gen.edge.RuleChainUpdateMsg;
  20 +import org.thingsboard.server.gen.edge.EntityUpdateMsg;
26 21 import org.thingsboard.server.gen.edge.UplinkMsg;
27 22 import org.thingsboard.server.gen.edge.UplinkResponseMsg;
28 23
... ... @@ -34,12 +29,7 @@ public interface EdgeRpcClient {
34 29 String integrationSecret,
35 30 Consumer<UplinkResponseMsg> onUplinkResponse,
36 31 Consumer<EdgeConfiguration> onEdgeUpdate,
37   - Consumer<DeviceUpdateMsg> onDeviceUpdate,
38   - Consumer<AssetUpdateMsg> onAssetUpdate,
39   - Consumer<EntityViewUpdateMsg> onEntityViewUpdate,
40   - Consumer<RuleChainUpdateMsg> onRuleChainUpdate,
41   - Consumer<RuleChainMetadataUpdateMsg> onRuleChainMetadataUpdate,
42   - Consumer<DashboardUpdateMsg> onDashboardUpdate,
  32 + Consumer<EntityUpdateMsg> onEntityUpdate,
43 33 Consumer<DownlinkMsg> onDownlink,
44 34 Consumer<Exception> onError);
45 35
... ...
... ... @@ -34,26 +34,29 @@ service EdgeRpcService {
34 34 message RequestMsg {
35 35 RequestMsgType msgType = 1;
36 36 ConnectRequestMsg connectRequestMsg = 2;
37   - DeviceUpdateMsg deviceUpdateMsg = 3;
38   - UplinkMsg uplinkMsg = 4;
  37 + UplinkMsg uplinkMsg = 3;
39 38 }
40 39
41 40 message ResponseMsg {
42 41 ConnectResponseMsg connectResponseMsg = 1;
43 42 UplinkResponseMsg uplinkResponseMsg = 2;
44   - DeviceUpdateMsg deviceUpdateMsg = 3;
45   - RuleChainUpdateMsg ruleChainUpdateMsg = 4;
46   - RuleChainMetadataUpdateMsg ruleChainMetadataUpdateMsg = 5;
47   - DashboardUpdateMsg dashboardUpdateMsg = 6;
48   - AssetUpdateMsg assetUpdateMsg = 7;
49   - EntityViewUpdateMsg entityViewUpdateMsg = 8;
50   - DownlinkMsg downlinkMsg = 9;
  43 + EntityUpdateMsg entityUpdateMsg = 3;
  44 + DownlinkMsg downlinkMsg = 4;
  45 +}
  46 +
  47 +message EntityUpdateMsg {
  48 + DeviceUpdateMsg deviceUpdateMsg = 1;
  49 + RuleChainUpdateMsg ruleChainUpdateMsg = 2;
  50 + RuleChainMetadataUpdateMsg ruleChainMetadataUpdateMsg = 3;
  51 + DashboardUpdateMsg dashboardUpdateMsg = 4;
  52 + AssetUpdateMsg assetUpdateMsg = 5;
  53 + EntityViewUpdateMsg entityViewUpdateMsg = 6;
  54 + AlarmUpdateMsg alarmUpdateMsg = 7;
51 55 }
52 56
53 57 enum RequestMsgType {
54 58 CONNECT_RPC_MESSAGE = 0;
55 59 UPLINK_RPC_MESSAGE = 1;
56   - DEVICE_UPDATE_RPC_MESSAGE = 2;
57 60 }
58 61
59 62 message ConnectRequestMsg {
... ... @@ -85,7 +88,9 @@ enum UpdateMsgType {
85 88 ENTITY_CREATED_RPC_MESSAGE = 0;
86 89 ENTITY_UPDATED_RPC_MESSAGE = 1;
87 90 ENTITY_DELETED_RPC_MESSAGE = 2;
88   - RULE_CHAIN_CUSTOM_MESSAGE = 3;
  91 + ALARM_ACK_RPC_MESSAGE = 3;
  92 + ALARM_CLEARK_RPC_MESSAGE = 4;
  93 + RULE_CHAIN_CUSTOM_MESSAGE = 5;
89 94 }
90 95
91 96 message EntityDataProto {
... ... @@ -166,6 +171,22 @@ message EntityViewUpdateMsg {
166 171 EntityType relatedEntityType = 6;
167 172 }
168 173
  174 +message AlarmUpdateMsg {
  175 + UpdateMsgType msgType = 1;
  176 + string name = 2;
  177 + string type = 3;
  178 + string originatorType = 4;
  179 + string originatorName = 5;
  180 + string severity = 6;
  181 + string status = 7;
  182 + int64 startTs = 8;
  183 + int64 endTs = 9;
  184 + int64 ackTs = 10;
  185 + int64 clearTs = 11;
  186 + string details = 12;
  187 + bool propagate = 13;
  188 +}
  189 +
169 190 enum EntityType {
170 191 DEVICE = 0;
171 192 ASSET = 1;
... ... @@ -178,6 +199,8 @@ enum EntityType {
178 199 message UplinkMsg {
179 200 int32 uplinkMsgId = 1;
180 201 repeated EntityDataProto entityData = 2;
  202 + repeated DeviceUpdateMsg deviceUpdateMsg = 3;
  203 + repeated AlarmUpdateMsg alarmUpdatemsg = 4;
181 204 }
182 205
183 206 message UplinkResponseMsg {
... ...
... ... @@ -37,6 +37,7 @@ import org.thingsboard.server.common.data.EntityView;
37 37 import org.thingsboard.server.common.data.Event;
38 38 import org.thingsboard.server.common.data.ShortEdgeInfo;
39 39 import org.thingsboard.server.common.data.Tenant;
  40 +import org.thingsboard.server.common.data.alarm.Alarm;
40 41 import org.thingsboard.server.common.data.asset.Asset;
41 42 import org.thingsboard.server.common.data.edge.Edge;
42 43 import org.thingsboard.server.common.data.edge.EdgeQueueEntityType;
... ... @@ -151,7 +152,6 @@ public class BaseEdgeService extends AbstractEntityService implements EdgeServic
151 152 }
152 153 }
153 154
154   -
155 155 @Override
156 156 public Edge findEdgeById(TenantId tenantId, EdgeId edgeId) {
157 157 log.trace("Executing findEdgeById [{}]", edgeId);
... ... @@ -349,6 +349,9 @@ public class BaseEdgeService extends AbstractEntityService implements EdgeServic
349 349 } else {
350 350 try {
351 351 switch (tbMsg.getOriginator().getEntityType()) {
  352 + case EDGE:
  353 + processEdge(tenantId, tbMsg, callback);
  354 + break;
352 355 case ASSET:
353 356 processAsset(tenantId, tbMsg, callback);
354 357 break;
... ... @@ -364,6 +367,9 @@ public class BaseEdgeService extends AbstractEntityService implements EdgeServic
364 367 case ENTITY_VIEW:
365 368 processEntityView(tenantId, tbMsg, callback);
366 369 break;
  370 + case ALARM:
  371 + processAlarm(tenantId, tbMsg, callback);
  372 + break;
367 373 default:
368 374 log.debug("Entity type [{}] is not designed to be pushed to edge", tbMsg.getOriginator().getEntityType());
369 375 }
... ... @@ -374,26 +380,9 @@ public class BaseEdgeService extends AbstractEntityService implements EdgeServic
374 380 }
375 381
376 382 private void processCustomTbMsg(TenantId tenantId, TbMsg tbMsg, FutureCallback<Void> callback) {
377   - EdgeId edgeId = null;
378   - EdgeQueueEntityType edgeQueueEntityType = null;
379   - switch (tbMsg.getOriginator().getEntityType()) {
380   - case DEVICE:
381   - edgeQueueEntityType = EdgeQueueEntityType.DEVICE;
382   - Device device = deviceService.findDeviceById(tenantId, new DeviceId(tbMsg.getOriginator().getId()));
383   - edgeId = device.getEdgeId();
384   - break;
385   - case ASSET:
386   - edgeQueueEntityType = EdgeQueueEntityType.ASSET;
387   - Asset asset = assetService.findAssetById(tenantId, new AssetId(tbMsg.getOriginator().getId()));
388   - edgeId = asset.getEdgeId();
389   - break;
390   - case ENTITY_VIEW:
391   - edgeQueueEntityType = EdgeQueueEntityType.ENTITY_VIEW;
392   - EntityView entityView = entityViewService.findEntityViewById(tenantId, new EntityViewId(tbMsg.getOriginator().getId()));
393   - edgeId = entityView.getEdgeId();
394   - break;
395   - }
396   - if (edgeId != null) {
  383 + EdgeId edgeId = getEdgeIdByOriginatorId(tenantId, tbMsg.getOriginator());
  384 + EdgeQueueEntityType edgeQueueEntityType = getEdgeQueueTypeByEntityType(tbMsg.getOriginator().getEntityType());
  385 + if (edgeId != null && edgeQueueEntityType != null) {
397 386 try {
398 387 saveEventToEdgeQueue(tenantId, edgeId, edgeQueueEntityType, tbMsg.getType(), mapper.writeValueAsString(tbMsg), callback);
399 388 } catch (IOException e) {
... ... @@ -402,6 +391,37 @@ public class BaseEdgeService extends AbstractEntityService implements EdgeServic
402 391 }
403 392 }
404 393
  394 + private EdgeQueueEntityType getEdgeQueueTypeByEntityType(EntityType entityType) {
  395 + switch (entityType) {
  396 + case DEVICE:
  397 + return EdgeQueueEntityType.DEVICE;
  398 + case ASSET:
  399 + return EdgeQueueEntityType.ASSET;
  400 + case ENTITY_VIEW:
  401 + return EdgeQueueEntityType.ENTITY_VIEW;
  402 + default:
  403 + log.info("Unsupported entity type: [{}]", entityType);
  404 + return null;
  405 + }
  406 + }
  407 +
  408 + private EdgeId getEdgeIdByOriginatorId(TenantId tenantId, EntityId originatorId) {
  409 + switch (originatorId.getEntityType()) {
  410 + case DEVICE:
  411 + Device device = deviceService.findDeviceById(tenantId, new DeviceId(originatorId.getId()));
  412 + return device.getEdgeId();
  413 + case ASSET:
  414 + Asset asset = assetService.findAssetById(tenantId, new AssetId(originatorId.getId()));
  415 + return asset.getEdgeId();
  416 + case ENTITY_VIEW:
  417 + EntityView entityView = entityViewService.findEntityViewById(tenantId, new EntityViewId(originatorId.getId()));
  418 + return entityView.getEdgeId();
  419 + default:
  420 + log.info("Unsupported entity type: [{}]", originatorId.getEntityType());
  421 + return null;
  422 + }
  423 + }
  424 +
405 425 private void processDevice(TenantId tenantId, TbMsg tbMsg, FutureCallback<Void> callback) throws IOException {
406 426 switch (tbMsg.getType()) {
407 427 case DataConstants.ENTITY_ASSIGNED_TO_EDGE:
... ... @@ -421,6 +441,21 @@ public class BaseEdgeService extends AbstractEntityService implements EdgeServic
421 441 }
422 442 }
423 443
  444 + private void processEdge(TenantId tenantId, TbMsg tbMsg, FutureCallback<Void> callback) throws IOException {
  445 + switch (tbMsg.getType()) {
  446 + case DataConstants.ENTITY_DELETED:
  447 + case DataConstants.ENTITY_CREATED:
  448 + case DataConstants.ENTITY_UPDATED:
  449 + Edge edge = mapper.readValue(tbMsg.getData(), Edge.class);
  450 + if (edge != null) {
  451 + pushEventToEdge(tenantId, edge.getId(), EdgeQueueEntityType.EDGE, tbMsg, callback);
  452 + }
  453 + break;
  454 + default:
  455 + log.warn("Unsupported msgType [{}], tbMsg [{}]", tbMsg.getType(), tbMsg);
  456 + }
  457 + }
  458 +
424 459 private void processAsset(TenantId tenantId, TbMsg tbMsg, FutureCallback<Void> callback) throws IOException {
425 460 switch (tbMsg.getType()) {
426 461 case DataConstants.ENTITY_ASSIGNED_TO_EDGE:
... ... @@ -459,6 +494,25 @@ public class BaseEdgeService extends AbstractEntityService implements EdgeServic
459 494 }
460 495 }
461 496
  497 + private void processAlarm(TenantId tenantId, TbMsg tbMsg, FutureCallback<Void> callback) throws IOException {
  498 + switch (tbMsg.getType()) {
  499 + case DataConstants.ENTITY_DELETED:
  500 + case DataConstants.ENTITY_CREATED:
  501 + case DataConstants.ENTITY_UPDATED:
  502 + case DataConstants.ALARM_ACK:
  503 + case DataConstants.ALARM_CLEAR:
  504 + Alarm alarm = mapper.readValue(tbMsg.getData(), Alarm.class);
  505 + EdgeId edgeId = getEdgeIdByOriginatorId(tenantId, alarm.getOriginator());
  506 + EdgeQueueEntityType edgeQueueEntityType = getEdgeQueueTypeByEntityType(alarm.getOriginator().getEntityType());
  507 + if (edgeId != null && edgeQueueEntityType != null) {
  508 + pushEventToEdge(tenantId, edgeId, EdgeQueueEntityType.ALARM, tbMsg, callback);
  509 + }
  510 + break;
  511 + default:
  512 + log.warn("Unsupported msgType [{}], tbMsg [{}]", tbMsg.getType(), tbMsg);
  513 + }
  514 + }
  515 +
462 516 private void processDashboard(TenantId tenantId, TbMsg tbMsg, FutureCallback<Void> callback) throws IOException {
463 517 processAssignedEntity(tenantId, tbMsg, EdgeQueueEntityType.DASHBOARD, callback);
464 518 }
... ... @@ -562,8 +616,8 @@ public class BaseEdgeService extends AbstractEntityService implements EdgeServic
562 616 @Override
563 617 public Edge setRootRuleChain(TenantId tenantId, Edge edge, RuleChainId ruleChainId) throws IOException {
564 618 edge.setRootRuleChainId(ruleChainId);
565   - Edge saveEdge = saveEdge(edge);
566   - ruleChainService.updateEdgeRuleChains(tenantId, saveEdge.getId());
  619 + Edge savedEdge = saveEdge(edge);
  620 + ruleChainService.updateEdgeRuleChains(tenantId, savedEdge.getId());
567 621 RuleChain ruleChain = ruleChainService.findRuleChainById(tenantId, ruleChainId);
568 622 saveEventToEdgeQueue(tenantId, edge.getId(), EdgeQueueEntityType.RULE_CHAIN, DataConstants.ENTITY_UPDATED, mapper.writeValueAsString(ruleChain), new FutureCallback<Void>() {
569 623 @Override
... ... @@ -576,7 +630,7 @@ public class BaseEdgeService extends AbstractEntityService implements EdgeServic
576 630 log.debug("Failure during event save", t);
577 631 }
578 632 });
579   - return saveEdge;
  633 + return savedEdge;
580 634 }
581 635
582 636 private DataValidator<Edge> edgeValidator =
... ...