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

Merge remote-tracking branch 'origin/develop/2.6-edge' into develop/3.3-edge

pull/3811/head
Volodymyr Babak 6 years ago
parent
c48315b54c b48d43f4dd
commit
64694532db
33 changed files with 774 additions and 337 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. 16
      application/src/main/data/upgrade/2.6.0/schema_update.cql
  2. 19
      application/src/main/java/org/thingsboard/server/actors/app/AppActor.java
  3. 20
      application/src/main/java/org/thingsboard/server/actors/device/DeviceActorMessageProcessor.java
  4. 7
      application/src/main/java/org/thingsboard/server/actors/ruleChain/DefaultTbContext.java
  5. 9
      application/src/main/java/org/thingsboard/server/actors/tenant/TenantActor.java
  6. 1
      application/src/main/java/org/thingsboard/server/controller/BaseController.java
  7. 7
      application/src/main/java/org/thingsboard/server/controller/RuleChainController.java
  8. 248
      application/src/main/java/org/thingsboard/server/service/edge/DefaultEdgeNotificationService.java
  9. 91
      application/src/main/java/org/thingsboard/server/service/edge/rpc/EdgeGrpcService.java
  10. 63
      application/src/main/java/org/thingsboard/server/service/edge/rpc/EdgeGrpcSession.java
  11. 2
      application/src/main/java/org/thingsboard/server/service/edge/rpc/EdgeRpcService.java
  12. 252
      application/src/main/java/org/thingsboard/server/service/edge/rpc/init/DefaultSyncEdgeService.java
  13. 17
      application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/BaseProcessor.java
  14. 20
      application/src/main/java/org/thingsboard/server/service/queue/DefaultTbClusterService.java
  15. 7
      application/src/main/java/org/thingsboard/server/service/queue/DefaultTbCoreConsumerService.java
  16. 3
      application/src/main/java/org/thingsboard/server/service/queue/TbClusterService.java
  17. 1
      application/src/main/java/org/thingsboard/server/service/rpc/DefaultTbRuleEngineRpcService.java
  18. 1
      application/src/main/resources/thingsboard.yml
  19. 27
      application/src/test/java/org/thingsboard/server/edge/BaseEdgeTest.java
  20. 22
      application/src/test/java/org/thingsboard/server/edge/imitator/EdgeImitator.java
  21. 3
      common/data/src/main/java/org/thingsboard/server/common/data/EdgeUtils.java
  22. 17
      common/edge-api/src/main/proto/edge.proto
  23. 7
      common/message/src/main/java/org/thingsboard/server/common/msg/MsgType.java
  24. 42
      common/message/src/main/java/org/thingsboard/server/common/msg/edge/EdgeEventUpdateMsg.java
  25. 1
      common/queue/src/main/proto/queue.proto
  26. 67
      dao/src/main/java/org/thingsboard/server/dao/edge/EdgeServiceImpl.java
  27. 3
      dao/src/main/java/org/thingsboard/server/dao/entity/AbstractEntityService.java
  28. 16
      dao/src/main/java/org/thingsboard/server/dao/event/BaseEventService.java
  29. 3
      rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/TbContext.java
  30. 101
      rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/edge/TbMsgPushToEdgeNode.java
  31. 14
      ui-ngx/src/assets/locale/locale.constant-en_US.json
  32. 2
      ui/src/app/edge/edge.controller.js
  33. 2
      ui/src/app/edge/edge.routes.js

16
application/src/main/data/upgrade/2.6.0/schema_update.cql
View File

@ -107,15 +107,15 @@ CREATE MATERIALIZED VIEW IF NOT EXISTS thingsboard.edge_by_customer_by_type_and_
WITH CLUSTERING ORDER BY ( tenant_id DESC, type ASC, search_text ASC, id DESC );
CREATE TABLE IF NOT EXISTS thingsboard.edge_event (
id timeuuid,
tenant_id timeuuid,
id timeuuid,
tenant_id timeuuid,
edge_id timeuuid,
edge_event_type text,
edge_event_action text,
edge_event_uid text,
entity_id timeuuid,
body text,
PRIMARY KEY ((tenant_id, edge_id), edge_event_type, edge_event_uid)
edge_event_type text,
edge_event_action text,
edge_event_uid text,
entity_id timeuuid,
body text,
PRIMARY KEY ((tenant_id, edge_id), edge_event_type, edge_event_uid)
);
CREATE MATERIALIZED VIEW IF NOT EXISTS thingsboard.edge_event_by_id AS

19
application/src/main/java/org/thingsboard/server/actors/app/AppActor.java
View File

@ -35,10 +35,12 @@ import org.thingsboard.server.common.data.plugin.ComponentLifecycleEvent;
import org.thingsboard.server.common.msg.MsgType;
import org.thingsboard.server.common.msg.TbActorMsg;
import org.thingsboard.server.common.msg.aware.TenantAwareMsg;
import org.thingsboard.server.common.msg.edge.EdgeEventUpdateMsg;
import org.thingsboard.server.common.msg.plugin.ComponentLifecycleMsg;
import org.thingsboard.server.common.msg.queue.QueueToRuleEngineMsg;
import org.thingsboard.server.common.msg.queue.RuleEngineException;
import org.thingsboard.server.common.msg.queue.ServiceType;
import org.thingsboard.server.dao.model.ModelConstants;
import org.thingsboard.server.dao.tenant.TenantService;
import org.thingsboard.server.dao.tenant.TbTenantProfileCache;
import org.thingsboard.server.service.transport.msg.TransportToDeviceActorMsgWrapper;
@ -95,6 +97,9 @@ public class AppActor extends ContextAwareActor {
case SERVER_RPC_RESPONSE_TO_DEVICE_ACTOR_MSG:
onToDeviceActorMsg((TenantAwareMsg) msg, true);
break;
case EDGE_EVENT_UPDATE_TO_EDGE_SESSION_MSG:
onToTenantActorMsg((EdgeEventUpdateMsg) msg);
break;
default:
return false;
}
@ -194,6 +199,20 @@ public class AppActor extends ContextAwareActor {
() -> new TenantActor.ActorCreator(systemContext, tenantId));
}
private void onToTenantActorMsg(EdgeEventUpdateMsg msg) {
TbActorRef target = null;
if (ModelConstants.SYSTEM_TENANT.equals(msg.getTenantId())) {
log.warn("Message has system tenant id: {}", msg);
} else {
target = getOrCreateTenantActor(msg.getTenantId());
}
if (target != null) {
target.tellWithHighPriority(msg);
} else {
log.debug("[{}] Invalid edge event update msg: {}", msg.getTenantId(), msg);
}
}
public static class ActorCreator extends ContextBasedCreator {
public ActorCreator(ActorSystemContext context) {

20
application/src/main/java/org/thingsboard/server/actors/device/DeviceActorMessageProcessor.java
View File

@ -145,8 +145,13 @@ class DeviceActorMessageProcessor extends AbstractContextAwareMsgProcessor {
if (result != null && result.size() > 0) {
EntityRelation relationToEdge = result.get(0);
if (relationToEdge.getFrom() != null && relationToEdge.getFrom().getId() != null) {
log.trace("[{}][{}] found edge [{}] for device", tenantId, deviceId, relationToEdge.getFrom().getId());
return new EdgeId(relationToEdge.getFrom().getId());
} else {
log.trace("[{}][{}] edge relation is empty {}", tenantId, deviceId, relationToEdge);
}
} else {
log.trace("[{}][{}] device doesn't have any related edge", tenantId, deviceId);
}
return null;
}
@ -165,6 +170,7 @@ class DeviceActorMessageProcessor extends AbstractContextAwareMsgProcessor {
boolean sent;
if (systemContext.isEdgesRpcEnabled() && edgeId != null) {
log.debug("[{}][{}] device is related to edge [{}]. Saving RPC request to edge queue", tenantId, deviceId, edgeId.getId());
saveRpcRequestToEdgeQueue(request, rpcRequest.getRequestId());
sent = true;
} else {
@ -516,6 +522,7 @@ class DeviceActorMessageProcessor extends AbstractContextAwareMsgProcessor {
}
void processEdgeUpdate(DeviceEdgeUpdateMsg msg) {
log.trace("[{}] Processing edge update {}", deviceId, msg);
this.edgeId = msg.getEdgeId();
}
@ -568,7 +575,18 @@ class DeviceActorMessageProcessor extends AbstractContextAwareMsgProcessor {
edgeEvent.setBody(body);
edgeEvent.setEdgeId(edgeId);
systemContext.getEdgeEventService().saveAsync(edgeEvent);
ListenableFuture<EdgeEvent> future = systemContext.getEdgeEventService().saveAsync(edgeEvent);
Futures.addCallback(future, new FutureCallback<EdgeEvent>() {
@Override
public void onSuccess( EdgeEvent result) {
systemContext.getClusterService().onEdgeEventUpdate(tenantId, edgeId);
}
@Override
public void onFailure(Throwable t) {
log.warn("[{}] Can't save edge event [{}] for edge [{}]", tenantId.getId(), edgeEvent, edgeId.getId(), t);
}
}, systemContext.getDbCallbackExecutor());
}
private List<TsKvProto> toTsKvProtos(@Nullable List<AttributeKvEntry> result) {

7
application/src/main/java/org/thingsboard/server/actors/ruleChain/DefaultTbContext.java
View File

@ -34,7 +34,6 @@ import org.thingsboard.rule.engine.api.TbRelationTypes;
import org.thingsboard.rule.engine.api.sms.SmsSenderFactory;
import org.thingsboard.server.actors.ActorSystemContext;
import org.thingsboard.server.actors.TbActorRef;
import org.thingsboard.server.common.data.ApiUsageRecordKey;
import org.thingsboard.server.common.data.Customer;
import org.thingsboard.server.common.data.DataConstants;
import org.thingsboard.server.common.data.Device;
@ -43,6 +42,7 @@ import org.thingsboard.server.common.data.TenantProfile;
import org.thingsboard.server.common.data.alarm.Alarm;
import org.thingsboard.server.common.data.asset.Asset;
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.RuleChainId;
import org.thingsboard.server.common.data.id.RuleNodeId;
@ -280,6 +280,11 @@ class DefaultTbContext implements TbContext {
return entityActionMsg(alarm, alarm.getId(), ruleNodeId, action);
}
@Override
public void onEdgeEventUpdate(TenantId tenantId, EdgeId edgeId) {
mainCtx.getClusterService().onEdgeEventUpdate(tenantId, edgeId);
}
public <E, I extends EntityId> TbMsg entityActionMsg(E entity, I id, RuleNodeId ruleNodeId, String action) {
try {
return TbMsg.newMsg(action, id, getActionMetaData(ruleNodeId), mapper.writeValueAsString(mapper.valueToTree(entity)));

9
application/src/main/java/org/thingsboard/server/actors/tenant/TenantActor.java
View File

@ -47,6 +47,7 @@ import org.thingsboard.server.common.msg.TbActorMsg;
import org.thingsboard.server.common.msg.TbMsg;
import org.thingsboard.server.common.msg.aware.DeviceAwareMsg;
import org.thingsboard.server.common.msg.aware.RuleChainAwareMsg;
import org.thingsboard.server.common.msg.edge.EdgeEventUpdateMsg;
import org.thingsboard.server.common.msg.plugin.ComponentLifecycleMsg;
import org.thingsboard.server.common.msg.queue.PartitionChangeMsg;
import org.thingsboard.server.common.msg.queue.QueueToRuleEngineMsg;
@ -169,6 +170,9 @@ public class TenantActor extends RuleChainManagerActor {
case RULE_CHAIN_TO_RULE_CHAIN_MSG:
onRuleChainMsg((RuleChainAwareMsg) msg);
break;
case EDGE_EVENT_UPDATE_TO_EDGE_SESSION_MSG:
onToEdgeSessionMsg((EdgeEventUpdateMsg) msg);
break;
default:
return false;
}
@ -271,6 +275,11 @@ public class TenantActor extends RuleChainManagerActor {
() -> new DeviceActorCreator(systemContext, tenantId, deviceId));
}
private void onToEdgeSessionMsg(EdgeEventUpdateMsg msg) {
log.trace("[{}] onToEdgeSessionMsg [{}]", msg.getTenantId(), msg);
systemContext.getEdgeRpcService().onEdgeEvent(msg.getEdgeId());
}
public static class ActorCreator extends ContextBasedCreator {
private final TenantId tenantId;

1
application/src/main/java/org/thingsboard/server/controller/BaseController.java
View File

@ -955,6 +955,7 @@ public abstract class BaseController {
builder.setBody(body);
}
TransportProtos.EdgeNotificationMsgProto msg = builder.build();
log.trace("[{}] sending notification to edge service {}", tenantId.getId(), msg);
tbClusterService.pushMsgToCore(tenantId, entityId != null ? entityId : tenantId,
TransportProtos.ToCoreMsg.newBuilder().setEdgeNotificationMsg(msg).build(), null);
}

7
application/src/main/java/org/thingsboard/server/controller/RuleChainController.java
View File

@ -282,12 +282,11 @@ public class RuleChainController extends BaseController {
try {
TenantId tenantId = getCurrentUser().getTenantId();
PageLink pageLink = createPageLink(pageSize, page, textSearch, sortProperty, sortOrder);
RuleChainType type = RuleChainType.CORE;
if (typeStr != null && typeStr.trim().length() > 0) {
RuleChainType type = RuleChainType.valueOf(typeStr);
return checkNotNull(ruleChainService.findTenantRuleChainsByType(tenantId, type, pageLink));
} else {
return checkNotNull(ruleChainService.findTenantRuleChainsByType(tenantId, RuleChainType.CORE, pageLink));
type = RuleChainType.valueOf(typeStr);
}
return checkNotNull(ruleChainService.findTenantRuleChainsByType(tenantId, type, pageLink));
} catch (Exception e) {
throw handleException(e);
}

248
application/src/main/java/org/thingsboard/server/service/edge/DefaultEdgeNotificationService.java
View File

@ -25,6 +25,7 @@ import lombok.extern.slf4j.Slf4j;
import org.checkerframework.checker.nullness.qual.Nullable;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.thingsboard.server.common.data.EdgeUtils;
import org.thingsboard.server.common.data.EntityType;
import org.thingsboard.server.common.data.User;
import org.thingsboard.server.common.data.alarm.Alarm;
@ -54,6 +55,7 @@ import org.thingsboard.server.dao.user.UserService;
import org.thingsboard.server.gen.transport.TransportProtos;
import org.thingsboard.server.queue.util.TbCoreComponent;
import org.thingsboard.server.service.executors.DbCallbackExecutorService;
import org.thingsboard.server.service.queue.TbClusterService;
import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
@ -73,6 +75,8 @@ public class DefaultEdgeNotificationService implements EdgeNotificationService {
private static final ObjectMapper mapper = new ObjectMapper();
private static final int DEFAULT_LIMIT = 100;
@Autowired
private EdgeService edgeService;
@ -88,6 +92,9 @@ public class DefaultEdgeNotificationService implements EdgeNotificationService {
@Autowired
private EdgeEventService edgeEventService;
@Autowired
private TbClusterService clusterService;
@Autowired
private DbCallbackExecutorService dbCallbackExecutorService;
@ -136,7 +143,19 @@ public class DefaultEdgeNotificationService implements EdgeNotificationService {
edgeEvent.setEntityId(entityId.getId());
}
edgeEvent.setBody(body);
edgeEventService.saveAsync(edgeEvent);
ListenableFuture<EdgeEvent> future = edgeEventService.saveAsync(edgeEvent);
Futures.addCallback(future, new FutureCallback<EdgeEvent>() {
@Override
public void onSuccess(@Nullable EdgeEvent result) {
clusterService.onEdgeEventUpdate(tenantId, edgeId);
}
@Override
public void onFailure(Throwable t) {
log.warn("[{}] Can't save edge event [{}] for edge [{}]", tenantId.getId(), edgeEvent, edgeId.getId(), t);
}
}, dbCallbackExecutorService);
}
@Override
@ -194,13 +213,20 @@ public class DefaultEdgeNotificationService implements EdgeNotificationService {
public void onSuccess(@Nullable Edge edge) {
if (edge != null && !customerId.isNullUid()) {
saveEdgeEvent(edge.getTenantId(), edge.getId(), EdgeEventType.CUSTOMER, EdgeEventActionType.ADDED, customerId, null);
PageData<User> pageData = userService.findCustomerUsers(tenantId, customerId, new PageLink(Integer.MAX_VALUE));
if (pageData != null && pageData.getData() != null && !pageData.getData().isEmpty()) {
log.trace("[{}] [{}] user(s) are going to be added to edge.", edge.getId(), pageData.getData().size());
for (User user : pageData.getData()) {
saveEdgeEvent(edge.getTenantId(), edge.getId(), EdgeEventType.USER, EdgeEventActionType.ADDED, user.getId(), null);
PageLink pageLink = new PageLink(DEFAULT_LIMIT);
PageData<User> pageData;
do {
pageData = userService.findCustomerUsers(tenantId, customerId, pageLink);
if (pageData != null && pageData.getData() != null && !pageData.getData().isEmpty()) {
log.trace("[{}] [{}] user(s) are going to be added to edge.", edge.getId(), pageData.getData().size());
for (User user : pageData.getData()) {
saveEdgeEvent(edge.getTenantId(), edge.getId(), EdgeEventType.USER, EdgeEventActionType.ADDED, user.getId(), null);
}
if (pageData.hasNext()) {
pageLink = pageLink.nextPageLink();
}
}
}
} while (pageData != null && pageData.hasNext());
}
}
@ -241,12 +267,19 @@ public class DefaultEdgeNotificationService implements EdgeNotificationService {
case ADDED:
case UPDATED:
case DELETED:
PageData<Edge> edgesByTenantId = edgeService.findEdgesByTenantId(tenantId, new PageLink(Integer.MAX_VALUE));
if (edgesByTenantId != null && edgesByTenantId.getData() != null && !edgesByTenantId.getData().isEmpty()) {
for (Edge edge : edgesByTenantId.getData()) {
saveEdgeEvent(tenantId, edge.getId(), type, actionType, entityId, null);
PageLink pageLink = new PageLink(DEFAULT_LIMIT);
PageData<Edge> pageData;
do {
pageData = edgeService.findEdgesByTenantId(tenantId, pageLink);
if (pageData != null && pageData.getData() != null && !pageData.getData().isEmpty()) {
for (Edge edge : pageData.getData()) {
saveEdgeEvent(tenantId, edge.getId(), type, actionType, entityId, null);
}
if (pageData.hasNext()) {
pageLink = pageLink.nextPageLink();
}
}
}
} while (pageData != null && pageData.hasNext());
break;
}
}
@ -255,21 +288,28 @@ public class DefaultEdgeNotificationService implements EdgeNotificationService {
EdgeEventActionType actionType = EdgeEventActionType.valueOf(edgeNotificationMsg.getAction());
EdgeEventType type = EdgeEventType.valueOf(edgeNotificationMsg.getType());
EntityId entityId = EntityIdFactory.getByEdgeEventTypeAndUuid(type, new UUID(edgeNotificationMsg.getEntityIdMSB(), edgeNotificationMsg.getEntityIdLSB()));
PageData<Edge> edgesByTenantId = edgeService.findEdgesByTenantId(tenantId, new PageLink(Integer.MAX_VALUE));
if (edgesByTenantId != null && edgesByTenantId.getData() != null && !edgesByTenantId.getData().isEmpty()) {
for (Edge edge : edgesByTenantId.getData()) {
switch (actionType) {
case UPDATED:
if (!edge.getCustomerId().isNullUid() && edge.getCustomerId().equals(entityId)) {
PageLink pageLink = new PageLink(DEFAULT_LIMIT);
PageData<Edge> pageData;
do {
pageData = edgeService.findEdgesByTenantId(tenantId, pageLink);
if (pageData != null && pageData.getData() != null && !pageData.getData().isEmpty()) {
for (Edge edge : pageData.getData()) {
switch (actionType) {
case UPDATED:
if (!edge.getCustomerId().isNullUid() && edge.getCustomerId().equals(entityId)) {
saveEdgeEvent(tenantId, edge.getId(), type, actionType, entityId, null);
}
break;
case DELETED:
saveEdgeEvent(tenantId, edge.getId(), type, actionType, entityId, null);
}
break;
case DELETED:
saveEdgeEvent(tenantId, edge.getId(), type, actionType, entityId, null);
break;
break;
}
}
if (pageData.hasNext()) {
pageLink = pageLink.nextPageLink();
}
}
}
} while (pageData != null && pageData.hasNext());
}
private void processEntity(TenantId tenantId, TransportProtos.EdgeNotificationMsgProto edgeNotificationMsg) {
@ -336,12 +376,19 @@ public class DefaultEdgeNotificationService implements EdgeNotificationService {
}, dbCallbackExecutorService);
break;
case DELETED:
PageData<Edge> edgesByTenantId = edgeService.findEdgesByTenantId(tenantId, new PageLink(Integer.MAX_VALUE));
if (edgesByTenantId != null && edgesByTenantId.getData() != null && !edgesByTenantId.getData().isEmpty()) {
for (Edge edge : edgesByTenantId.getData()) {
saveEdgeEvent(tenantId, edge.getId(), type, actionType, entityId, null);
PageLink pageLink = new PageLink(DEFAULT_LIMIT);
PageData<Edge> pageData;
do {
pageData = edgeService.findEdgesByTenantId(tenantId, pageLink);
if (pageData != null && pageData.getData() != null && !pageData.getData().isEmpty()) {
for (Edge edge : pageData.getData()) {
saveEdgeEvent(tenantId, edge.getId(), type, actionType, entityId, null);
}
if (pageData.hasNext()) {
pageLink = pageLink.nextPageLink();
}
}
}
} while (pageData != null && pageData.hasNext());
break;
case ASSIGNED_TO_EDGE:
case UNASSIGNED_FROM_EDGE:
@ -355,53 +402,74 @@ public class DefaultEdgeNotificationService implements EdgeNotificationService {
}
private void updateDependentRuleChains(TenantId tenantId, RuleChainId processingRuleChainId, EdgeId edgeId) {
PageData<RuleChain> pageData = ruleChainService.findRuleChainsByTenantIdAndEdgeId(tenantId, edgeId, new TimePageLink(Integer.MAX_VALUE));
if (pageData != null && pageData.getData() != null && !pageData.getData().isEmpty()) {
for (RuleChain ruleChain : pageData.getData()) {
if (!ruleChain.getId().equals(processingRuleChainId)) {
List<RuleChainConnectionInfo> connectionInfos =
ruleChainService.loadRuleChainMetaData(ruleChain.getTenantId(), ruleChain.getId()).getRuleChainConnections();
if (connectionInfos != null && !connectionInfos.isEmpty()) {
for (RuleChainConnectionInfo connectionInfo : connectionInfos) {
if (connectionInfo.getTargetRuleChainId().equals(processingRuleChainId)) {
saveEdgeEvent(tenantId,
edgeId,
EdgeEventType.RULE_CHAIN_METADATA,
EdgeEventActionType.UPDATED,
ruleChain.getId(),
null);
TimePageLink pageLink = new TimePageLink(DEFAULT_LIMIT);
PageData<RuleChain> pageData;
do {
pageData = ruleChainService.findRuleChainsByTenantIdAndEdgeId(tenantId, edgeId, pageLink);
if (pageData != null && pageData.getData() != null && !pageData.getData().isEmpty()) {
for (RuleChain ruleChain : pageData.getData()) {
if (!ruleChain.getId().equals(processingRuleChainId)) {
List<RuleChainConnectionInfo> connectionInfos =
ruleChainService.loadRuleChainMetaData(ruleChain.getTenantId(), ruleChain.getId()).getRuleChainConnections();
if (connectionInfos != null && !connectionInfos.isEmpty()) {
for (RuleChainConnectionInfo connectionInfo : connectionInfos) {
if (connectionInfo.getTargetRuleChainId().equals(processingRuleChainId)) {
saveEdgeEvent(tenantId,
edgeId,
EdgeEventType.RULE_CHAIN_METADATA,
EdgeEventActionType.UPDATED,
ruleChain.getId(),
null);
}
}
}
}
}
if (pageData.hasNext()) {
pageLink = pageLink.nextPageLink();
}
}
}
} while (pageData != null && pageData.hasNext());
}
private void processAlarm(TenantId tenantId, TransportProtos.EdgeNotificationMsgProto edgeNotificationMsg) {
AlarmId alarmId = new AlarmId(new UUID(edgeNotificationMsg.getEntityIdMSB(), edgeNotificationMsg.getEntityIdLSB()));
ListenableFuture<Alarm> alarmFuture = alarmService.findAlarmByIdAsync(tenantId, alarmId);
Futures.transform(alarmFuture, alarm -> {
if (alarm != null) {
EdgeEventType type = getEdgeQueueTypeByEntityType(alarm.getOriginator().getEntityType());
if (type != null) {
ListenableFuture<List<EdgeId>> relatedEdgeIdsByEntityIdFuture = edgeService.findRelatedEdgeIdsByEntityId(tenantId, alarm.getOriginator());
Futures.transform(relatedEdgeIdsByEntityIdFuture, relatedEdgeIdsByEntityId -> {
if (relatedEdgeIdsByEntityId != null) {
for (EdgeId edgeId : relatedEdgeIdsByEntityId) {
saveEdgeEvent(tenantId,
edgeId,
EdgeEventType.ALARM,
EdgeEventActionType.valueOf(edgeNotificationMsg.getAction()),
alarmId,
null);
Futures.addCallback(alarmFuture, new FutureCallback<Alarm>() {
@Override
public void onSuccess(@Nullable Alarm alarm) {
if (alarm != null) {
EdgeEventType type = EdgeUtils.getEdgeEventTypeByEntityType(alarm.getOriginator().getEntityType());
if (type != null) {
ListenableFuture<List<EdgeId>> relatedEdgeIdsByEntityIdFuture = edgeService.findRelatedEdgeIdsByEntityId(tenantId, alarm.getOriginator());
Futures.addCallback(relatedEdgeIdsByEntityIdFuture, new FutureCallback<List<EdgeId>>() {
@Override
public void onSuccess(@Nullable List<EdgeId> relatedEdgeIdsByEntityId) {
if (relatedEdgeIdsByEntityId != null) {
for (EdgeId edgeId : relatedEdgeIdsByEntityId) {
saveEdgeEvent(tenantId,
edgeId,
EdgeEventType.ALARM,
EdgeEventActionType.valueOf(edgeNotificationMsg.getAction()),
alarmId,
null);
}
}
}
}
return null;
}, dbCallbackExecutorService);
@Override
public void onFailure(Throwable t) {
log.warn("[{}] can't find related edge ids by entity id [{}]", tenantId.getId(), alarm.getOriginator(), t);
}
}, dbCallbackExecutorService);
}
}
}
return null;
@Override
public void onFailure(Throwable t) {
log.warn("[{}] can't find alarm by id [{}]", tenantId.getId(), alarmId.getId(), t);
}
}, dbCallbackExecutorService);
}
@ -413,41 +481,35 @@ public class DefaultEdgeNotificationService implements EdgeNotificationService {
futures.add(edgeService.findRelatedEdgeIdsByEntityId(tenantId, relation.getTo()));
futures.add(edgeService.findRelatedEdgeIdsByEntityId(tenantId, relation.getFrom()));
ListenableFuture<List<List<EdgeId>>> combinedFuture = Futures.allAsList(futures);
Futures.transform(combinedFuture, listOfListsEdgeIds -> {
Set<EdgeId> uniqueEdgeIds = new HashSet<>();
if (listOfListsEdgeIds != null && !listOfListsEdgeIds.isEmpty()) {
for (List<EdgeId> listOfListsEdgeId : listOfListsEdgeIds) {
if (listOfListsEdgeId != null) {
uniqueEdgeIds.addAll(listOfListsEdgeId);
Futures.addCallback(combinedFuture, new FutureCallback<List<List<EdgeId>>>() {
@Override
public void onSuccess(@Nullable List<List<EdgeId>> listOfListsEdgeIds) {
Set<EdgeId> uniqueEdgeIds = new HashSet<>();
if (listOfListsEdgeIds != null && !listOfListsEdgeIds.isEmpty()) {
for (List<EdgeId> listOfListsEdgeId : listOfListsEdgeIds) {
if (listOfListsEdgeId != null) {
uniqueEdgeIds.addAll(listOfListsEdgeId);
}
}
}
}
if (!uniqueEdgeIds.isEmpty()) {
for (EdgeId edgeId : uniqueEdgeIds) {
saveEdgeEvent(tenantId,
edgeId,
EdgeEventType.RELATION,
EdgeEventActionType.valueOf(edgeNotificationMsg.getAction()),
null,
mapper.valueToTree(relation));
if (!uniqueEdgeIds.isEmpty()) {
for (EdgeId edgeId : uniqueEdgeIds) {
saveEdgeEvent(tenantId,
edgeId,
EdgeEventType.RELATION,
EdgeEventActionType.valueOf(edgeNotificationMsg.getAction()),
null,
mapper.valueToTree(relation));
}
}
}
return null;
}, dbCallbackExecutorService);
}
}
private EdgeEventType getEdgeQueueTypeByEntityType(EntityType entityType) {
switch (entityType) {
case DEVICE:
return EdgeEventType.DEVICE;
case ASSET:
return EdgeEventType.ASSET;
case ENTITY_VIEW:
return EdgeEventType.ENTITY_VIEW;
default:
log.debug("Unsupported entity type: [{}]", entityType);
return null;
@Override
public void onFailure(Throwable t) {
log.warn("[{}] can't find related edge ids by relation to id [{}] and relation from id [{}]" ,
tenantId.getId(), relation.getTo().getId(), relation.getFrom().getId(), t);
}
}, dbCallbackExecutorService);
}
}
}

91
application/src/main/java/org/thingsboard/server/service/edge/rpc/EdgeGrpcService.java
View File

@ -26,6 +26,7 @@ import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.stereotype.Service;
import org.thingsboard.common.util.ThingsBoardThreadFactory;
import org.thingsboard.server.common.data.DataConstants;
import org.thingsboard.server.common.data.edge.Edge;
import org.thingsboard.server.common.data.id.EdgeId;
@ -48,9 +49,12 @@ import java.io.File;
import java.io.IOException;
import java.util.Collections;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;
@Service
@ -59,7 +63,9 @@ import java.util.concurrent.TimeUnit;
@TbCoreComponent
public class EdgeGrpcService extends EdgeRpcServiceGrpc.EdgeRpcServiceImplBase implements EdgeRpcService {
private final Map<EdgeId, EdgeGrpcSession> sessions = new ConcurrentHashMap<>();
private final ConcurrentMap<EdgeId, EdgeGrpcSession> sessions = new ConcurrentHashMap<>();
private final ConcurrentMap<EdgeId, Boolean> sessionNewEvents = new ConcurrentHashMap<>();
private final ConcurrentMap<EdgeId, ScheduledFuture<?>> sessionEdgeEventChecks = new ConcurrentHashMap<>();
private static final ObjectMapper mapper = new ObjectMapper();
@Value("${edges.rpc.port}")
@ -75,6 +81,9 @@ public class EdgeGrpcService extends EdgeRpcServiceGrpc.EdgeRpcServiceImplBase i
@Value("${edges.rpc.client_max_keep_alive_time_sec}")
private int clientMaxKeepAliveTimeSec;
@Value("${edges.scheduler_pool_size}")
private int schedulerPoolSize;
@Autowired
private EdgeContextComponent ctx;
@ -83,7 +92,7 @@ public class EdgeGrpcService extends EdgeRpcServiceGrpc.EdgeRpcServiceImplBase i
private Server server;
private ExecutorService executor;
private ScheduledExecutorService scheduler;
@PostConstruct
public void init() {
@ -109,9 +118,8 @@ public class EdgeGrpcService extends EdgeRpcServiceGrpc.EdgeRpcServiceImplBase i
log.error("Failed to start Edge RPC server!", e);
throw new RuntimeException("Failed to start Edge RPC server!");
}
this.scheduler = Executors.newScheduledThreadPool(schedulerPoolSize, ThingsBoardThreadFactory.forName("edge-scheduler"));
log.info("Edge RPC service initialized!");
executor = Executors.newSingleThreadExecutor();
processHandleMessages();
}
@PreDestroy
@ -119,8 +127,16 @@ public class EdgeGrpcService extends EdgeRpcServiceGrpc.EdgeRpcServiceImplBase i
if (server != null) {
server.shutdownNow();
}
if (executor != null) {
executor.shutdownNow();
for (Map.Entry<EdgeId, ScheduledFuture<?>> entry : sessionEdgeEventChecks.entrySet()) {
EdgeId edgeId = entry.getKey();
ScheduledFuture<?> sessionEdgeEventCheck = entry.getValue();
if (sessionEdgeEventCheck != null && !sessionEdgeEventCheck.isCancelled() && !sessionEdgeEventCheck.isDone()) {
sessionEdgeEventCheck.cancel(true);
sessionEdgeEventChecks.remove(edgeId);
}
}
if (scheduler != null) {
scheduler.shutdownNow();
}
}
@ -147,14 +163,27 @@ public class EdgeGrpcService extends EdgeRpcServiceGrpc.EdgeRpcServiceImplBase i
log.debug("Closing and removing session for edge [{}]", edgeId);
session.close();
sessions.remove(edgeId);
sessionNewEvents.remove(edgeId);
cancelScheduleEdgeEventsCheck(edgeId);
}
}
@Override
public void onEdgeEvent(EdgeId edgeId) {
log.trace("[{}] onEdgeEvent", edgeId.getId());
if (!sessionNewEvents.get(edgeId)) {
log.trace("[{}] set session new events flag to true", edgeId.getId());
sessionNewEvents.put(edgeId, true);
}
}
private void onEdgeConnect(EdgeId edgeId, EdgeGrpcSession edgeGrpcSession) {
log.debug("[{}] onEdgeConnect [{}]", edgeId, edgeGrpcSession.getSessionId());
sessions.put(edgeId, edgeGrpcSession);
sessionNewEvents.put(edgeId, false);
save(edgeId, DefaultDeviceStateService.ACTIVITY_STATE, true);
save(edgeId, DefaultDeviceStateService.LAST_CONNECT_TIME, System.currentTimeMillis());
scheduleEdgeEventsCheck(edgeGrpcSession);
}
public EdgeGrpcSession getEdgeGrpcSessionById(EdgeId edgeId) {
@ -166,37 +195,47 @@ public class EdgeGrpcService extends EdgeRpcServiceGrpc.EdgeRpcServiceImplBase i
}
}
private void processHandleMessages() {
executor.submit(() -> {
while (!Thread.interrupted()) {
private void scheduleEdgeEventsCheck(EdgeGrpcSession session) {
EdgeId edgeId = session.getEdge().getId();
UUID tenantId = session.getEdge().getTenantId().getId();
if (sessions.containsKey(edgeId)) {
ScheduledFuture<?> schedule = scheduler.schedule(() -> {
try {
if (sessions.size() > 0) {
for (EdgeGrpcSession session : sessions.values()) {
session.processHandleMessages();
}
} else {
log.trace("No sessions available, sleep for the next run");
try {
Thread.sleep(1000);
} catch (InterruptedException ignore) {
}
if (sessionNewEvents.get(edgeId)) {
log.trace("[{}] Set session new events flag to false", edgeId.getId());
sessionNewEvents.put(edgeId, false);
session.processEdgeEvents();
}
} catch (Exception e) {
log.warn("Failed to process messages handling!", e);
try {
Thread.sleep(1000);
} catch (InterruptedException ignore) {
}
log.warn("[{}] Failed to process edge events for edge [{}]!", tenantId, session.getEdge().getId().getId(), e);
}
scheduleEdgeEventsCheck(session);
}, ctx.getEdgeEventStorageSettings().getNoRecordsSleepInterval(), TimeUnit.MILLISECONDS);
sessionEdgeEventChecks.put(edgeId, schedule);
log.trace("[{}] Check edge event was scheduler for edge [{}]", tenantId, edgeId.getId());
} else {
log.debug("[{}] Session was removed and edge event check schedule must not be started [{}]",
tenantId, edgeId.getId());
}
}
private void cancelScheduleEdgeEventsCheck(EdgeId edgeId) {
if (sessionEdgeEventChecks.containsKey(edgeId)) {
ScheduledFuture<?> sessionEdgeEventCheck = sessionEdgeEventChecks.get(edgeId);
if (sessionEdgeEventCheck != null && !sessionEdgeEventCheck.isCancelled() && !sessionEdgeEventCheck.isDone()) {
sessionEdgeEventCheck.cancel(true);
sessionEdgeEventChecks.remove(edgeId);
}
});
}
}
private void onEdgeDisconnect(EdgeId edgeId) {
log.debug("[{}] onEdgeDisconnect", edgeId);
sessions.remove(edgeId);
sessionNewEvents.remove(edgeId);
save(edgeId, DefaultDeviceStateService.ACTIVITY_STATE, false);
save(edgeId, DefaultDeviceStateService.LAST_DISCONNECT_TIME, System.currentTimeMillis());
cancelScheduleEdgeEventsCheck(edgeId);
}
private void save(EdgeId edgeId, String key, long value) {

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

@ -68,6 +68,7 @@ import org.thingsboard.server.common.data.security.UserCredentials;
import org.thingsboard.server.common.data.widget.WidgetType;
import org.thingsboard.server.common.data.widget.WidgetsBundle;
import org.thingsboard.server.common.transport.util.JsonUtils;
import org.thingsboard.server.dao.util.mapping.JacksonUtil;
import org.thingsboard.server.gen.edge.AdminSettingsUpdateMsg;
import org.thingsboard.server.gen.edge.AlarmUpdateMsg;
import org.thingsboard.server.gen.edge.AssetUpdateMsg;
@ -121,7 +122,7 @@ import java.util.function.Consumer;
@Data
public final class EdgeGrpcSession implements Closeable {
private static final ReentrantLock responseMsgLock = new ReentrantLock();
private static final ReentrantLock downlinkMsgLock = new ReentrantLock();
private static final String QUEUE_START_TS_ATTR_KEY = "queueStartTs";
@ -200,7 +201,7 @@ public final class EdgeGrpcSession implements Closeable {
@Override
public void onSuccess(@Nullable List<Void> result) {
UplinkResponseMsg uplinkResponseMsg = UplinkResponseMsg.newBuilder().setSuccess(true).build();
sendResponseMsg(ResponseMsg.newBuilder()
sendDownlinkMsg(ResponseMsg.newBuilder()
.setUplinkResponseMsg(uplinkResponseMsg)
.build());
}
@ -208,7 +209,7 @@ public final class EdgeGrpcSession implements Closeable {
@Override
public void onFailure(Throwable t) {
UplinkResponseMsg uplinkResponseMsg = UplinkResponseMsg.newBuilder().setSuccess(false).setErrorMsg(t.getMessage()).build();
sendResponseMsg(ResponseMsg.newBuilder()
sendDownlinkMsg(ResponseMsg.newBuilder()
.setUplinkResponseMsg(uplinkResponseMsg)
.build());
}
@ -228,38 +229,35 @@ public final class EdgeGrpcSession implements Closeable {
}
}
private void sendResponseMsg(ResponseMsg responseMsg) {
log.trace("[{}] Sending response msg [{}]", this.sessionId, responseMsg);
private void sendDownlinkMsg(ResponseMsg downlinkMsg) {
log.trace("[{}] Sending downlink msg [{}]", this.sessionId, downlinkMsg);
if (isConnected()) {
try {
responseMsgLock.lock();
outputStream.onNext(responseMsg);
downlinkMsgLock.lock();
outputStream.onNext(downlinkMsg);
} catch (Exception e) {
log.error("[{}] Failed to send response message [{}]", this.sessionId, responseMsg, e);
log.error("[{}] Failed to send downlink message [{}]", this.sessionId, downlinkMsg, e);
connected = false;
sessionCloseListener.accept(edge.getId());
} finally {
responseMsgLock.unlock();
downlinkMsgLock.unlock();
}
log.trace("[{}] Response msg successfully sent [{}]", this.sessionId, responseMsg);
log.trace("[{}] Response msg successfully sent [{}]", this.sessionId, downlinkMsg);
}
}
void onConfigurationUpdate(Edge edge) {
log.debug("[{}] onConfigurationUpdate [{}]", this.sessionId, edge);
try {
this.edge = edge;
EdgeUpdateMsg edgeConfig = EdgeUpdateMsg.newBuilder()
.setConfiguration(constructEdgeConfigProto(edge)).build();
outputStream.onNext(ResponseMsg.newBuilder()
.setEdgeUpdateMsg(edgeConfig)
.build());
} catch (Exception e) {
log.error("[{}] Failed to construct proto objects!", this.sessionId, e);
}
this.edge = edge;
EdgeUpdateMsg edgeConfig = EdgeUpdateMsg.newBuilder()
.setConfiguration(constructEdgeConfigProto(edge)).build();
ResponseMsg edgeConfigMsg = ResponseMsg.newBuilder()
.setEdgeUpdateMsg(edgeConfig)
.build();
sendDownlinkMsg(edgeConfigMsg);
}
void processHandleMessages() throws ExecutionException, InterruptedException {
void processEdgeEvents() throws ExecutionException, InterruptedException {
log.trace("[{}] processHandleMessages started", this.sessionId);
if (isConnected()) {
Long queueStartTs = getQueueStartTs().get();
@ -282,7 +280,7 @@ public final class EdgeGrpcSession implements Closeable {
latch = new CountDownLatch(downlinkMsgsPack.size());
for (DownlinkMsg downlinkMsg : downlinkMsgsPack) {
sendResponseMsg(ResponseMsg.newBuilder()
sendDownlinkMsg(ResponseMsg.newBuilder()
.setDownlinkMsg(downlinkMsg)
.build());
}
@ -310,11 +308,6 @@ public final class EdgeGrpcSession implements Closeable {
Long newStartTs = Uuids.unixTimestamp(ifOffset);
updateQueueStartTs(newStartTs);
}
try {
Thread.sleep(ctx.getEdgeEventStorageSettings().getNoRecordsSleepInterval());
} catch (InterruptedException e) {
log.error("[{}] Error during sleep between no records interval", this.sessionId, e);
}
}
log.trace("[{}] processHandleMessages finished", this.sessionId);
}
@ -446,6 +439,9 @@ public final class EdgeGrpcSession implements Closeable {
case CUSTOMER:
entityId = new CustomerId(edgeEvent.getEntityId());
break;
case EDGE:
entityId = new EdgeId(edgeEvent.getEntityId());
break;
}
DownlinkMsg downlinkMsg = null;
if (entityId != null) {
@ -969,17 +965,24 @@ public final class EdgeGrpcSession implements Closeable {
}
private EdgeConfiguration constructEdgeConfigProto(Edge edge) {
return EdgeConfiguration.newBuilder()
EdgeConfiguration.Builder builder = EdgeConfiguration.newBuilder()
.setEdgeIdMSB(edge.getId().getId().getMostSignificantBits())
.setEdgeIdLSB(edge.getId().getId().getLeastSignificantBits())
.setTenantIdMSB(edge.getTenantId().getId().getMostSignificantBits())
.setTenantIdLSB(edge.getTenantId().getId().getLeastSignificantBits())
.setName(edge.getName())
.setRoutingKey(edge.getRoutingKey())
.setType(edge.getType())
.setRoutingKey(edge.getRoutingKey())
.setSecret(edge.getSecret())
.setEdgeLicenseKey(edge.getEdgeLicenseKey())
.setCloudEndpoint(edge.getCloudEndpoint())
.setCloudType("CE")
.setConfiguration(JacksonUtil.toString(edge.getConfiguration()))
.setCloudType("CE");
if (edge.getCustomerId() != null) {
builder.setCustomerIdMSB(edge.getCustomerId().getId().getMostSignificantBits())
.setCustomerIdLSB(edge.getCustomerId().getId().getLeastSignificantBits());
}
return builder
.build();
}

2
application/src/main/java/org/thingsboard/server/service/edge/rpc/EdgeRpcService.java
View File

@ -23,4 +23,6 @@ public interface EdgeRpcService {
void updateEdge(Edge edge);
void deleteEdge(EdgeId edgeId);
void onEdgeEvent(EdgeId edgeId);
}

252
application/src/main/java/org/thingsboard/server/service/edge/rpc/init/DefaultSyncEdgeService.java
View File

@ -35,6 +35,7 @@ import org.thingsboard.server.common.data.AdminSettings;
import org.thingsboard.server.common.data.DashboardInfo;
import org.thingsboard.server.common.data.DataConstants;
import org.thingsboard.server.common.data.Device;
import org.thingsboard.server.common.data.EdgeUtils;
import org.thingsboard.server.common.data.EntityType;
import org.thingsboard.server.common.data.EntityView;
import org.thingsboard.server.common.data.User;
@ -81,6 +82,7 @@ import org.thingsboard.server.gen.edge.RelationRequestMsg;
import org.thingsboard.server.gen.edge.RuleChainMetadataRequestMsg;
import org.thingsboard.server.gen.edge.UserCredentialsRequestMsg;
import org.thingsboard.server.service.executors.DbCallbackExecutorService;
import org.thingsboard.server.service.queue.TbClusterService;
import java.io.File;
import java.nio.charset.StandardCharsets;
@ -98,6 +100,8 @@ public class DefaultSyncEdgeService implements SyncEdgeService {
private static final ObjectMapper mapper = new ObjectMapper();
private static final int DEFAULT_LIMIT = 100;
@Autowired
private EdgeEventService edgeEventService;
@ -137,9 +141,12 @@ public class DefaultSyncEdgeService implements SyncEdgeService {
@Autowired
private DbCallbackExecutorService dbCallbackExecutorService;
@Autowired
private TbClusterService tbClusterService;
@Override
public void sync(Edge edge) {
log.trace("[{}] staring sync process for edge [{}]", edge.getTenantId(), edge.getName());
log.trace("[{}][{}] Staring edge sync process", edge.getTenantId(), edge.getId());
try {
syncWidgetsBundleAndWidgetTypes(edge);
syncAdminSettings(edge);
@ -150,21 +157,27 @@ public class DefaultSyncEdgeService implements SyncEdgeService {
syncEntityViews(edge);
syncDashboards(edge);
} catch (Exception e) {
log.error("Exception during sync process", e);
log.error("[{}][{}] Exception during sync process", edge.getTenantId(), edge.getId(), e);
}
}
private void syncRuleChains(Edge edge) {
log.trace("[{}] syncRuleChains [{}]", edge.getTenantId(), edge.getName());
try {
PageData<RuleChain> pageData =
ruleChainService.findRuleChainsByTenantIdAndEdgeId(edge.getTenantId(), edge.getId(), new TimePageLink(Integer.MAX_VALUE));
if (pageData != null && pageData.getData() != null && !pageData.getData().isEmpty()) {
log.trace("[{}] [{}] rule chains(s) are going to be pushed to edge.", edge.getId(), pageData.getData().size());
for (RuleChain ruleChain : pageData.getData()) {
saveEdgeEvent(edge.getTenantId(), edge.getId(), EdgeEventType.RULE_CHAIN, EdgeEventActionType.ADDED, ruleChain.getId(), null);
TimePageLink pageLink = new TimePageLink(DEFAULT_LIMIT);
PageData<RuleChain> pageData;
do {
pageData = ruleChainService.findRuleChainsByTenantIdAndEdgeId(edge.getTenantId(), edge.getId(), pageLink);
if (pageData != null && pageData.getData() != null && !pageData.getData().isEmpty()) {
log.trace("[{}] [{}] rule chains(s) are going to be pushed to edge.", edge.getId(), pageData.getData().size());
for (RuleChain ruleChain : pageData.getData()) {
saveEdgeEvent(edge.getTenantId(), edge.getId(), EdgeEventType.RULE_CHAIN, EdgeEventActionType.ADDED, ruleChain.getId(), null);
}
if (pageData.hasNext()) {
pageLink = pageLink.nextPageLink();
}
}
}
} while (pageData != null && pageData.hasNext());
} catch (Exception e) {
log.error("Exception during loading edge rule chain(s) on sync!", e);
}
@ -173,14 +186,20 @@ public class DefaultSyncEdgeService implements SyncEdgeService {
private void syncDevices(Edge edge) {
log.trace("[{}] syncDevices [{}]", edge.getTenantId(), edge.getName());
try {
PageData<Device> pageData =
deviceService.findDevicesByTenantIdAndEdgeId(edge.getTenantId(), edge.getId(), new TimePageLink(Integer.MAX_VALUE));
if (pageData != null && pageData.getData() != null && !pageData.getData().isEmpty()) {
log.trace("[{}] [{}] device(s) are going to be pushed to edge.", edge.getId(), pageData.getData().size());
for (Device device : pageData.getData()) {
saveEdgeEvent(edge.getTenantId(), edge.getId(), EdgeEventType.DEVICE, EdgeEventActionType.ADDED, device.getId(), null);
TimePageLink pageLink = new TimePageLink(DEFAULT_LIMIT);
PageData<Device> pageData;
do {
pageData = deviceService.findDevicesByTenantIdAndEdgeId(edge.getTenantId(), edge.getId(), pageLink);
if (pageData != null && pageData.getData() != null && !pageData.getData().isEmpty()) {
log.trace("[{}] [{}] device(s) are going to be pushed to edge.", edge.getId(), pageData.getData().size());
for (Device device : pageData.getData()) {
saveEdgeEvent(edge.getTenantId(), edge.getId(), EdgeEventType.DEVICE, EdgeEventActionType.ADDED, device.getId(), null);
}
if (pageData.hasNext()) {
pageLink = pageLink.nextPageLink();
}
}
}
} while (pageData != null && pageData.hasNext());
} catch (Exception e) {
log.error("Exception during loading edge device(s) on sync!", e);
}
@ -189,13 +208,20 @@ public class DefaultSyncEdgeService implements SyncEdgeService {
private void syncAssets(Edge edge) {
log.trace("[{}] syncAssets [{}]", edge.getTenantId(), edge.getName());
try {
PageData<Asset> pageData = assetService.findAssetsByTenantIdAndEdgeId(edge.getTenantId(), edge.getId(), new TimePageLink(Integer.MAX_VALUE));
if (pageData != null && pageData.getData() != null && !pageData.getData().isEmpty()) {
log.trace("[{}] [{}] asset(s) are going to be pushed to edge.", edge.getId(), pageData.getData().size());
for (Asset asset : pageData.getData()) {
saveEdgeEvent(edge.getTenantId(), edge.getId(), EdgeEventType.ASSET, EdgeEventActionType.ADDED, asset.getId(), null);
TimePageLink pageLink = new TimePageLink(DEFAULT_LIMIT);
PageData<Asset> pageData;
do {
pageData = assetService.findAssetsByTenantIdAndEdgeId(edge.getTenantId(), edge.getId(), pageLink);
if (pageData != null && pageData.getData() != null && !pageData.getData().isEmpty()) {
log.trace("[{}] [{}] asset(s) are going to be pushed to edge.", edge.getId(), pageData.getData().size());
for (Asset asset : pageData.getData()) {
saveEdgeEvent(edge.getTenantId(), edge.getId(), EdgeEventType.ASSET, EdgeEventActionType.ADDED, asset.getId(), null);
}
if (pageData.hasNext()) {
pageLink = pageLink.nextPageLink();
}
}
}
} while (pageData != null && pageData.hasNext());
} catch (Exception e) {
log.error("Exception during loading edge asset(s) on sync!", e);
}
@ -204,13 +230,20 @@ public class DefaultSyncEdgeService implements SyncEdgeService {
private void syncEntityViews(Edge edge) {
log.trace("[{}] syncEntityViews [{}]", edge.getTenantId(), edge.getName());
try {
PageData<EntityView> pageData = entityViewService.findEntityViewsByTenantIdAndEdgeId(edge.getTenantId(), edge.getId(), new TimePageLink(Integer.MAX_VALUE));
if (pageData != null && pageData.getData() != null && !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()) {
saveEdgeEvent(edge.getTenantId(), edge.getId(), EdgeEventType.ENTITY_VIEW, EdgeEventActionType.ADDED, entityView.getId(), null);
TimePageLink pageLink = new TimePageLink(DEFAULT_LIMIT);
PageData<EntityView> pageData;
do {
pageData = entityViewService.findEntityViewsByTenantIdAndEdgeId(edge.getTenantId(), edge.getId(), pageLink);
if (pageData != null && pageData.getData() != null && !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()) {
saveEdgeEvent(edge.getTenantId(), edge.getId(), EdgeEventType.ENTITY_VIEW, EdgeEventActionType.ADDED, entityView.getId(), null);
}
if (pageData.hasNext()) {
pageLink = pageLink.nextPageLink();
}
}
}
} while (pageData != null && pageData.hasNext());
} catch (Exception e) {
log.error("Exception during loading edge entity view(s) on sync!", e);
}
@ -219,13 +252,20 @@ public class DefaultSyncEdgeService implements SyncEdgeService {
private void syncDashboards(Edge edge) {
log.trace("[{}] syncDashboards [{}]", edge.getTenantId(), edge.getName());
try {
PageData<DashboardInfo> pageData = dashboardService.findDashboardsByTenantIdAndEdgeId(edge.getTenantId(), edge.getId(), new TimePageLink(Integer.MAX_VALUE));
if (pageData != null && pageData.getData() != null && !pageData.getData().isEmpty()) {
log.trace("[{}] [{}] dashboard(s) are going to be pushed to edge.", edge.getId(), pageData.getData().size());
for (DashboardInfo dashboardInfo : pageData.getData()) {
saveEdgeEvent(edge.getTenantId(), edge.getId(), EdgeEventType.DASHBOARD, EdgeEventActionType.ADDED, dashboardInfo.getId(), null);
TimePageLink pageLink = new TimePageLink(DEFAULT_LIMIT);
PageData<DashboardInfo> pageData;
do {
pageData = dashboardService.findDashboardsByTenantIdAndEdgeId(edge.getTenantId(), edge.getId(), pageLink);
if (pageData != null && pageData.getData() != null && !pageData.getData().isEmpty()) {
log.trace("[{}] [{}] dashboard(s) are going to be pushed to edge.", edge.getId(), pageData.getData().size());
for (DashboardInfo dashboardInfo : pageData.getData()) {
saveEdgeEvent(edge.getTenantId(), edge.getId(), EdgeEventType.DASHBOARD, EdgeEventActionType.ADDED, dashboardInfo.getId(), null);
}
if (pageData.hasNext()) {
pageLink = pageLink.nextPageLink();
}
}
}
} while (pageData != null && pageData.hasNext());
} catch (Exception e) {
log.error("Exception during loading edge dashboard(s) on sync!", e);
}
@ -234,18 +274,36 @@ public class DefaultSyncEdgeService implements SyncEdgeService {
private void syncUsers(Edge edge) {
log.trace("[{}] syncUsers [{}]", edge.getTenantId(), edge.getName());
try {
PageData<User> pageData = userService.findTenantAdmins(edge.getTenantId(), new PageLink(Integer.MAX_VALUE));
pushUsersToEdge(pageData, edge);
if (edge.getCustomerId() != null && !EntityId.NULL_UUID.equals(edge.getCustomerId().getId())) {
saveEdgeEvent(edge.getTenantId(), edge.getId(), EdgeEventType.CUSTOMER, EdgeEventActionType.ADDED, edge.getCustomerId(), null);
pageData = userService.findCustomerUsers(edge.getTenantId(), edge.getCustomerId(), new PageLink(Integer.MAX_VALUE));
TimePageLink pageLink = new TimePageLink(DEFAULT_LIMIT);
PageData<User> pageData;
do {
pageData = userService.findTenantAdmins(edge.getTenantId(), pageLink);
pushUsersToEdge(pageData, edge);
}
if (pageData.hasNext()) {
pageLink = pageLink.nextPageLink();
}
} while (pageData.hasNext());
syncCustomerUsers(edge);
} catch (Exception e) {
log.error("Exception during loading edge user(s) on sync!", e);
}
}
private void syncCustomerUsers(Edge edge) {
if (edge.getCustomerId() != null && !EntityId.NULL_UUID.equals(edge.getCustomerId().getId())) {
saveEdgeEvent(edge.getTenantId(), edge.getId(), EdgeEventType.CUSTOMER, EdgeEventActionType.ADDED, edge.getCustomerId(), null);
TimePageLink pageLink = new TimePageLink(DEFAULT_LIMIT);
PageData<User> pageData;
do {
pageData = userService.findCustomerUsers(edge.getTenantId(), edge.getCustomerId(), pageLink);
pushUsersToEdge(pageData, edge);
if (pageData != null && pageData.hasNext()) {
pageLink = pageLink.nextPageLink();
}
} while (pageData != null && pageData.hasNext());
}
}
private void syncWidgetsBundleAndWidgetTypes(Edge edge) {
log.trace("[{}] syncWidgetsBundleAndWidgetTypes [{}]", edge.getTenantId(), edge.getName());
List<WidgetsBundle> widgetsBundlesToPush = new ArrayList<>();
@ -372,10 +430,11 @@ public class DefaultSyncEdgeService implements SyncEdgeService {
EntityId entityId = EntityIdFactory.getByTypeAndUuid(
EntityType.valueOf(attributesRequestMsg.getEntityType()),
new UUID(attributesRequestMsg.getEntityIdMSB(), attributesRequestMsg.getEntityIdLSB()));
final EdgeEventType type = getEdgeQueueTypeByEntityType(entityId.getEntityType());
final EdgeEventType type = EdgeUtils.getEdgeEventTypeByEntityType(entityId.getEntityType());
if (type != null) {
SettableFuture<Void> futureToSet = SettableFuture.create();
ListenableFuture<List<AttributeKvEntry>> ssAttrFuture = attributesService.findAll(edge.getTenantId(), entityId, DataConstants.SERVER_SCOPE);
String scope = attributesRequestMsg.getScope();
ListenableFuture<List<AttributeKvEntry>> ssAttrFuture = attributesService.findAll(edge.getTenantId(), entityId, scope);
Futures.addCallback(ssAttrFuture, new FutureCallback<List<AttributeKvEntry>>() {
@Override
public void onSuccess(@Nullable List<AttributeKvEntry> ssAttributes) {
@ -395,7 +454,7 @@ public class DefaultSyncEdgeService implements SyncEdgeService {
}
}
entityData.put("kv", attributes);
entityData.put("scope", DataConstants.SERVER_SCOPE);
entityData.put("scope", scope);
JsonNode body = mapper.valueToTree(entityData);
log.debug("Sending attributes data msg, entityId [{}], attributes [{}]", entityId, body);
saveEdgeEvent(edge.getTenantId(),
@ -408,6 +467,11 @@ public class DefaultSyncEdgeService implements SyncEdgeService {
log.error("[{}] Failed to send attribute updates to the edge", edge.getName(), e);
throw new RuntimeException("[" + edge.getName() + "] Failed to send attribute updates to the edge", e);
}
} else {
log.trace("[{}][{}] No attributes found for entity {} [{}]", edge.getTenantId(),
edge.getName(),
entityId.getEntityType(),
entityId.getId());
}
futureToSet.set(null);
}
@ -419,27 +483,12 @@ public class DefaultSyncEdgeService implements SyncEdgeService {
}
}, dbCallbackExecutorService);
return futureToSet;
// TODO: voba - push shared attributes to edge?
// ListenableFuture<List<AttributeKvEntry>> shAttrFuture = attributesService.findAll(edge.getTenantId(), entityId, DataConstants.SHARED_SCOPE);
// ListenableFuture<List<AttributeKvEntry>> clAttrFuture = attributesService.findAll(edge.getTenantId(), entityId, DataConstants.CLIENT_SCOPE);
} else {
log.warn("[{}] Type doesn't supported {}", edge.getTenantId(), entityId.getEntityType());
return Futures.immediateFuture(null);
}
}
private EdgeEventType getEdgeQueueTypeByEntityType(EntityType entityType) {
switch (entityType) {
case DEVICE:
return EdgeEventType.DEVICE;
case ASSET:
return EdgeEventType.ASSET;
case ENTITY_VIEW:
return EdgeEventType.ENTITY_VIEW;
default:
return null;
}
}
@Override
public ListenableFuture<Void> processRelationRequestMsg(Edge edge, RelationRequestMsg relationRequestMsg) {
log.trace("[{}] processRelationRequestMsg [{}][{}]", edge.getTenantId(), edge.getName(), relationRequestMsg);
@ -451,34 +500,47 @@ public class DefaultSyncEdgeService implements SyncEdgeService {
futures.add(findRelationByQuery(edge, entityId, EntitySearchDirection.FROM));
futures.add(findRelationByQuery(edge, entityId, EntitySearchDirection.TO));
ListenableFuture<List<List<EntityRelation>>> relationsListFuture = Futures.allAsList(futures);
return Futures.transform(relationsListFuture, relationsList -> {
try {
if (relationsList != null && !relationsList.isEmpty()) {
for (List<EntityRelation> entityRelations : relationsList) {
log.trace("[{}] [{}] [{}] relation(s) are going to be pushed to edge.", edge.getId(), entityId, entityRelations.size());
for (EntityRelation relation : entityRelations) {
try {
if (!relation.getFrom().getEntityType().equals(EntityType.EDGE) &&
!relation.getTo().getEntityType().equals(EntityType.EDGE)) {
saveEdgeEvent(edge.getTenantId(),
edge.getId(),
EdgeEventType.RELATION,
EdgeEventActionType.ADDED,
null,
mapper.valueToTree(relation));
SettableFuture<Void> futureToSet = SettableFuture.create();
Futures.addCallback(relationsListFuture, new FutureCallback<List<List<EntityRelation>>>() {
@Override
public void onSuccess(@Nullable List<List<EntityRelation>> relationsList) {
try {
if (relationsList != null && !relationsList.isEmpty()) {
for (List<EntityRelation> entityRelations : relationsList) {
log.trace("[{}] [{}] [{}] relation(s) are going to be pushed to edge.", edge.getId(), entityId, entityRelations.size());
for (EntityRelation relation : entityRelations) {
try {
if (!relation.getFrom().getEntityType().equals(EntityType.EDGE) &&
!relation.getTo().getEntityType().equals(EntityType.EDGE)) {
saveEdgeEvent(edge.getTenantId(),
edge.getId(),
EdgeEventType.RELATION,
EdgeEventActionType.ADDED,
null,
mapper.valueToTree(relation));
}
} catch (Exception e) {
log.error("Exception during loading relation [{}] to edge on sync!", relation, e);
futureToSet.setException(e);
return;
}
}
}
} catch (Exception e) {
log.error("Exception during loading relation [{}] to edge on sync!", relation, e);
}
futureToSet.set(null);
} catch (Exception e) {
log.error("Exception during loading relation(s) to edge on sync!", e);
futureToSet.setException(e);
}
}
}
} catch (Exception e) {
log.error("Exception during loading relation(s) to edge on sync!", e);
throw new RuntimeException("Exception during loading relation(s) to edge on sync!", e);
}
return null;
}, dbCallbackExecutorService);
@Override
public void onFailure(Throwable t) {
log.error("[{}] Can't find relation by query. Entity id [{}]", edge.getTenantId(), entityId, t);
futureToSet.setException(t);
}
}, dbCallbackExecutorService);
return futureToSet;
}
private ListenableFuture<List<EntityRelation>> findRelationByQuery(Edge edge, EntityId entityId, EntitySearchDirection direction) {
@ -534,11 +596,11 @@ public class DefaultSyncEdgeService implements SyncEdgeService {
}
private ListenableFuture<EdgeEvent> saveEdgeEvent(TenantId tenantId,
EdgeId edgeId,
EdgeEventType type,
EdgeEventActionType action,
EntityId entityId,
JsonNode body) {
EdgeId edgeId,
EdgeEventType type,
EdgeEventActionType action,
EntityId entityId,
JsonNode body) {
log.trace("Pushing edge event to edge queue. tenantId [{}], edgeId [{}], type [{}], action[{}], entityId [{}], body [{}]",
tenantId, edgeId, type, action, entityId, body);
@ -551,6 +613,18 @@ public class DefaultSyncEdgeService implements SyncEdgeService {
edgeEvent.setEntityId(entityId.getId());
}
edgeEvent.setBody(body);
return edgeEventService.saveAsync(edgeEvent);
ListenableFuture<EdgeEvent> future = edgeEventService.saveAsync(edgeEvent);
Futures.addCallback(future, new FutureCallback<EdgeEvent>() {
@Override
public void onSuccess(@Nullable EdgeEvent result) {
tbClusterService.onEdgeEventUpdate(tenantId, edgeId);
}
@Override
public void onFailure(Throwable t) {
log.warn("[{}] Can't save edge event [{}] for edge [{}]", tenantId.getId(), edgeEvent, edgeId.getId(), t);
}
}, dbCallbackExecutorService);
return future;
}
}

17
application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/BaseProcessor.java
View File

@ -17,8 +17,11 @@ package org.thingsboard.server.service.edge.rpc.processor;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.google.common.util.concurrent.FutureCallback;
import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.ListenableFuture;
import lombok.extern.slf4j.Slf4j;
import org.checkerframework.checker.nullness.qual.Nullable;
import org.springframework.beans.factory.annotation.Autowired;
import org.thingsboard.server.common.data.edge.EdgeEvent;
import org.thingsboard.server.common.data.edge.EdgeEventActionType;
@ -107,6 +110,18 @@ public abstract class BaseProcessor {
edgeEvent.setEntityId(entityId.getId());
}
edgeEvent.setBody(body);
return edgeEventService.saveAsync(edgeEvent);
ListenableFuture<EdgeEvent> future = edgeEventService.saveAsync(edgeEvent);
Futures.addCallback(future, new FutureCallback<EdgeEvent>() {
@Override
public void onSuccess(@Nullable EdgeEvent result) {
tbClusterService.onEdgeEventUpdate(tenantId, edgeId);
}
@Override
public void onFailure(Throwable t) {
log.warn("[{}] Can't save edge event [{}] for edge [{}]", tenantId.getId(), edgeEvent, edgeId.getId(), t);
}
}, dbCallbackExecutorService);
return future;
}
}

20
application/src/main/java/org/thingsboard/server/service/queue/DefaultTbClusterService.java
View File

@ -30,11 +30,13 @@ import org.thingsboard.server.common.data.Tenant;
import org.thingsboard.server.common.data.TenantProfile;
import org.thingsboard.server.common.data.id.DeviceId;
import org.thingsboard.server.common.data.id.DeviceProfileId;
import org.thingsboard.server.common.data.id.EdgeId;
import org.thingsboard.server.common.data.id.EntityId;
import org.thingsboard.server.common.data.id.RuleChainId;
import org.thingsboard.server.common.data.id.TenantId;
import org.thingsboard.server.common.data.plugin.ComponentLifecycleEvent;
import org.thingsboard.server.common.msg.TbMsg;
import org.thingsboard.server.common.msg.edge.EdgeEventUpdateMsg;
import org.thingsboard.server.common.msg.plugin.ComponentLifecycleMsg;
import org.thingsboard.server.common.msg.queue.ServiceType;
import org.thingsboard.server.common.msg.queue.TopicPartitionInfo;
@ -279,6 +281,21 @@ public class DefaultTbClusterService implements TbClusterService {
}
}
@Override
public void onEdgeEventUpdate(TenantId tenantId, EdgeId edgeId) {
log.trace("[{}] Processing edge {} event update ", tenantId, edgeId);
EdgeEventUpdateMsg msg = new EdgeEventUpdateMsg(tenantId, edgeId);
byte[] msgBytes = encodingService.encode(msg);
TbQueueProducer<TbProtoQueueMsg<ToCoreNotificationMsg>> toCoreNfProducer = producerProvider.getTbCoreNotificationsMsgProducer();
Set<String> tbCoreServices = partitionService.getAllServiceIds(ServiceType.TB_CORE);
for (String serviceId : tbCoreServices) {
TopicPartitionInfo tpi = partitionService.getNotificationsTopic(ServiceType.TB_CORE, serviceId);
ToCoreNotificationMsg toCoreMsg = ToCoreNotificationMsg.newBuilder().setEdgeEventUpdateMsg(ByteString.copyFrom(msgBytes)).build();
toCoreNfProducer.send(tpi, new TbProtoQueueMsg<>(msg.getEdgeId().getId(), toCoreMsg), null);
toCoreNfs.incrementAndGet();
}
}
private void broadcast(ComponentLifecycleMsg msg) {
byte[] msgBytes = encodingService.encode(msg);
TbQueueProducer<TbProtoQueueMsg<ToRuleEngineNotificationMsg>> toRuleEngineProducer = producerProvider.getRuleEngineNotificationsMsgProducer();
@ -286,7 +303,8 @@ public class DefaultTbClusterService implements TbClusterService {
if (msg.getEntityId().getEntityType().equals(EntityType.TENANT)
|| msg.getEntityId().getEntityType().equals(EntityType.TENANT_PROFILE)
|| msg.getEntityId().getEntityType().equals(EntityType.DEVICE_PROFILE)
|| msg.getEntityId().getEntityType().equals(EntityType.API_USAGE_STATE)) {
|| msg.getEntityId().getEntityType().equals(EntityType.API_USAGE_STATE)
|| msg.getEntityId().getEntityType().equals(EntityType.EDGE)) {
TbQueueProducer<TbProtoQueueMsg<ToCoreNotificationMsg>> toCoreNfProducer = producerProvider.getTbCoreNotificationsMsgProducer();
Set<String> tbCoreServices = partitionService.getAllServiceIds(ServiceType.TB_CORE);
for (String serviceId : tbCoreServices) {

7
application/src/main/java/org/thingsboard/server/service/queue/DefaultTbCoreConsumerService.java
View File

@ -280,6 +280,13 @@ public class DefaultTbCoreConsumerService extends AbstractConsumerService<ToCore
} else if (toCoreNotification.getComponentLifecycleMsg() != null && !toCoreNotification.getComponentLifecycleMsg().isEmpty()) {
handleComponentLifecycleMsg(id, toCoreNotification.getComponentLifecycleMsg());
callback.onSuccess();
} else if (toCoreNotification.getEdgeEventUpdateMsg() != null && !toCoreNotification.getEdgeEventUpdateMsg().isEmpty()) {
Optional<TbActorMsg> actorMsg = encodingService.decode(toCoreNotification.getEdgeEventUpdateMsg().toByteArray());
if (actorMsg.isPresent()) {
log.trace("[{}] Forwarding message to App Actor {}", id, actorMsg.get());
actorContext.tellWithHighPriority(actorMsg.get());
}
callback.onSuccess();
}
if (statsEnabled) {
stats.log(toCoreNotification);

3
application/src/main/java/org/thingsboard/server/service/queue/TbClusterService.java
View File

@ -21,6 +21,7 @@ import org.thingsboard.server.common.data.Device;
import org.thingsboard.server.common.data.DeviceProfile;
import org.thingsboard.server.common.data.Tenant;
import org.thingsboard.server.common.data.TenantProfile;
import org.thingsboard.server.common.data.id.EdgeId;
import org.thingsboard.server.common.data.id.EntityId;
import org.thingsboard.server.common.data.id.TenantId;
import org.thingsboard.server.common.data.plugin.ComponentLifecycleEvent;
@ -71,4 +72,6 @@ public interface TbClusterService {
void onDeviceChange(Device device, TbQueueCallback callback);
void onDeviceDeleted(Device device, TbQueueCallback callback);
void onEdgeEventUpdate(TenantId tenantId, EdgeId edgeId);
}

1
application/src/main/java/org/thingsboard/server/service/rpc/DefaultTbRuleEngineRpcService.java
View File

@ -52,7 +52,6 @@ public class DefaultTbRuleEngineRpcService implements TbRuleEngineDeviceRpcServi
private final TbClusterService clusterService;
private final TbServiceInfoProvider serviceInfoProvider;
private final ConcurrentMap<UUID, Consumer<FromDeviceRpcResponse>> toDeviceRpcRequests = new ConcurrentHashMap<>();
private Optional<TbCoreDeviceRpcService> tbCoreRpcService;

1
application/src/main/resources/thingsboard.yml
View File

@ -589,6 +589,7 @@ edges:
max_read_records_count: "${EDGES_RPC_STORAGE_MAX_READ_RECORDS_COUNT:50}"
no_read_records_sleep: "${EDGES_RPC_NO_READ_RECORDS_SLEEP:1000}"
sleep_between_batches: "${EDGES_RPC_SLEEP_BETWEEN_BATCHES:1000}"
scheduler_pool_size: "${EDGES_SCHEDULER_POOL_SIZE:4}"
edge_events_ttl: "${EDGES_EDGE_EVENTS_TTL:0}"
state:
persistToTelemetry: "${EDGES_PERSIST_STATE_TO_TELEMETRY:false}"

27
application/src/test/java/org/thingsboard/server/edge/BaseEdgeTest.java
View File

@ -98,6 +98,7 @@ import org.thingsboard.server.gen.edge.UserCredentialsUpdateMsg;
import org.thingsboard.server.gen.edge.WidgetTypeUpdateMsg;
import org.thingsboard.server.gen.edge.WidgetsBundleUpdateMsg;
import org.thingsboard.server.gen.transport.TransportProtos;
import org.thingsboard.server.service.queue.TbClusterService;
import java.util.ArrayList;
import java.util.List;
@ -122,6 +123,9 @@ abstract public class BaseEdgeTest extends AbstractControllerTest {
@Autowired
private EdgeEventService edgeEventService;
@Autowired
private TbClusterService clusterService;
@Before
public void beforeTest() throws Exception {
loginSysAdmin();
@ -228,6 +232,7 @@ abstract public class BaseEdgeTest extends AbstractControllerTest {
EdgeEvent edgeEvent = constructEdgeEvent(tenantId, edge.getId(), EdgeEventActionType.RPC_CALL, device.getId().getId(), EdgeEventType.DEVICE, body);
edgeImitator.expectMessageAmount(1);
edgeEventService.saveAsync(edgeEvent);
clusterService.onEdgeEventUpdate(tenantId, edge.getId());
edgeImitator.waitForMessages();
AbstractMessage latestMessage = edgeImitator.getLatestMessage();
@ -845,6 +850,7 @@ abstract public class BaseEdgeTest extends AbstractControllerTest {
EdgeEvent edgeEvent1 = constructEdgeEvent(tenantId, edge.getId(), EdgeEventActionType.TIMESERIES_UPDATED, device.getId().getId(), EdgeEventType.DEVICE, timeseriesEntityData);
edgeImitator.expectMessageAmount(1);
edgeEventService.saveAsync(edgeEvent1);
clusterService.onEdgeEventUpdate(tenantId, edge.getId());
edgeImitator.waitForMessages();
AbstractMessage latestMessage = edgeImitator.getLatestMessage();
@ -883,6 +889,7 @@ abstract public class BaseEdgeTest extends AbstractControllerTest {
EdgeEvent edgeEvent = constructEdgeEvent(tenantId, edge.getId(), EdgeEventActionType.ATTRIBUTES_DELETED, device.getId().getId(), EdgeEventType.DEVICE, deleteAttributesEntityData);
edgeImitator.expectMessageAmount(1);
edgeEventService.saveAsync(edgeEvent);
clusterService.onEdgeEventUpdate(tenantId, edge.getId());
edgeImitator.waitForMessages();
AbstractMessage latestMessage = edgeImitator.getLatestMessage();
@ -908,6 +915,7 @@ abstract public class BaseEdgeTest extends AbstractControllerTest {
EdgeEvent edgeEvent = constructEdgeEvent(tenantId, edge.getId(), EdgeEventActionType.POST_ATTRIBUTES, device.getId().getId(), EdgeEventType.DEVICE, postAttributesEntityData);
edgeImitator.expectMessageAmount(1);
edgeEventService.saveAsync(edgeEvent);
clusterService.onEdgeEventUpdate(tenantId, edge.getId());
edgeImitator.waitForMessages();
AbstractMessage latestMessage = edgeImitator.getLatestMessage();
@ -932,6 +940,7 @@ abstract public class BaseEdgeTest extends AbstractControllerTest {
EdgeEvent edgeEvent1 = constructEdgeEvent(tenantId, edge.getId(), EdgeEventActionType.ATTRIBUTES_UPDATED, device.getId().getId(), EdgeEventType.DEVICE, attributesEntityData);
edgeImitator.expectMessageAmount(1);
edgeEventService.saveAsync(edgeEvent1);
clusterService.onEdgeEventUpdate(tenantId, edge.getId());
edgeImitator.waitForMessages();
AbstractMessage latestMessage = edgeImitator.getLatestMessage();
@ -1158,6 +1167,7 @@ abstract public class BaseEdgeTest extends AbstractControllerTest {
edgeImitator.sendUplinkMsg(uplinkMsgBuilder2.build());
edgeImitator.waitForResponses();
// Wait before device attributes saved to database before requesting them from controller
Thread.sleep(1000);
Map<String, List<Map<String, String>>> timeseries = doGetAsync("/api/plugins/telemetry/DEVICE/" + device.getUuidId() + "/values/timeseries?keys=" + timeseriesKey, Map.class);
Assert.assertTrue(timeseries.containsKey(timeseriesKey));
@ -1300,18 +1310,25 @@ abstract public class BaseEdgeTest extends AbstractControllerTest {
private void sendAttributesRequest() throws Exception {
Device device = findDeviceByName("Edge Device 1");
sendAttributesRequest(device, DataConstants.SERVER_SCOPE, "{\"key1\":\"value1\"}", "key1", "value1");
sendAttributesRequest(device, DataConstants.SHARED_SCOPE, "{\"key2\":\"value2\"}", "key2", "value2");
}
String attributesDataStr = "{\"key1\":\"value1\"}";
private void sendAttributesRequest(Device device, String scope, String attributesDataStr, String expectedKey, String expectedValue) throws Exception {
JsonNode attributesData = mapper.readTree(attributesDataStr);
doPost("/api/plugins/telemetry/DEVICE/" + device.getId().getId().toString() + "/attributes/" + DataConstants.SERVER_SCOPE,
doPost("/api/plugins/telemetry/DEVICE/" + device.getId().getId().toString() + "/attributes/" + scope,
attributesData);
// Wait before device attributes saved to database before requesting them from edge
Thread.sleep(1000);
UplinkMsg.Builder uplinkMsgBuilder = UplinkMsg.newBuilder();
AttributesRequestMsg.Builder attributesRequestMsgBuilder = AttributesRequestMsg.newBuilder();
attributesRequestMsgBuilder.setEntityIdMSB(device.getUuidId().getMostSignificantBits());
attributesRequestMsgBuilder.setEntityIdLSB(device.getUuidId().getLeastSignificantBits());
attributesRequestMsgBuilder.setEntityType(EntityType.DEVICE.name());
attributesRequestMsgBuilder.setScope(scope);
testAutoGeneratedCodeByProtobuf(attributesRequestMsgBuilder);
uplinkMsgBuilder.addAttributesRequestMsg(attributesRequestMsgBuilder.build());
testAutoGeneratedCodeByProtobuf(uplinkMsgBuilder);
@ -1328,14 +1345,14 @@ abstract public class BaseEdgeTest extends AbstractControllerTest {
Assert.assertEquals(device.getUuidId().getMostSignificantBits(), latestEntityDataMsg.getEntityIdMSB());
Assert.assertEquals(device.getUuidId().getLeastSignificantBits(), latestEntityDataMsg.getEntityIdLSB());
Assert.assertEquals(device.getId().getEntityType().name(), latestEntityDataMsg.getEntityType());
Assert.assertEquals("SERVER_SCOPE", latestEntityDataMsg.getPostAttributeScope());
Assert.assertEquals(scope, latestEntityDataMsg.getPostAttributeScope());
Assert.assertTrue(latestEntityDataMsg.hasAttributesUpdatedMsg());
TransportProtos.PostAttributeMsg attributesUpdatedMsg = latestEntityDataMsg.getAttributesUpdatedMsg();
Assert.assertEquals(1, attributesUpdatedMsg.getKvCount());
TransportProtos.KeyValueProto keyValueProto = attributesUpdatedMsg.getKv(0);
Assert.assertEquals("key1", keyValueProto.getKey());
Assert.assertEquals("value1", keyValueProto.getStringV());
Assert.assertEquals(expectedKey, keyValueProto.getKey());
Assert.assertEquals(expectedValue, keyValueProto.getStringV());
}
private void sendDeleteDeviceOnEdge() throws Exception {

22
application/src/test/java/org/thingsboard/server/edge/imitator/EdgeImitator.java
View File

@ -56,6 +56,8 @@ import java.util.Optional;
import java.util.UUID;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
@Slf4j
public class EdgeImitator {
@ -65,6 +67,8 @@ public class EdgeImitator {
private EdgeRpcClient edgeRpcClient;
private final Lock lock = new ReentrantLock();
private CountDownLatch messagesLatch;
private CountDownLatch responsesLatch;
private List<Class<? extends AbstractMessage>> ignoredTypes;
@ -74,7 +78,7 @@ public class EdgeImitator {
@Getter
private UserId userId;
@Getter
private List<AbstractMessage> downlinkMsgs;
private final List<AbstractMessage> downlinkMsgs;
public EdgeImitator(String host, int port, String routingKey, String routingSecret) throws NoSuchFieldException, IllegalAccessException {
edgeRpcClient = new EdgeGrpcClient();
@ -241,7 +245,12 @@ public class EdgeImitator {
private ListenableFuture<Void> saveDownlinkMsg(AbstractMessage message) {
if (!ignoredTypes.contains(message.getClass())) {
downlinkMsgs.add(message);
try {
lock.lock();
downlinkMsgs.add(message);
} finally {
lock.unlock();
}
messagesLatch.countDown();
}
return Futures.immediateFuture(null);
@ -262,7 +271,14 @@ public class EdgeImitator {
}
public <T> Optional<T> findMessageByType(Class<T> tClass) {
return (Optional<T>) downlinkMsgs.stream().filter(downlinkMsg -> downlinkMsg.getClass().isAssignableFrom(tClass)).findAny();
Optional<T> result;
try {
lock.lock();
result = (Optional<T>) downlinkMsgs.stream().filter(downlinkMsg -> downlinkMsg.getClass().isAssignableFrom(tClass)).findAny();
} finally {
lock.unlock();
}
return result;
}
public AbstractMessage getLatestMessage() {

3
common/data/src/main/java/org/thingsboard/server/common/data/EdgeUtils.java
View File

@ -15,8 +15,10 @@
*/
package org.thingsboard.server.common.data;
import lombok.extern.slf4j.Slf4j;
import org.thingsboard.server.common.data.edge.EdgeEventType;
@Slf4j
public final class EdgeUtils {
private EdgeUtils() {
@ -49,6 +51,7 @@ public final class EdgeUtils {
case WIDGET_TYPE:
return EdgeEventType.WIDGET_TYPE;
default:
log.warn("Unsupported entity type [{}]", entityType);
return null;
}
}

17
common/edge-api/src/main/proto/edge.proto
View File

@ -79,12 +79,16 @@ message EdgeConfiguration {
int64 edgeIdLSB = 2;
int64 tenantIdMSB = 3;
int64 tenantIdLSB = 4;
string name = 5;
string routingKey = 6;
string type = 7;
string edgeLicenseKey = 8;
string cloudEndpoint = 9;
string cloudType = 10;
int64 customerIdMSB = 5;
int64 customerIdLSB = 6;
string name = 7;
string type = 8;
string routingKey = 9;
string secret = 10;
string edgeLicenseKey = 11;
string cloudEndpoint = 12;
string configuration = 13;
string cloudType = 14;
}
enum UpdateMsgType {
@ -315,6 +319,7 @@ message AttributesRequestMsg {
int64 entityIdMSB = 1;
int64 entityIdLSB = 2;
string entityType = 3;
string scope = 4;
}
message RelationRequestMsg {

7
common/message/src/main/java/org/thingsboard/server/common/msg/MsgType.java
View File

@ -105,6 +105,11 @@ public enum MsgType {
/**
* Message that is sent by TransportRuleEngineService to Device Actor. Represents messages from the device itself.
*/
TRANSPORT_TO_DEVICE_ACTOR_MSG;
TRANSPORT_TO_DEVICE_ACTOR_MSG,
/**
* Message that is sent on Edge Event to Edge Session
*/
EDGE_EVENT_UPDATE_TO_EDGE_SESSION_MSG;
}

42
common/message/src/main/java/org/thingsboard/server/common/msg/edge/EdgeEventUpdateMsg.java
View File

@ -0,0 +1,42 @@
/**
* Copyright © 2016-2020 The Thingsboard Authors
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.thingsboard.server.common.msg.edge;
import lombok.Getter;
import lombok.ToString;
import org.thingsboard.server.common.data.id.EdgeId;
import org.thingsboard.server.common.data.id.TenantId;
import org.thingsboard.server.common.msg.MsgType;
import org.thingsboard.server.common.msg.aware.TenantAwareMsg;
import org.thingsboard.server.common.msg.cluster.ToAllNodesMsg;
@ToString
public class EdgeEventUpdateMsg implements TenantAwareMsg, ToAllNodesMsg {
@Getter
private final TenantId tenantId;
@Getter
private final EdgeId edgeId;
public EdgeEventUpdateMsg(TenantId tenantId, EdgeId edgeId) {
this.tenantId = tenantId;
this.edgeId = edgeId;
}
@Override
public MsgType getMsgType() {
return MsgType.EDGE_EVENT_UPDATE_TO_EDGE_SESSION_MSG;
}
}

1
common/queue/src/main/proto/queue.proto
View File

@ -520,6 +520,7 @@ message ToCoreNotificationMsg {
LocalSubscriptionServiceMsgProto toLocalSubscriptionServiceMsg = 1;
FromDeviceRPCResponseProto fromDeviceRpcResponse = 2;
bytes componentLifecycleMsg = 3;
bytes edgeEventUpdateMsg = 4;
}
/* Messages that are handled by ThingsBoard RuleEngine Service */

67
dao/src/main/java/org/thingsboard/server/dao/edge/EdgeServiceImpl.java
View File

@ -16,6 +16,7 @@
package org.thingsboard.server.dao.edge;
import com.google.common.base.Function;
import com.google.common.util.concurrent.FutureCallback;
import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.ListenableFuture;
import com.google.common.util.concurrent.MoreExecutors;
@ -98,6 +99,8 @@ public class EdgeServiceImpl extends AbstractEntityService implements EdgeServic
public static final String INCORRECT_CUSTOMER_ID = "Incorrect customerId ";
public static final String INCORRECT_EDGE_ID = "Incorrect edgeId ";
private static final int DEFAULT_LIMIT = 100;
private RestTemplate restTemplate;
private static final String EDGE_LICENSE_SERVER_ENDPOINT = "https://license.thingsboard.io";
@ -353,13 +356,20 @@ public class EdgeServiceImpl extends AbstractEntityService implements EdgeServic
public void assignDefaultRuleChainsToEdge(TenantId tenantId, EdgeId edgeId) {
log.trace("Executing assignDefaultRuleChainsToEdge, tenantId [{}], edgeId [{}]", tenantId, edgeId);
ListenableFuture<List<RuleChain>> future = ruleChainService.findDefaultEdgeRuleChainsByTenantId(tenantId);
Futures.transform(future, ruleChains -> {
if (ruleChains != null && !ruleChains.isEmpty()) {
for (RuleChain ruleChain : ruleChains) {
ruleChainService.assignRuleChainToEdge(tenantId, ruleChain.getId(), edgeId);
Futures.addCallback(future, new FutureCallback<List<RuleChain>>() {
@Override
public void onSuccess(List<RuleChain> ruleChains) {
if (ruleChains != null && !ruleChains.isEmpty()) {
for (RuleChain ruleChain : ruleChains) {
ruleChainService.assignRuleChainToEdge(tenantId, ruleChain.getId(), edgeId);
}
}
}
return null;
@Override
public void onFailure(Throwable t) {
log.warn("[{}] can't find default edge rule chains [{}]", tenantId.getId(), edgeId.getId(), t);
}
}, MoreExecutors.directExecutor());
}
@ -462,9 +472,26 @@ public class EdgeServiceImpl extends AbstractEntityService implements EdgeServic
@Override
public ListenableFuture<List<EdgeId>> findRelatedEdgeIdsByEntityId(TenantId tenantId, EntityId entityId) {
log.trace("[{}] Executing findRelatedEdgeIdsByEntityId [{}]", tenantId, entityId);
if (EntityType.TENANT.equals(entityId.getEntityType())) {
PageData<Edge> edgesByTenantId = findEdgesByTenantId(tenantId, new PageLink(Integer.MAX_VALUE));
return Futures.immediateFuture(edgesByTenantId.getData().stream().map(IdBased::getId).collect(Collectors.toList()));
if (EntityType.TENANT.equals(entityId.getEntityType()) || EntityType.CUSTOMER.equals(entityId.getEntityType())) {
List<EdgeId> result = new ArrayList<>();
PageLink pageLink = new PageLink(DEFAULT_LIMIT);
PageData<Edge> pageData;
do {
if (EntityType.TENANT.equals(entityId.getEntityType())) {
pageData = findEdgesByTenantId(tenantId, pageLink);
} else {
pageData = findEdgesByTenantIdAndCustomerId(tenantId, new CustomerId(entityId.getId()), pageLink);
}
if (pageData != null && pageData.getData() != null && !pageData.getData().isEmpty()) {
for (Edge edge : pageData.getData()) {
result.add(edge.getId());
}
if (pageData.hasNext()) {
pageLink = pageLink.nextPageLink();
}
}
} while (pageData != null && pageData.hasNext());
return Futures.immediateFuture(result);
} else {
switch (entityId.getEntityType()) {
case DEVICE:
@ -489,13 +516,23 @@ public class EdgeServiceImpl extends AbstractEntityService implements EdgeServic
if (userById == null) {
return Futures.immediateFuture(Collections.emptyList());
}
PageData<Edge> edges;
if (userById.getCustomerId() == null || userById.getCustomerId().isNullUid()) {
edges = findEdgesByTenantId(tenantId, new PageLink(Integer.MAX_VALUE));
} else {
edges = findEdgesByTenantIdAndCustomerId(tenantId, new CustomerId(entityId.getId()), new PageLink(Integer.MAX_VALUE));
}
return convertToEdgeIds(Futures.immediateFuture(edges.getData()));
List<Edge> result = new ArrayList<>();
PageLink pageLink = new PageLink(DEFAULT_LIMIT);
PageData<Edge> pageData;
do {
if (userById.getCustomerId() == null || userById.getCustomerId().isNullUid()) {
pageData = findEdgesByTenantId(tenantId, pageLink);
} else {
pageData = findEdgesByTenantIdAndCustomerId(tenantId, new CustomerId(entityId.getId()), pageLink);
}
if (pageData != null && pageData.getData() != null && !pageData.getData().isEmpty()) {
result.addAll(pageData.getData());
if (pageData.hasNext()) {
pageLink = pageLink.nextPageLink();
}
}
} while (pageData != null && pageData.hasNext());
return convertToEdgeIds(Futures.immediateFuture(result));
default:
return Futures.immediateFuture(Collections.emptyList());
}

3
dao/src/main/java/org/thingsboard/server/dao/entity/AbstractEntityService.java
View File

@ -77,6 +77,7 @@ public abstract class AbstractEntityService {
List<EntityView> entityViews = entityViewService.findEntityViewsByTenantIdAndEntityIdAsync(tenantId, entityId).get();
if (entityViews != null && !entityViews.isEmpty()) {
EntityView entityView = entityViews.get(0);
// TODO: voba - refactor this blocking operation in 3.3+
Boolean relationExists = relationService.checkRelation(tenantId,edgeId, entityView.getId(),
EntityRelation.CONTAINS_TYPE, RelationTypeGroup.EDGE).get();
if (relationExists) {
@ -84,7 +85,7 @@ public abstract class AbstractEntityService {
}
}
} catch (Exception e) {
log.error("Exception while finding entity views for entityId [{}]", entityId, e);
log.error("[{}] Exception while finding entity views for entityId [{}]", tenantId, entityId, e);
throw new RuntimeException("Exception while finding entity views for entityId [" + entityId + "]", e);
}
}

16
dao/src/main/java/org/thingsboard/server/dao/event/BaseEventService.java
View File

@ -59,7 +59,7 @@ public class BaseEventService implements EventService {
public Optional<Event> saveIfNotExists(Event event) {
eventValidator.validate(event, Event::getTenantId);
if (StringUtils.isEmpty(event.getUid())) {
throw new DataValidationException("Event uid should be specified!");
throw new DataValidationException("Event uid should be specified!.");
}
checkAndTruncateDebugEvent(event);
return eventDao.saveIfNotExists(event);
@ -79,16 +79,16 @@ public class BaseEventService implements EventService {
@Override
public Optional<Event> findEvent(TenantId tenantId, EntityId entityId, String eventType, String eventUid) {
if (tenantId == null) {
throw new DataValidationException("Tenant id should be specified!");
throw new DataValidationException("Tenant id should be specified!.");
}
if (entityId == null) {
throw new DataValidationException("Entity id should be specified!");
throw new DataValidationException("Entity id should be specified!.");
}
if (StringUtils.isEmpty(eventType)) {
throw new DataValidationException("Event type should be specified!");
throw new DataValidationException("Event type should be specified!.");
}
if (StringUtils.isEmpty(eventUid)) {
throw new DataValidationException("Event uid should be specified!");
throw new DataValidationException("Event uid should be specified!.");
}
Event event = eventDao.findEvent(tenantId.getId(), entityId, eventType, eventUid);
return event != null ? Optional.of(event) : Optional.empty();
@ -129,13 +129,13 @@ public class BaseEventService implements EventService {
@Override
protected void validateDataImpl(TenantId tenantId, Event event) {
if (event.getEntityId() == null) {
throw new DataValidationException("Entity id should be specified!");
throw new DataValidationException("Entity id should be specified!.");
}
if (StringUtils.isEmpty(event.getType())) {
throw new DataValidationException("Event type should be specified!");
throw new DataValidationException("Event type should be specified!.");
}
if (event.getBody() == null) {
throw new DataValidationException("Event body should be specified!");
throw new DataValidationException("Event body should be specified!.");
}
}
};

3
rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/TbContext.java
View File

@ -27,6 +27,7 @@ import org.thingsboard.server.common.data.TenantProfile;
import org.thingsboard.server.common.data.alarm.Alarm;
import org.thingsboard.server.common.data.asset.Asset;
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.RuleNodeId;
import org.thingsboard.server.common.data.id.TenantId;
@ -153,6 +154,8 @@ public interface TbContext {
// TODO: Does this changes the message?
TbMsg alarmActionMsg(Alarm alarm, RuleNodeId ruleNodeId, String action);
void onEdgeEventUpdate(TenantId tenantId, EdgeId edgeId);
/*
*
* METHODS TO PROCESS THE MESSAGES

101
rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/edge/TbMsgPushToEdgeNode.java
View File

@ -57,7 +57,7 @@ import static org.thingsboard.rule.engine.api.TbRelationTypes.SUCCESS;
name = "push to edge",
configClazz = EmptyNodeConfiguration.class,
nodeDescription = "Pushes messages to edge",
nodeDetails = "Pushes messages to edge, if Message Originator assigned to particular edge or is EDGE entity. This node is used only on Cloud instances to push messages from Cloud to Edge. Supports only DEVICE, ENTITY_VIEW, ASSET and EDGE Message Originator(s).",
nodeDetails = "Pushes messages to edge, if Message Originator assigned to particular edge or is EDGE entity. This node is used only on Cloud instances to push messages from Cloud to Edge. Supports only DEVICE, ENTITY_VIEW, ASSET, ENTITY_VIEW, DASHBOARD, TENANT, CUSTOMER and EDGE Message Originator(s).",
uiResources = {"static/rulenode/rulenode-core-config.js", "static/rulenode/rulenode-core-config.css"},
configDirective = "tbNodeEmptyConfig",
icon = "cloud_download",
@ -97,47 +97,75 @@ public class TbMsgPushToEdgeNode implements TbNode {
}
private void processMsg(TbContext ctx, TbMsg msg) {
ListenableFuture<List<EdgeId>> getEdgeIdsFuture = ctx.getEdgeService().findRelatedEdgeIdsByEntityId(ctx.getTenantId(), msg.getOriginator());
Futures.addCallback(getEdgeIdsFuture, new FutureCallback<List<EdgeId>>() {
@Override
public void onSuccess(@Nullable List<EdgeId> edgeIds) {
if (edgeIds != null && !edgeIds.isEmpty()) {
for (EdgeId edgeId : edgeIds) {
try {
EdgeEvent edgeEvent = buildEdgeEvent(msg, ctx);
if (edgeEvent == null) {
log.debug("Edge event type is null. Entity Type {}", msg.getOriginator().getEntityType());
ctx.tellFailure(msg, new RuntimeException("Edge event type is null. Entity Type '" + msg.getOriginator().getEntityType() + "'"));
} else {
edgeEvent.setEdgeId(edgeId);
ListenableFuture<EdgeEvent> saveFuture = ctx.getEdgeEventService().saveAsync(edgeEvent);
Futures.addCallback(saveFuture, new FutureCallback<EdgeEvent>() {
@Override
public void onSuccess(@Nullable EdgeEvent event) {
ctx.tellNext(msg, SUCCESS);
}
@Override
public void onFailure(Throwable th) {
log.error("Could not save edge event", th);
ctx.tellFailure(msg, th);
}
}, ctx.getDbCallbackExecutor());
if (EntityType.EDGE.equals(msg.getOriginator().getEntityType())) {
try {
EdgeEvent edgeEvent = buildEdgeEvent(msg, ctx);
if (edgeEvent != null) {
EdgeId edgeId = new EdgeId(msg.getOriginator().getId());
edgeEvent.setEdgeId(edgeId);
ListenableFuture<EdgeEvent> saveFuture = ctx.getEdgeEventService().saveAsync(edgeEvent);
Futures.addCallback(saveFuture, new FutureCallback<EdgeEvent>() {
@Override
public void onSuccess(@Nullable EdgeEvent event) {
ctx.tellNext(msg, SUCCESS);
ctx.onEdgeEventUpdate(ctx.getTenantId(), edgeId);
}
@Override
public void onFailure(Throwable th) {
log.warn("[{}] Can't save edge event [{}] for edge [{}]", ctx.getTenantId().getId(), edgeEvent, edgeId.getId(), th);
ctx.tellFailure(msg, th);
}
}, ctx.getDbCallbackExecutor());
}
} catch (JsonProcessingException e) {
log.error("Failed to build edge event", e);
ctx.tellFailure(msg, e);
}
} else {
ListenableFuture<List<EdgeId>> getEdgeIdsFuture = ctx.getEdgeService().findRelatedEdgeIdsByEntityId(ctx.getTenantId(), msg.getOriginator());
Futures.addCallback(getEdgeIdsFuture, new FutureCallback<List<EdgeId>>() {
@Override
public void onSuccess(@Nullable List<EdgeId> edgeIds) {
if (edgeIds != null && !edgeIds.isEmpty()) {
for (EdgeId edgeId : edgeIds) {
try {
EdgeEvent edgeEvent = buildEdgeEvent(msg, ctx);
if (edgeEvent == null) {
log.debug("Edge event type is null. Entity Type {}", msg.getOriginator().getEntityType());
ctx.tellFailure(msg, new RuntimeException("Edge event type is null. Entity Type '" + msg.getOriginator().getEntityType() + "'"));
} else {
edgeEvent.setEdgeId(edgeId);
ListenableFuture<EdgeEvent> saveFuture = ctx.getEdgeEventService().saveAsync(edgeEvent);
Futures.addCallback(saveFuture, new FutureCallback<EdgeEvent>() {
@Override
public void onSuccess(@Nullable EdgeEvent event) {
ctx.tellNext(msg, SUCCESS);
ctx.onEdgeEventUpdate(ctx.getTenantId(), edgeId);
}
@Override
public void onFailure(Throwable th) {
log.warn("[{}] Can't save edge event [{}] for edge [{}]", ctx.getTenantId().getId(), edgeEvent, edgeId.getId(), th);
ctx.tellFailure(msg, th);
}
}, ctx.getDbCallbackExecutor());
}
} catch (JsonProcessingException e) {
log.error("Failed to build edge event", e);
ctx.tellFailure(msg, e);
}
} catch (JsonProcessingException e) {
log.error("Failed to build edge event", e);
ctx.tellFailure(msg, e);
}
}
}
}
@Override
public void onFailure(Throwable t) {
ctx.tellFailure(msg, t);
}
@Override
public void onFailure(Throwable t) {
ctx.tellFailure(msg, t);
}
}, ctx.getDbCallbackExecutor());
}, ctx.getDbCallbackExecutor());
}
}
private EdgeEvent buildEdgeEvent(TbMsg msg, TbContext ctx) throws JsonProcessingException {
@ -220,6 +248,7 @@ public class TbMsgPushToEdgeNode implements TbNode {
case DASHBOARD:
case TENANT:
case CUSTOMER:
case EDGE:
return true;
default:
return false;

14
ui-ngx/src/assets/locale/locale.constant-en_US.json
View File

@ -307,10 +307,11 @@
"filter-type-device-type-description": "Devices of type '{{deviceType}}'",
"filter-type-device-type-and-name-description": "Devices of type '{{deviceType}}' and with name starting with '{{prefix}}'",
"filter-type-entity-view-type": "Entity View type",
"filter-type-entity-view-type-description": "Entity Views of type '{{entityView}}'",
"filter-type-entity-view-type-and-name-description": "Entity Views of type '{{entityView}}' and with name starting with '{{prefix}}'",
"filter-type-entity-view-type-description": "Entity Views of type '{{entityViewType}}'",
"filter-type-entity-view-type-and-name-description": "Entity Views of type '{{entityViewType}}' and with name starting with '{{prefix}}'",
"filter-type-edge-type": "Edge type",
"filter-type-edge-type-description": "Edges of type '{{edgeType}}'",
"filter-type-edge-type-and-name-description": "Edges of type '{{edgeType}}' and with name starting with '{{prefix}}'",
"filter-type-relations-query": "Relations query",
"filter-type-relations-query-description": "{{entities}} that have {{relationType}} relation {{direction}} {{rootEntity}}",
"filter-type-asset-search-query": "Asset search query",
@ -1131,6 +1132,7 @@
"delete-edges-action-title": "Delete { count, plural, 1 {1 edge} other {# edges} }",
"delete-edges-text": "Be careful, after the confirmation all selected edges will be removed and all related data will become unrecoverable.",
"name": "Name",
"name-starts-with": "Edge name starts with",
"name-required": "Name is required.",
"edge-license-key": "Edge License Key",
"edge-license-key-required": "Edge License Key is required.",
@ -1201,7 +1203,8 @@
"selected-edges": "{ count, plural, 1 {1 edge} other {# edges} } selected",
"enter-edge-type": "Enter entity view type",
"any-edge": "Any edge",
"no-edge-types-matching": "No edge types matching '{{entitySubtype}}' were found."
"no-edge-types-matching": "No edge types matching '{{entitySubtype}}' were found.",
"unassign-edges-action-title": "Unassign { count, plural, 1 {1 edge} other {# edges} } from customer"
},
"error": {
"unable-to-connect": "Unable to connect to the server! Please check your internet connection.",
@ -1993,6 +1996,7 @@
"rulechain": {
"rulechain": "Rule chain",
"rulechains": "Rule chains",
"default-root": "Default root",
"root": "Root",
"delete": "Delete rule chain",
"name": "Name",
@ -2041,10 +2045,10 @@
"set-default-root-edge-rulechain-title": "Are you sure you want to make the rule chain '{{ruleChainName}}' default edge root?",
"set-default-root-edge-rulechain-text": "After the confirmation the rule chain will become default edge root and will handle all incoming transport messages.",
"invalid-rulechain-type-error": "Unable to import rule chain: Invalid rule chain type. Expected type is {{expectedRuleChainType}}.",
"set-default-edge": "Make edge rule chain default",
"set-default-edge": "Make rule chain default",
"set-default-edge-title": "Are you sure you want to make the edge rule chain '{{ruleChainName}}' default?",
"set-default-edge-text": "After the confirmation the edge rule chain will be added to default list and assigned to newly created edge(s).",
"remove-default-edge": "Remove edge rule chain from defaults",
"remove-default-edge": "Remove rule chain from defaults",
"remove-default-edge-title": "Are you sure you want to remove the edge rule chain '{{ruleChainName}}' from default list?",
"remove-default-edge-text": "After the confirmation the edge rule chain will not be assigned for a newly created edges.",
"unassign-rulechain-title": "Are you sure you want to unassign the rulechain '{{ruleChainName}}'?",

2
ui/src/app/edge/edge.controller.js
View File

@ -283,7 +283,7 @@ export function EdgeController($rootScope, userService, edgeService, customerSer
details: function() {
return $translate.instant('edge.manage-edge-rulechains');
},
icon: "settings_ethernet"
icon: "code"
}
);

2
ui/src/app/edge/edge.routes.js
View File

@ -49,7 +49,7 @@ export default function EdgeRoutes($stateProvider, types) {
pageTitle: 'edge.edges'
},
ncyBreadcrumb: {
label: '{"icon": "transform", "label": "edge.edges"}'
label: '{"icon": "router", "label": "edge.edges"}'
}
}).state('home.edges.entityViews', {
url: '/:edgeId/entityViews',

Write Preview
Loading…
Cancel
Save