简柏林 / thingskit · Commits

Commit 600c9ec565fe803e7405b3f6ea9a1c2e6896c23c

Authored by Volodymyr Babak
1 parent e8afd976

Refactoring of notification to edge. Added EdgeNotificationService and queue msg proto

Inline Side-by-side
Showing 9 changed files with 490 additions and 393 deletions
  1 +package org.thingsboard.server.service.edge;
  2 +
  3 +import com.fasterxml.jackson.databind.ObjectMapper;
  4 +import com.google.common.util.concurrent.FutureCallback;
  5 +import com.google.common.util.concurrent.Futures;
  6 +import com.google.common.util.concurrent.ListenableFuture;
  7 +import com.google.common.util.concurrent.MoreExecutors;
  8 +import lombok.extern.slf4j.Slf4j;
  9 +import org.apache.commons.codec.binary.Base64;
  10 +import org.springframework.beans.factory.annotation.Autowired;
  11 +import org.springframework.stereotype.Service;
  12 +import org.thingsboard.server.common.data.Dashboard;
  13 +import org.thingsboard.server.common.data.DataConstants;
  14 +import org.thingsboard.server.common.data.Device;
  15 +import org.thingsboard.server.common.data.EntityType;
  16 +import org.thingsboard.server.common.data.EntityView;
  17 +import org.thingsboard.server.common.data.Event;
  18 +import org.thingsboard.server.common.data.alarm.Alarm;
  19 +import org.thingsboard.server.common.data.asset.Asset;
  20 +import org.thingsboard.server.common.data.edge.Edge;
  21 +import org.thingsboard.server.common.data.edge.EdgeQueueEntityType;
  22 +import org.thingsboard.server.common.data.edge.EdgeQueueEntry;
  23 +import org.thingsboard.server.common.data.id.EdgeId;
  24 +import org.thingsboard.server.common.data.id.EntityId;
  25 +import org.thingsboard.server.common.data.id.RuleChainId;
  26 +import org.thingsboard.server.common.data.id.TenantId;
  27 +import org.thingsboard.server.common.data.page.TimePageData;
  28 +import org.thingsboard.server.common.data.page.TimePageLink;
  29 +import org.thingsboard.server.common.data.relation.EntityRelation;
  30 +import org.thingsboard.server.common.data.relation.RelationTypeGroup;
  31 +import org.thingsboard.server.common.data.rule.RuleChain;
  32 +import org.thingsboard.server.common.data.rule.RuleChainMetaData;
  33 +import org.thingsboard.server.common.data.rule.RuleChainType;
  34 +import org.thingsboard.server.common.msg.TbMsg;
  35 +import org.thingsboard.server.common.msg.queue.TbCallback;
  36 +import org.thingsboard.server.common.msg.session.SessionMsgType;
  37 +import org.thingsboard.server.dao.asset.AssetService;
  38 +import org.thingsboard.server.dao.device.DeviceService;
  39 +import org.thingsboard.server.dao.edge.EdgeService;
  40 +import org.thingsboard.server.dao.entityview.EntityViewService;
  41 +import org.thingsboard.server.dao.event.EventService;
  42 +import org.thingsboard.server.dao.relation.RelationService;
  43 +import org.thingsboard.server.dao.rule.RuleChainService;
  44 +import org.thingsboard.server.gen.transport.TransportProtos;
  45 +import org.thingsboard.server.queue.util.TbCoreComponent;
  46 +
  47 +import javax.annotation.Nullable;
  48 +import javax.annotation.PostConstruct;
  49 +import javax.annotation.PreDestroy;
  50 +import java.io.IOException;
  51 +import java.util.List;
  52 +import java.util.UUID;
  53 +import java.util.concurrent.ExecutorService;
  54 +import java.util.concurrent.Executors;
  55 +
  56 +@Service
  57 +@TbCoreComponent
  58 +@Slf4j
  59 +public class DefaultEdgeNotificationService implements EdgeNotificationService {
  60 +
  61 + private static final ObjectMapper mapper = new ObjectMapper();
  62 +
  63 + @Autowired
  64 + private EdgeService edgeService;
  65 +
  66 + @Autowired
  67 + private DeviceService deviceService;
  68 +
  69 + @Autowired
  70 + private AssetService assetService;
  71 +
  72 + @Autowired
  73 + private EntityViewService entityViewService;
  74 +
  75 + @Autowired
  76 + private RuleChainService ruleChainService;
  77 +
  78 + @Autowired
  79 + private RelationService relationService;
  80 +
  81 + @Autowired
  82 + private EventService eventService;
  83 +
  84 + private ExecutorService tsCallBackExecutor;
  85 +
  86 + @PostConstruct
  87 + public void initExecutor() {
  88 + tsCallBackExecutor = Executors.newSingleThreadExecutor();
  89 + }
  90 +
  91 + @PreDestroy
  92 + public void shutdownExecutor() {
  93 + if (tsCallBackExecutor != null) {
  94 + tsCallBackExecutor.shutdownNow();
  95 + }
  96 + }
  97 +
  98 + @Override
  99 + public TimePageData<Event> findQueueEvents(TenantId tenantId, EdgeId edgeId, TimePageLink pageLink) {
  100 + return eventService.findEvents(tenantId, edgeId, DataConstants.EDGE_QUEUE_EVENT_TYPE, pageLink);
  101 + }
  102 +
  103 + @Override
  104 + public Edge setEdgeRootRuleChain(TenantId tenantId, Edge edge, RuleChainId ruleChainId) throws IOException {
  105 + edge.setRootRuleChainId(ruleChainId);
  106 + Edge savedEdge = edgeService.saveEdge(edge);
  107 + RuleChain ruleChain = ruleChainService.findRuleChainById(tenantId, ruleChainId);
  108 + saveEventToEdgeQueue(tenantId, edge.getId(), EdgeQueueEntityType.RULE_CHAIN, DataConstants.ENTITY_UPDATED, mapper.writeValueAsString(ruleChain), new FutureCallback<Void>() {
  109 + @Override
  110 + public void onSuccess(@Nullable Void aVoid) {
  111 + log.debug("Event saved successfully!");
  112 + }
  113 +
  114 + @Override
  115 + public void onFailure(Throwable t) {
  116 + log.debug("Failure during event save", t);
  117 + }
  118 + });
  119 + return savedEdge;
  120 + }
  121 +
  122 + private void saveEventToEdgeQueue(TenantId tenantId, EdgeId edgeId, EdgeQueueEntityType entityType, String type, String data, FutureCallback<Void> callback) throws IOException {
  123 + log.debug("Pushing single event to edge queue. tenantId [{}], edgeId [{}], entityType [{}], type[{}], data [{}]", tenantId, edgeId, entityType, type, data);
  124 +
  125 + EdgeQueueEntry queueEntry = new EdgeQueueEntry();
  126 + queueEntry.setEntityType(entityType);
  127 + queueEntry.setType(type);
  128 + queueEntry.setData(data);
  129 +
  130 + Event event = new Event();
  131 + event.setEntityId(edgeId);
  132 + event.setTenantId(tenantId);
  133 + event.setType(DataConstants.EDGE_QUEUE_EVENT_TYPE);
  134 + event.setBody(mapper.valueToTree(queueEntry));
  135 + ListenableFuture<Event> saveFuture = eventService.saveAsync(event);
  136 +
  137 + addMainCallback(saveFuture, callback);
  138 + }
  139 +
  140 + private void addMainCallback(ListenableFuture<Event> saveFuture, final FutureCallback<Void> callback) {
  141 + Futures.addCallback(saveFuture, new FutureCallback<Event>() {
  142 + @Override
  143 + public void onSuccess(@Nullable Event result) {
  144 + callback.onSuccess(null);
  145 + }
  146 +
  147 + @Override
  148 + public void onFailure(Throwable t) {
  149 + callback.onFailure(t);
  150 + }
  151 + }, tsCallBackExecutor);
  152 + }
  153 +
  154 + @Override
  155 + public void pushNotificationToEdge(TransportProtos.EdgeNotificationMsgProto edgeNotificationMsg, TbCallback callback) {
  156 + if (tbMsg.getType().equals(SessionMsgType.POST_TELEMETRY_REQUEST.name()) ||
  157 + tbMsg.getType().equals(SessionMsgType.POST_ATTRIBUTES_REQUEST.name()) ||
  158 + tbMsg.getType().equals(DataConstants.ATTRIBUTES_UPDATED) ||
  159 + tbMsg.getType().equals(DataConstants.ATTRIBUTES_DELETED)) {
  160 + processCustomTbMsg(tenantId, tbMsg, callback);
  161 + } else {
  162 + try {
  163 + switch (tbMsg.getOriginator().getEntityType()) {
  164 + case EDGE:
  165 + processEdge(tenantId, tbMsg, callback);
  166 + break;
  167 + case ASSET:
  168 + processAsset(tenantId, tbMsg, callback);
  169 + break;
  170 + case DEVICE:
  171 + processDevice(tenantId, tbMsg, callback);
  172 + break;
  173 + case DASHBOARD:
  174 + processDashboard(tenantId, tbMsg, callback);
  175 + break;
  176 + case RULE_CHAIN:
  177 + processRuleChain(tenantId, tbMsg, callback);
  178 + break;
  179 + case ENTITY_VIEW:
  180 + processEntityView(tenantId, tbMsg, callback);
  181 + break;
  182 + case ALARM:
  183 + processAlarm(tenantId, tbMsg, callback);
  184 + break;
  185 + default:
  186 + log.debug("Entity type [{}] is not designed to be pushed to edge", tbMsg.getOriginator().getEntityType());
  187 + }
  188 + } catch (IOException e) {
  189 + log.error("Can't push to edge updates, entity type [{}], data [{}]", tbMsg.getOriginator().getEntityType(), tbMsg.getData(), e);
  190 + }
  191 + }
  192 + }
  193 +
  194 +
  195 + private void processCustomTbMsg(TenantId tenantId, TbMsg tbMsg, FutureCallback<Void> callback) {
  196 + ListenableFuture<EdgeId> edgeIdFuture = getEdgeIdByOriginatorId(tenantId, tbMsg.getOriginator());
  197 + Futures.transform(edgeIdFuture, edgeId -> {
  198 + EdgeQueueEntityType edgeQueueEntityType = getEdgeQueueTypeByEntityType(tbMsg.getOriginator().getEntityType());
  199 + if (edgeId != null && edgeQueueEntityType != null) {
  200 + try {
  201 + saveEventToEdgeQueue(tenantId, edgeId, edgeQueueEntityType, tbMsg.getType(), Base64.encodeBase64String(TbMsg.toByteArray(tbMsg)), callback);
  202 + } catch (IOException e) {
  203 + log.error("Error while saving custom tbMsg into Edge Queue", e);
  204 + }
  205 + }
  206 + return null;
  207 + }, MoreExecutors.directExecutor());
  208 + }
  209 +
  210 + private void processDevice(TenantId tenantId, TbMsg tbMsg, FutureCallback<Void> callback) throws IOException {
  211 + switch (tbMsg.getType()) {
  212 + case DataConstants.ENTITY_ASSIGNED_TO_EDGE:
  213 + case DataConstants.ENTITY_UNASSIGNED_FROM_EDGE:
  214 + processAssignedEntity(tenantId, tbMsg, EdgeQueueEntityType.DEVICE, callback);
  215 + break;
  216 + case DataConstants.ENTITY_DELETED:
  217 + case DataConstants.ENTITY_CREATED:
  218 + case DataConstants.ENTITY_UPDATED:
  219 + Device device = mapper.readValue(tbMsg.getData(), Device.class);
  220 + pushEventToEdge(tenantId, device.getId(), EdgeQueueEntityType.DEVICE, tbMsg, callback);
  221 + break;
  222 + default:
  223 + log.warn("Unsupported msgType [{}], tbMsg [{}]", tbMsg.getType(), tbMsg);
  224 + }
  225 + }
  226 +
  227 + private void processEdge(TenantId tenantId, TbMsg tbMsg, FutureCallback<Void> callback) throws IOException {
  228 + switch (tbMsg.getType()) {
  229 + case DataConstants.ENTITY_DELETED:
  230 + case DataConstants.ENTITY_CREATED:
  231 + case DataConstants.ENTITY_UPDATED:
  232 + // TODO: voba - handle properly edge creation
  233 + break;
  234 + default:
  235 + log.warn("Unsupported msgType [{}], tbMsg [{}]", tbMsg.getType(), tbMsg);
  236 + }
  237 + }
  238 +
  239 + private void processAsset(TenantId tenantId, TbMsg tbMsg, FutureCallback<Void> callback) throws IOException {
  240 + switch (tbMsg.getType()) {
  241 + case DataConstants.ENTITY_ASSIGNED_TO_EDGE:
  242 + case DataConstants.ENTITY_UNASSIGNED_FROM_EDGE:
  243 + processAssignedEntity(tenantId, tbMsg, EdgeQueueEntityType.ASSET, callback);
  244 + break;
  245 + case DataConstants.ENTITY_DELETED:
  246 + case DataConstants.ENTITY_CREATED:
  247 + case DataConstants.ENTITY_UPDATED:
  248 + Asset asset = mapper.readValue(tbMsg.getData(), Asset.class);
  249 + pushEventToEdge(tenantId, asset.getId(), EdgeQueueEntityType.ASSET, tbMsg, callback);
  250 + break;
  251 + default:
  252 + log.warn("Unsupported msgType [{}], tbMsg [{}]", tbMsg.getType(), tbMsg);
  253 + }
  254 + }
  255 +
  256 + private void processEntityView(TenantId tenantId, TbMsg tbMsg, FutureCallback<Void> callback) throws IOException {
  257 + switch (tbMsg.getType()) {
  258 + case DataConstants.ENTITY_ASSIGNED_TO_EDGE:
  259 + case DataConstants.ENTITY_UNASSIGNED_FROM_EDGE:
  260 + processAssignedEntity(tenantId, tbMsg, EdgeQueueEntityType.ENTITY_VIEW, callback);
  261 + break;
  262 + case DataConstants.ENTITY_DELETED:
  263 + case DataConstants.ENTITY_CREATED:
  264 + case DataConstants.ENTITY_UPDATED:
  265 + EntityView entityView = mapper.readValue(tbMsg.getData(), EntityView.class);
  266 + pushEventToEdge(tenantId, entityView.getId(), EdgeQueueEntityType.ENTITY_VIEW, tbMsg, callback);
  267 + break;
  268 + default:
  269 + log.warn("Unsupported msgType [{}], tbMsg [{}]", tbMsg.getType(), tbMsg);
  270 + }
  271 + }
  272 +
  273 + private void processAlarm(TenantId tenantId, TbMsg tbMsg, FutureCallback<Void> callback) throws IOException {
  274 + switch (tbMsg.getType()) {
  275 + case DataConstants.ENTITY_DELETED:
  276 + case DataConstants.ENTITY_CREATED:
  277 + case DataConstants.ENTITY_UPDATED:
  278 + case DataConstants.ALARM_ACK:
  279 + case DataConstants.ALARM_CLEAR:
  280 + Alarm alarm = mapper.readValue(tbMsg.getData(), Alarm.class);
  281 + EdgeQueueEntityType edgeQueueEntityType = getEdgeQueueTypeByEntityType(alarm.getOriginator().getEntityType());
  282 + if (edgeQueueEntityType != null) {
  283 + pushEventToEdge(tenantId, alarm.getOriginator(), EdgeQueueEntityType.ALARM, tbMsg, callback);
  284 + }
  285 + break;
  286 + default:
  287 + log.warn("Unsupported msgType [{}], tbMsg [{}]", tbMsg.getType(), tbMsg);
  288 + }
  289 + }
  290 +
  291 + private void processDashboard(TenantId tenantId, TbMsg tbMsg, FutureCallback<Void> callback) throws IOException {
  292 + switch (tbMsg.getType()) {
  293 + case DataConstants.ENTITY_ASSIGNED_TO_EDGE:
  294 + case DataConstants.ENTITY_UNASSIGNED_FROM_EDGE:
  295 + processAssignedEntity(tenantId, tbMsg, EdgeQueueEntityType.DASHBOARD, callback);
  296 + break;
  297 + case DataConstants.ENTITY_DELETED:
  298 + case DataConstants.ENTITY_CREATED:
  299 + case DataConstants.ENTITY_UPDATED:
  300 + Dashboard dashboard = mapper.readValue(tbMsg.getData(), Dashboard.class);
  301 + ListenableFuture<TimePageData<Edge>> future = edgeService.findEdgesByTenantIdAndDashboardId(tenantId, dashboard.getId(), new TimePageLink(Integer.MAX_VALUE));
  302 + Futures.transform(future, edges -> {
  303 + if (edges != null && edges.getData() != null && !edges.getData().isEmpty()) {
  304 + try {
  305 + for (Edge edge : edges.getData()) {
  306 + pushEventToEdge(tenantId, edge.getId(), EdgeQueueEntityType.DASHBOARD, tbMsg, callback);
  307 + }
  308 + } catch (IOException e) {
  309 + log.error("Can't push event to edge", e);
  310 + }
  311 + }
  312 + return null;
  313 + }, MoreExecutors.directExecutor());
  314 + break;
  315 + default:
  316 + log.warn("Unsupported msgType [{}], tbMsg [{}]", tbMsg.getType(), tbMsg);
  317 + }
  318 + }
  319 +
  320 + private void processRuleChain(TenantId tenantId, TbMsg tbMsg, FutureCallback<Void> callback) throws IOException {
  321 + switch (tbMsg.getType()) {
  322 + case DataConstants.ENTITY_ASSIGNED_TO_EDGE:
  323 + case DataConstants.ENTITY_UNASSIGNED_FROM_EDGE:
  324 + processAssignedEntity(tenantId, tbMsg, EdgeQueueEntityType.RULE_CHAIN, callback);
  325 + break;
  326 + case DataConstants.ENTITY_DELETED:
  327 + case DataConstants.ENTITY_CREATED:
  328 + case DataConstants.ENTITY_UPDATED:
  329 + RuleChain ruleChain = mapper.readValue(tbMsg.getData(), RuleChain.class);
  330 + if (RuleChainType.EDGE.equals(ruleChain.getType())) {
  331 + ListenableFuture<TimePageData<Edge>> future = edgeService.findEdgesByTenantIdAndRuleChainId(tenantId, ruleChain.getId(), new TimePageLink(Integer.MAX_VALUE));
  332 + Futures.transform(future, edges -> {
  333 + if (edges != null && edges.getData() != null && !edges.getData().isEmpty()) {
  334 + try {
  335 + for (Edge edge : edges.getData()) {
  336 + pushEventToEdge(tenantId, edge.getId(), EdgeQueueEntityType.RULE_CHAIN, tbMsg, callback);
  337 + }
  338 + } catch (IOException e) {
  339 + log.error("Can't push event to edge", e);
  340 + }
  341 + }
  342 + return null;
  343 + }, MoreExecutors.directExecutor());
  344 + }
  345 + break;
  346 + default:
  347 + log.warn("Unsupported msgType [{}], tbMsg [{}]", tbMsg.getType(), tbMsg);
  348 + }
  349 + }
  350 +
  351 +
  352 + private void processAssignedEntity(TenantId tenantId, TbMsg tbMsg, EdgeQueueEntityType entityType, FutureCallback<Void> callback) throws IOException {
  353 + EdgeId edgeId;
  354 + switch (tbMsg.getType()) {
  355 + case DataConstants.ENTITY_ASSIGNED_TO_EDGE:
  356 + edgeId = new EdgeId(UUID.fromString(tbMsg.getMetaData().getValue("assignedEdgeId")));
  357 + pushEventToEdge(tenantId, edgeId, entityType, tbMsg, callback);
  358 + break;
  359 + case DataConstants.ENTITY_UNASSIGNED_FROM_EDGE:
  360 + edgeId = new EdgeId(UUID.fromString(tbMsg.getMetaData().getValue("unassignedEdgeId")));
  361 + pushEventToEdge(tenantId, edgeId, entityType, tbMsg, callback);
  362 + break;
  363 +
  364 + }
  365 + }
  366 +
  367 +
  368 + private void pushEventToEdge(TenantId tenantId, EntityId originatorId, EdgeQueueEntityType edgeQueueEntityType, TbMsg tbMsg, FutureCallback<Void> callback) {
  369 + ListenableFuture<EdgeId> edgeIdFuture = getEdgeIdByOriginatorId(tenantId, originatorId);
  370 + Futures.transform(edgeIdFuture, edgeId -> {
  371 + if (edgeId != null) {
  372 + try {
  373 + pushEventToEdge(tenantId, edgeId, edgeQueueEntityType, tbMsg, callback);
  374 + } catch (Exception e) {
  375 + log.error("Failed to push event to edge, edgeId [{}], tbMsg [{}]", edgeId, tbMsg, e);
  376 + }
  377 + }
  378 + return null;
  379 + },
  380 + MoreExecutors.directExecutor());
  381 + }
  382 +
  383 +
  384 + private ListenableFuture<EdgeId> getEdgeIdByOriginatorId(TenantId tenantId, EntityId originatorId) {
  385 + List<EntityRelation> originatorEdgeRelations = relationService.findByToAndType(tenantId, originatorId, EntityRelation.CONTAINS_TYPE, RelationTypeGroup.EDGE);
  386 + if (originatorEdgeRelations != null && originatorEdgeRelations.size() > 0) {
  387 + return Futures.immediateFuture(new EdgeId(originatorEdgeRelations.get(0).getFrom().getId()));
  388 + } else {
  389 + return Futures.immediateFuture(null);
  390 + }
  391 + }
  392 +
  393 +
  394 + private void pushEventToEdge(TenantId tenantId, EdgeId edgeId, EdgeQueueEntityType entityType, TbMsg tbMsg, FutureCallback<Void> callback) throws IOException {
  395 + log.debug("Pushing event(s) to edge queue. tenantId [{}], edgeId [{}], entityType [{}], tbMsg [{}]", tenantId, edgeId, entityType, tbMsg);
  396 +
  397 + saveEventToEdgeQueue(tenantId, edgeId, entityType, tbMsg.getType(), tbMsg.getData(), callback);
  398 +
  399 + if (entityType.equals(EdgeQueueEntityType.RULE_CHAIN)) {
  400 + pushRuleChainMetadataToEdge(tenantId, edgeId, tbMsg, callback);
  401 + }
  402 + }
  403 +
  404 + private void pushRuleChainMetadataToEdge(TenantId tenantId, EdgeId edgeId, TbMsg tbMsg, FutureCallback<Void> callback) throws IOException {
  405 + RuleChain ruleChain = mapper.readValue(tbMsg.getData(), RuleChain.class);
  406 + switch (tbMsg.getType()) {
  407 + case DataConstants.ENTITY_ASSIGNED_TO_EDGE:
  408 + case DataConstants.ENTITY_UNASSIGNED_FROM_EDGE:
  409 + case DataConstants.ENTITY_UPDATED:
  410 + RuleChainMetaData ruleChainMetaData = ruleChainService.loadRuleChainMetaData(tenantId, ruleChain.getId());
  411 + saveEventToEdgeQueue(tenantId, edgeId, EdgeQueueEntityType.RULE_CHAIN_METADATA, tbMsg.getType(), mapper.writeValueAsString(ruleChainMetaData), callback);
  412 + break;
  413 + default:
  414 + log.warn("Unsupported msgType [{}], tbMsg [{}]", tbMsg.getType(), tbMsg);
  415 + }
  416 + }
  417 +
  418 +
  419 + private EdgeQueueEntityType getEdgeQueueTypeByEntityType(EntityType entityType) {
  420 + switch (entityType) {
  421 + case DEVICE:
  422 + return EdgeQueueEntityType.DEVICE;
  423 + case ASSET:
  424 + return EdgeQueueEntityType.ASSET;
  425 + case ENTITY_VIEW:
  426 + return EdgeQueueEntityType.ENTITY_VIEW;
  427 + default:
  428 + log.info("Unsupported entity type: [{}]", entityType);
  429 + return null;
  430 + }
  431 + }
  432 +}
  433 +
... ...
... ... @@ -55,6 +55,10 @@ public class EdgeContextComponent {
55 55
56 56 @Lazy
57 57 @Autowired
  58 + private EdgeNotificationService edgeNotificationService;
  59 +
  60 + @Lazy
  61 + @Autowired
58 62 private AssetService assetService;
59 63
60 64 @Lazy
... ...
  1 +package org.thingsboard.server.service.edge;
  2 +
  3 +import org.thingsboard.server.common.data.Event;
  4 +import org.thingsboard.server.common.data.edge.Edge;
  5 +import org.thingsboard.server.common.data.id.EdgeId;
  6 +import org.thingsboard.server.common.data.id.RuleChainId;
  7 +import org.thingsboard.server.common.data.id.TenantId;
  8 +import org.thingsboard.server.common.data.page.TimePageData;
  9 +import org.thingsboard.server.common.data.page.TimePageLink;
  10 +import org.thingsboard.server.common.msg.queue.TbCallback;
  11 +import org.thingsboard.server.gen.transport.TransportProtos;
  12 +
  13 +import java.io.IOException;
  14 +
  15 +public interface EdgeNotificationService {
  16 +
  17 + TimePageData<Event> findQueueEvents(TenantId tenantId, EdgeId edgeId, TimePageLink pageLink);
  18 +
  19 + Edge setEdgeRootRuleChain(TenantId tenantId, Edge edge, RuleChainId ruleChainId) throws IOException;
  20 +
  21 + void pushNotificationToEdge(TransportProtos.EdgeNotificationMsgProto edgeNotificationMsg, TbCallback callback);
  22 +}
... ...
... ... @@ -172,7 +172,7 @@ public final class EdgeGrpcSession implements Closeable {
172 172 TimePageData<Event> pageData;
173 173 UUID ifOffset = null;
174 174 do {
175   - pageData = ctx.getEdgeService().findQueueEvents(edge.getTenantId(), edge.getId(), pageLink);
  175 + pageData = ctx.getEdgeNotificationService().findQueueEvents(edge.getTenantId(), edge.getId(), pageLink);
176 176 if (isConnected() && !pageData.getData().isEmpty()) {
177 177 log.trace("[{}] [{}] event(s) are going to be processed.", this.sessionId, pageData.getData().size());
178 178 for (Event event : pageData.getData()) {
... ...
... ... @@ -28,6 +28,7 @@ import org.thingsboard.server.common.msg.TbActorMsg;
28 28 import org.thingsboard.server.common.msg.queue.ServiceType;
29 29 import org.thingsboard.server.common.msg.queue.TbCallback;
30 30 import org.thingsboard.server.gen.transport.TransportProtos.DeviceStateServiceMsgProto;
  31 +import org.thingsboard.server.gen.transport.TransportProtos.EdgeNotificationMsgProto;
31 32 import org.thingsboard.server.gen.transport.TransportProtos.FromDeviceRPCResponseProto;
32 33 import org.thingsboard.server.gen.transport.TransportProtos.LocalSubscriptionServiceMsgProto;
33 34 import org.thingsboard.server.gen.transport.TransportProtos.SubscriptionMgrMsgProto;
... ... @@ -42,6 +43,7 @@ import org.thingsboard.server.queue.common.TbProtoQueueMsg;
42 43 import org.thingsboard.server.queue.discovery.PartitionChangeEvent;
43 44 import org.thingsboard.server.queue.provider.TbCoreQueueFactory;
44 45 import org.thingsboard.server.queue.util.TbCoreComponent;
  46 +import org.thingsboard.server.service.edge.EdgeNotificationService;
45 47 import org.thingsboard.server.service.encoding.DataDecodingEncodingService;
46 48 import org.thingsboard.server.service.queue.processing.AbstractConsumerService;
47 49 import org.thingsboard.server.service.rpc.FromDeviceRpcResponse;
... ... @@ -82,18 +84,20 @@ public class DefaultTbCoreConsumerService extends AbstractConsumerService<ToCore
82 84 private final TbLocalSubscriptionService localSubscriptionService;
83 85 private final SubscriptionManagerService subscriptionManagerService;
84 86 private final TbCoreDeviceRpcService tbCoreDeviceRpcService;
  87 + private final EdgeNotificationService edgeNotificationService;
85 88 private final TbCoreConsumerStats stats = new TbCoreConsumerStats();
86 89
87 90 public DefaultTbCoreConsumerService(TbCoreQueueFactory tbCoreQueueFactory, ActorSystemContext actorContext,
88 91 DeviceStateService stateService, TbLocalSubscriptionService localSubscriptionService,
89 92 SubscriptionManagerService subscriptionManagerService, DataDecodingEncodingService encodingService,
90   - TbCoreDeviceRpcService tbCoreDeviceRpcService) {
  93 + TbCoreDeviceRpcService tbCoreDeviceRpcService, EdgeNotificationService edgeNotificationService) {
91 94 super(actorContext, encodingService, tbCoreQueueFactory.createToCoreNotificationsMsgConsumer());
92 95 this.mainConsumer = tbCoreQueueFactory.createToCoreMsgConsumer();
93 96 this.stateService = stateService;
94 97 this.localSubscriptionService = localSubscriptionService;
95 98 this.subscriptionManagerService = subscriptionManagerService;
96 99 this.tbCoreDeviceRpcService = tbCoreDeviceRpcService;
  100 + this.edgeNotificationService = edgeNotificationService;
97 101 }
98 102
99 103 @PostConstruct
... ... @@ -142,6 +146,9 @@ public class DefaultTbCoreConsumerService extends AbstractConsumerService<ToCore
142 146 } else if (toCoreMsg.hasDeviceStateServiceMsg()) {
143 147 log.trace("[{}] Forwarding message to state service {}", id, toCoreMsg.getDeviceStateServiceMsg());
144 148 forwardToStateService(toCoreMsg.getDeviceStateServiceMsg(), callback);
  149 + } else if (toCoreMsg.hasEdgeNotificationMsg()) {
  150 + log.trace("[{}] Forwarding message to edge service {}", id, toCoreMsg.getEdgeNotificationMsg());
  151 + forwardToEdgeNotificationService(toCoreMsg.getEdgeNotificationMsg(), callback);
145 152 } else if (toCoreMsg.getToDeviceActorNotificationMsg() != null && !toCoreMsg.getToDeviceActorNotificationMsg().isEmpty()) {
146 153 Optional<TbActorMsg> actorMsg = encodingService.decode(toCoreMsg.getToDeviceActorNotificationMsg().toByteArray());
147 154 if (actorMsg.isPresent()) {
... ... @@ -275,6 +282,13 @@ public class DefaultTbCoreConsumerService extends AbstractConsumerService<ToCore
275 282 stateService.onQueueMsg(deviceStateServiceMsg, callback);
276 283 }
277 284
  285 + private void forwardToEdgeNotificationService(EdgeNotificationMsgProto edgeNotificationMsg, TbCallback callback) {
  286 + if (statsEnabled) {
  287 + stats.log(edgeNotificationMsg);
  288 + }
  289 + edgeNotificationService.pushNotificationToEdge(edgeNotificationMsg, callback);
  290 + }
  291 +
278 292 private void forwardToDeviceActor(TransportToDeviceActorMsg toDeviceActorMsg, TbCallback callback) {
279 293 if (statsEnabled) {
280 294 stats.log(toDeviceActorMsg);
... ...
... ... @@ -33,6 +33,7 @@ public class TbCoreConsumerStats {
33 33 private final AtomicInteger claimDeviceCounter = new AtomicInteger(0);
34 34
35 35 private final AtomicInteger deviceStateCounter = new AtomicInteger(0);
  36 + private final AtomicInteger edgeNotificationCounter = new AtomicInteger(0);
36 37 private final AtomicInteger subscriptionMsgCounter = new AtomicInteger(0);
37 38 private final AtomicInteger toCoreNotificationsCounter = new AtomicInteger(0);
38 39
... ... @@ -66,6 +67,11 @@ public class TbCoreConsumerStats {
66 67 deviceStateCounter.incrementAndGet();
67 68 }
68 69
  70 + public void log(TransportProtos.EdgeNotificationMsgProto msg) {
  71 + totalCounter.incrementAndGet();
  72 + edgeNotificationCounter.incrementAndGet();
  73 + }
  74 +
69 75 public void log(TransportProtos.SubscriptionMgrMsgProto msg) {
70 76 totalCounter.incrementAndGet();
71 77 subscriptionMsgCounter.incrementAndGet();
... ...
... ... @@ -5,7 +5,7 @@
5 5 * you may not use this file except in compliance with the License.
6 6 * You may obtain a copy of the License at
7 7 *
8   - * http://www.apache.org/licenses/LICENSE-2.0
  8 + * http://www.apache.org/licenses/LICENSE-2.0
9 9 *
10 10 * Unless required by applicable law or agreed to in writing, software
11 11 * distributed under the License is distributed on an "AS IS" BASIS,
... ... @@ -15,10 +15,8 @@
15 15 */
16 16 package org.thingsboard.server.dao.edge;
17 17
18   -import com.google.common.util.concurrent.FutureCallback;
19 18 import com.google.common.util.concurrent.ListenableFuture;
20 19 import org.thingsboard.server.common.data.EntitySubtype;
21   -import org.thingsboard.server.common.data.Event;
22 20 import org.thingsboard.server.common.data.edge.Edge;
23 21 import org.thingsboard.server.common.data.edge.EdgeSearchQuery;
24 22 import org.thingsboard.server.common.data.id.CustomerId;
... ... @@ -30,13 +28,9 @@ import org.thingsboard.server.common.data.page.TextPageData;
30 28 import org.thingsboard.server.common.data.page.TextPageLink;
31 29 import org.thingsboard.server.common.data.page.TimePageData;
32 30 import org.thingsboard.server.common.data.page.TimePageLink;
33   -import org.thingsboard.server.common.data.rule.RuleChain;
34   -import org.thingsboard.server.common.msg.TbMsg;
35 31
36   -import java.io.IOException;
37 32 import java.util.List;
38 33 import java.util.Optional;
39   -import java.util.UUID;
40 34
41 35 public interface EdgeService {
42 36
... ... @@ -76,11 +70,6 @@ public interface EdgeService {
76 70
77 71 ListenableFuture<List<EntitySubtype>> findEdgeTypesByTenantId(TenantId tenantId);
78 72
79   - void pushEventToEdge(TenantId tenantId, TbMsg tbMsg, FutureCallback<Void> callback);
80   -
81   - TimePageData<Event> findQueueEvents(TenantId tenantId, EdgeId edgeId, TimePageLink pageLink);
82   -
83   - Edge setEdgeRootRuleChain(TenantId tenantId, Edge edge, RuleChainId ruleChainId) throws IOException;
84 73
85 74 void assignDefaultRuleChainsToEdge(TenantId tenantId, EdgeId edgeId);
86 75
... ...
... ... @@ -352,6 +352,12 @@ message FromDeviceRPCResponseProto {
352 352 string response = 3;
353 353 int32 error = 4;
354 354 }
  355 +
  356 +message EdgeNotificationMsgProto {
  357 + int64 tenantIdMSB = 1;
  358 + int64 tenantIdLSB = 2;
  359 +}
  360 +
355 361 /**
356 362 * Main messages;
357 363 */
... ... @@ -377,6 +383,7 @@ message ToCoreMsg {
377 383 DeviceStateServiceMsgProto deviceStateServiceMsg = 2;
378 384 SubscriptionMgrMsgProto toSubscriptionMgrMsg = 3;
379 385 bytes toDeviceActorNotificationMsg = 4;
  386 + EdgeNotificationMsgProto edgeNotificationMsg = 5;
380 387 }
381 388
382 389 /* High priority messages with low latency are handled by ThingsBoard Core Service separately */
... ...
... ... @@ -5,7 +5,7 @@
5 5 * you may not use this file except in compliance with the License.
6 6 * You may obtain a copy of the License at
7 7 *
8   - * http://www.apache.org/licenses/LICENSE-2.0
  8 + * http://www.apache.org/licenses/LICENSE-2.0
9 9 *
10 10 * Unless required by applicable law or agreed to in writing, software
11 11 * distributed under the License is distributed on an "AS IS" BASIS,
... ... @@ -15,14 +15,11 @@
15 15 */
16 16 package org.thingsboard.server.dao.edge;
17 17
18   -import com.fasterxml.jackson.databind.ObjectMapper;
19 18 import com.google.common.base.Function;
20   -import com.google.common.util.concurrent.FutureCallback;
21 19 import com.google.common.util.concurrent.Futures;
22 20 import com.google.common.util.concurrent.ListenableFuture;
23 21 import com.google.common.util.concurrent.MoreExecutors;
24 22 import lombok.extern.slf4j.Slf4j;
25   -import org.apache.commons.codec.binary.Base64;
26 23 import org.springframework.beans.factory.annotation.Autowired;
27 24 import org.springframework.cache.Cache;
28 25 import org.springframework.cache.CacheManager;
... ... @@ -31,19 +28,10 @@ import org.springframework.cache.annotation.Cacheable;
31 28 import org.springframework.stereotype.Service;
32 29 import org.springframework.util.StringUtils;
33 30 import org.thingsboard.server.common.data.Customer;
34   -import org.thingsboard.server.common.data.Dashboard;
35   -import org.thingsboard.server.common.data.DataConstants;
36   -import org.thingsboard.server.common.data.Device;
37 31 import org.thingsboard.server.common.data.EntitySubtype;
38 32 import org.thingsboard.server.common.data.EntityType;
39   -import org.thingsboard.server.common.data.EntityView;
40   -import org.thingsboard.server.common.data.Event;
41 33 import org.thingsboard.server.common.data.Tenant;
42   -import org.thingsboard.server.common.data.alarm.Alarm;
43   -import org.thingsboard.server.common.data.asset.Asset;
44 34 import org.thingsboard.server.common.data.edge.Edge;
45   -import org.thingsboard.server.common.data.edge.EdgeQueueEntityType;
46   -import org.thingsboard.server.common.data.edge.EdgeQueueEntry;
47 35 import org.thingsboard.server.common.data.edge.EdgeSearchQuery;
48 36 import org.thingsboard.server.common.data.id.CustomerId;
49 37 import org.thingsboard.server.common.data.id.DashboardId;
... ... @@ -57,19 +45,10 @@ import org.thingsboard.server.common.data.page.TimePageData;
57 45 import org.thingsboard.server.common.data.page.TimePageLink;
58 46 import org.thingsboard.server.common.data.relation.EntityRelation;
59 47 import org.thingsboard.server.common.data.relation.EntitySearchDirection;
60   -import org.thingsboard.server.common.data.relation.RelationTypeGroup;
61 48 import org.thingsboard.server.common.data.rule.RuleChain;
62   -import org.thingsboard.server.common.data.rule.RuleChainMetaData;
63   -import org.thingsboard.server.common.data.rule.RuleChainType;
64   -import org.thingsboard.server.common.msg.TbMsg;
65   -import org.thingsboard.server.common.msg.session.SessionMsgType;
66   -import org.thingsboard.server.dao.asset.AssetService;
67 49 import org.thingsboard.server.dao.customer.CustomerDao;
68 50 import org.thingsboard.server.dao.dashboard.DashboardService;
69   -import org.thingsboard.server.dao.device.DeviceService;
70 51 import org.thingsboard.server.dao.entity.AbstractEntityService;
71   -import org.thingsboard.server.dao.entityview.EntityViewService;
72   -import org.thingsboard.server.dao.event.EventService;
73 52 import org.thingsboard.server.dao.exception.DataValidationException;
74 53 import org.thingsboard.server.dao.relation.RelationService;
75 54 import org.thingsboard.server.dao.rule.RuleChainService;
... ... @@ -79,17 +58,11 @@ import org.thingsboard.server.dao.service.Validator;
79 58 import org.thingsboard.server.dao.tenant.TenantDao;
80 59
81 60 import javax.annotation.Nullable;
82   -import javax.annotation.PostConstruct;
83   -import javax.annotation.PreDestroy;
84   -import java.io.IOException;
85 61 import java.util.ArrayList;
86 62 import java.util.Collections;
87 63 import java.util.Comparator;
88 64 import java.util.List;
89 65 import java.util.Optional;
90   -import java.util.UUID;
91   -import java.util.concurrent.ExecutorService;
92   -import java.util.concurrent.Executors;
93 66 import java.util.stream.Collectors;
94 67
95 68 import static org.thingsboard.server.common.data.CacheConstants.EDGE_CACHE;
... ... @@ -104,8 +77,6 @@ import static org.thingsboard.server.dao.service.Validator.validateString;
104 77 @Slf4j
105 78 public class EdgeServiceImpl extends AbstractEntityService implements EdgeService {
106 79
107   - private static final ObjectMapper mapper = new ObjectMapper();
108   -
109 80 public static final String INCORRECT_TENANT_ID = "Incorrect tenantId ";
110 81 public static final String INCORRECT_PAGE_LINK = "Incorrect page link ";
111 82 public static final String INCORRECT_CUSTOMER_ID = "Incorrect customerId ";
... ... @@ -124,40 +95,14 @@ public class EdgeServiceImpl extends AbstractEntityService implements EdgeServic
124 95 private CacheManager cacheManager;
125 96
126 97 @Autowired
127   - private EventService eventService;
128   -
129   - @Autowired
130 98 private DashboardService dashboardService;
131 99
132 100 @Autowired
133 101 private RuleChainService ruleChainService;
134 102
135 103 @Autowired
136   - private DeviceService deviceService;
137   -
138   - @Autowired
139   - private AssetService assetService;
140   -
141   - @Autowired
142   - private EntityViewService entityViewService;
143   -
144   - @Autowired
145 104 private RelationService relationService;
146 105
147   - private ExecutorService tsCallBackExecutor;
148   -
149   - @PostConstruct
150   - public void initExecutor() {
151   - tsCallBackExecutor = Executors.newSingleThreadExecutor();
152   - }
153   -
154   - @PreDestroy
155   - public void shutdownExecutor() {
156   - if (tsCallBackExecutor != null) {
157   - tsCallBackExecutor.shutdownNow();
158   - }
159   - }
160   -
161 106 @Override
162 107 public Edge findEdgeById(TenantId tenantId, EdgeId edgeId) {
163 108 log.trace("Executing findEdgeById [{}]", edgeId);
... ... @@ -259,7 +204,6 @@ public class EdgeServiceImpl extends AbstractEntityService implements EdgeServic
259 204 return edgeDao.findEdgesByTenantIdAndIdsAsync(tenantId.getId(), toUUIDs(edgeIds));
260 205 }
261 206
262   -
263 207 @Override
264 208 public void deleteEdgesByTenantId(TenantId tenantId) {
265 209 log.trace("Executing deleteEdgesByTenantId, tenantId [{}]", tenantId);
... ... @@ -345,327 +289,6 @@ public class EdgeServiceImpl extends AbstractEntityService implements EdgeServic
345 289 }
346 290
347 291 @Override
348   - public void pushEventToEdge(TenantId tenantId, TbMsg tbMsg, FutureCallback<Void> callback) {
349   - if (tbMsg.getType().equals(SessionMsgType.POST_TELEMETRY_REQUEST.name()) ||
350   - tbMsg.getType().equals(SessionMsgType.POST_ATTRIBUTES_REQUEST.name()) ||
351   - tbMsg.getType().equals(DataConstants.ATTRIBUTES_UPDATED) ||
352   - tbMsg.getType().equals(DataConstants.ATTRIBUTES_DELETED)) {
353   - processCustomTbMsg(tenantId, tbMsg, callback);
354   - } else {
355   - try {
356   - switch (tbMsg.getOriginator().getEntityType()) {
357   - case EDGE:
358   - processEdge(tenantId, tbMsg, callback);
359   - break;
360   - case ASSET:
361   - processAsset(tenantId, tbMsg, callback);
362   - break;
363   - case DEVICE:
364   - processDevice(tenantId, tbMsg, callback);
365   - break;
366   - case DASHBOARD:
367   - processDashboard(tenantId, tbMsg, callback);
368   - break;
369   - case RULE_CHAIN:
370   - processRuleChain(tenantId, tbMsg, callback);
371   - break;
372   - case ENTITY_VIEW:
373   - processEntityView(tenantId, tbMsg, callback);
374   - break;
375   - case ALARM:
376   - processAlarm(tenantId, tbMsg, callback);
377   - break;
378   - default:
379   - log.debug("Entity type [{}] is not designed to be pushed to edge", tbMsg.getOriginator().getEntityType());
380   - }
381   - } catch (IOException e) {
382   - log.error("Can't push to edge updates, entity type [{}], data [{}]", tbMsg.getOriginator().getEntityType(), tbMsg.getData(), e);
383   - }
384   - }
385   - }
386   -
387   - private void processCustomTbMsg(TenantId tenantId, TbMsg tbMsg, FutureCallback<Void> callback) {
388   - ListenableFuture<EdgeId> edgeIdFuture = getEdgeIdByOriginatorId(tenantId, tbMsg.getOriginator());
389   - Futures.transform(edgeIdFuture, edgeId -> {
390   - EdgeQueueEntityType edgeQueueEntityType = getEdgeQueueTypeByEntityType(tbMsg.getOriginator().getEntityType());
391   - if (edgeId != null && edgeQueueEntityType != null) {
392   - try {
393   - saveEventToEdgeQueue(tenantId, edgeId, edgeQueueEntityType, tbMsg.getType(), Base64.encodeBase64String(TbMsg.toByteArray(tbMsg)), callback);
394   - } catch (IOException e) {
395   - log.error("Error while saving custom tbMsg into Edge Queue", e);
396   - }
397   - }
398   - return null;
399   - }, MoreExecutors.directExecutor());
400   - }
401   -
402   - private EdgeQueueEntityType getEdgeQueueTypeByEntityType(EntityType entityType) {
403   - switch (entityType) {
404   - case DEVICE:
405   - return EdgeQueueEntityType.DEVICE;
406   - case ASSET:
407   - return EdgeQueueEntityType.ASSET;
408   - case ENTITY_VIEW:
409   - return EdgeQueueEntityType.ENTITY_VIEW;
410   - default:
411   - log.info("Unsupported entity type: [{}]", entityType);
412   - return null;
413   - }
414   - }
415   -
416   - private ListenableFuture<EdgeId> getEdgeIdByOriginatorId(TenantId tenantId, EntityId originatorId) {
417   - List<EntityRelation> originatorEdgeRelations = relationService.findByToAndType(tenantId, originatorId, EntityRelation.CONTAINS_TYPE, RelationTypeGroup.EDGE);
418   - if (originatorEdgeRelations != null && originatorEdgeRelations.size() > 0) {
419   - return Futures.immediateFuture(new EdgeId(originatorEdgeRelations.get(0).getFrom().getId()));
420   - } else {
421   - return Futures.immediateFuture(null);
422   - }
423   - }
424   -
425   - private void pushEventToEdge(TenantId tenantId, EntityId originatorId, EdgeQueueEntityType edgeQueueEntityType, TbMsg tbMsg, FutureCallback<Void> callback) {
426   - ListenableFuture<EdgeId> edgeIdFuture = getEdgeIdByOriginatorId(tenantId, originatorId);
427   - Futures.transform(edgeIdFuture, edgeId -> {
428   - if (edgeId != null) {
429   - try {
430   - pushEventToEdge(tenantId, edgeId, edgeQueueEntityType, tbMsg, callback);
431   - } catch (Exception e) {
432   - log.error("Failed to push event to edge, edgeId [{}], tbMsg [{}]", edgeId, tbMsg, e);
433   - }
434   - }
435   - return null;
436   - },
437   - MoreExecutors.directExecutor());
438   - }
439   -
440   - private void processDevice(TenantId tenantId, TbMsg tbMsg, FutureCallback<Void> callback) throws IOException {
441   - switch (tbMsg.getType()) {
442   - case DataConstants.ENTITY_ASSIGNED_TO_EDGE:
443   - case DataConstants.ENTITY_UNASSIGNED_FROM_EDGE:
444   - processAssignedEntity(tenantId, tbMsg, EdgeQueueEntityType.DEVICE, callback);
445   - break;
446   - case DataConstants.ENTITY_DELETED:
447   - case DataConstants.ENTITY_CREATED:
448   - case DataConstants.ENTITY_UPDATED:
449   - Device device = mapper.readValue(tbMsg.getData(), Device.class);
450   - pushEventToEdge(tenantId, device.getId(), EdgeQueueEntityType.DEVICE, tbMsg, callback);
451   - break;
452   - default:
453   - log.warn("Unsupported msgType [{}], tbMsg [{}]", tbMsg.getType(), tbMsg);
454   - }
455   - }
456   -
457   - private void processEdge(TenantId tenantId, TbMsg tbMsg, FutureCallback<Void> callback) throws IOException {
458   - switch (tbMsg.getType()) {
459   - case DataConstants.ENTITY_DELETED:
460   - case DataConstants.ENTITY_CREATED:
461   - case DataConstants.ENTITY_UPDATED:
462   - // TODO: voba - handle properly edge creation
463   - break;
464   - default:
465   - log.warn("Unsupported msgType [{}], tbMsg [{}]", tbMsg.getType(), tbMsg);
466   - }
467   - }
468   -
469   - private void processAsset(TenantId tenantId, TbMsg tbMsg, FutureCallback<Void> callback) throws IOException {
470   - switch (tbMsg.getType()) {
471   - case DataConstants.ENTITY_ASSIGNED_TO_EDGE:
472   - case DataConstants.ENTITY_UNASSIGNED_FROM_EDGE:
473   - processAssignedEntity(tenantId, tbMsg, EdgeQueueEntityType.ASSET, callback);
474   - break;
475   - case DataConstants.ENTITY_DELETED:
476   - case DataConstants.ENTITY_CREATED:
477   - case DataConstants.ENTITY_UPDATED:
478   - Asset asset = mapper.readValue(tbMsg.getData(), Asset.class);
479   - pushEventToEdge(tenantId, asset.getId(), EdgeQueueEntityType.ASSET, tbMsg, callback);
480   - break;
481   - default:
482   - log.warn("Unsupported msgType [{}], tbMsg [{}]", tbMsg.getType(), tbMsg);
483   - }
484   - }
485   -
486   - private void processEntityView(TenantId tenantId, TbMsg tbMsg, FutureCallback<Void> callback) throws IOException {
487   - switch (tbMsg.getType()) {
488   - case DataConstants.ENTITY_ASSIGNED_TO_EDGE:
489   - case DataConstants.ENTITY_UNASSIGNED_FROM_EDGE:
490   - processAssignedEntity(tenantId, tbMsg, EdgeQueueEntityType.ENTITY_VIEW, callback);
491   - break;
492   - case DataConstants.ENTITY_DELETED:
493   - case DataConstants.ENTITY_CREATED:
494   - case DataConstants.ENTITY_UPDATED:
495   - EntityView entityView = mapper.readValue(tbMsg.getData(), EntityView.class);
496   - pushEventToEdge(tenantId, entityView.getId(), EdgeQueueEntityType.ENTITY_VIEW, tbMsg, callback);
497   - break;
498   - default:
499   - log.warn("Unsupported msgType [{}], tbMsg [{}]", tbMsg.getType(), tbMsg);
500   - }
501   - }
502   -
503   - private void processAlarm(TenantId tenantId, TbMsg tbMsg, FutureCallback<Void> callback) throws IOException {
504   - switch (tbMsg.getType()) {
505   - case DataConstants.ENTITY_DELETED:
506   - case DataConstants.ENTITY_CREATED:
507   - case DataConstants.ENTITY_UPDATED:
508   - case DataConstants.ALARM_ACK:
509   - case DataConstants.ALARM_CLEAR:
510   - Alarm alarm = mapper.readValue(tbMsg.getData(), Alarm.class);
511   - EdgeQueueEntityType edgeQueueEntityType = getEdgeQueueTypeByEntityType(alarm.getOriginator().getEntityType());
512   - if (edgeQueueEntityType != null) {
513   - pushEventToEdge(tenantId, alarm.getOriginator(), EdgeQueueEntityType.ALARM, tbMsg, callback);
514   - }
515   - break;
516   - default:
517   - log.warn("Unsupported msgType [{}], tbMsg [{}]", tbMsg.getType(), tbMsg);
518   - }
519   - }
520   -
521   - private void processDashboard(TenantId tenantId, TbMsg tbMsg, FutureCallback<Void> callback) throws IOException {
522   - processAssignedEntity(tenantId, tbMsg, EdgeQueueEntityType.DASHBOARD, callback);
523   - }
524   -
525   - private void processRuleChain(TenantId tenantId, TbMsg tbMsg, FutureCallback<Void> callback) throws IOException {
526   - switch (tbMsg.getType()) {
527   - case DataConstants.ENTITY_ASSIGNED_TO_EDGE:
528   - case DataConstants.ENTITY_UNASSIGNED_FROM_EDGE:
529   - processAssignedEntity(tenantId, tbMsg, EdgeQueueEntityType.RULE_CHAIN, callback);
530   - break;
531   - case DataConstants.ENTITY_DELETED:
532   - case DataConstants.ENTITY_CREATED:
533   - case DataConstants.ENTITY_UPDATED:
534   - RuleChain ruleChain = mapper.readValue(tbMsg.getData(), RuleChain.class);
535   - if (RuleChainType.EDGE.equals(ruleChain.getType())) {
536   - ListenableFuture<TimePageData<Edge>> future = findEdgesByTenantIdAndRuleChainId(tenantId, ruleChain.getId(), new TimePageLink(Integer.MAX_VALUE));
537   - Futures.transform(future, edges -> {
538   - if (edges != null && edges.getData() != null && !edges.getData().isEmpty()) {
539   - try {
540   - for (Edge edge : edges.getData()) {
541   - pushEventToEdge(tenantId, edge.getId(), EdgeQueueEntityType.RULE_CHAIN, tbMsg, callback);
542   - }
543   - } catch (IOException e) {
544   - log.error("Can't push event to edge", e);
545   - }
546   - }
547   - return null;
548   - }, MoreExecutors.directExecutor());
549   - }
550   - break;
551   - default:
552   - log.warn("Unsupported msgType [{}], tbMsg [{}]", tbMsg.getType(), tbMsg);
553   - }
554   - }
555   -
556   - private void processAssignedEntity(TenantId tenantId, TbMsg tbMsg, EdgeQueueEntityType entityType, FutureCallback<Void> callback) throws IOException {
557   - EdgeId edgeId;
558   - switch (tbMsg.getType()) {
559   - case DataConstants.ENTITY_ASSIGNED_TO_EDGE:
560   - edgeId = new EdgeId(UUID.fromString(tbMsg.getMetaData().getValue("assignedEdgeId")));
561   - pushEventToEdge(tenantId, edgeId, entityType, tbMsg, callback);
562   - break;
563   - case DataConstants.ENTITY_UNASSIGNED_FROM_EDGE:
564   - edgeId = new EdgeId(UUID.fromString(tbMsg.getMetaData().getValue("unassignedEdgeId")));
565   - pushEventToEdge(tenantId, edgeId, entityType, tbMsg, callback);
566   - break;
567   - case DataConstants.ENTITY_DELETED:
568   - case DataConstants.ENTITY_CREATED:
569   - case DataConstants.ENTITY_UPDATED:
570   - Dashboard dashboard = mapper.readValue(tbMsg.getData(), Dashboard.class);
571   - ListenableFuture<TimePageData<Edge>> future = findEdgesByTenantIdAndDashboardId(tenantId, dashboard.getId(), new TimePageLink(Integer.MAX_VALUE));
572   - Futures.transform(future, edges -> {
573   - if (edges != null && edges.getData() != null && !edges.getData().isEmpty()) {
574   - try {
575   - for (Edge edge : edges.getData()) {
576   - pushEventToEdge(tenantId, edge.getId(), EdgeQueueEntityType.DASHBOARD, tbMsg, callback);
577   - }
578   - } catch (IOException e) {
579   - log.error("Can't push event to edge", e);
580   - }
581   - }
582   - return null;
583   - }, MoreExecutors.directExecutor());
584   - break;
585   - }
586   - }
587   -
588   - private void pushEventToEdge(TenantId tenantId, EdgeId edgeId, EdgeQueueEntityType entityType, TbMsg tbMsg, FutureCallback<Void> callback) throws IOException {
589   - log.debug("Pushing event(s) to edge queue. tenantId [{}], edgeId [{}], entityType [{}], tbMsg [{}]", tenantId, edgeId, entityType, tbMsg);
590   -
591   - saveEventToEdgeQueue(tenantId, edgeId, entityType, tbMsg.getType(), tbMsg.getData(), callback);
592   -
593   - if (entityType.equals(EdgeQueueEntityType.RULE_CHAIN)) {
594   - pushRuleChainMetadataToEdge(tenantId, edgeId, tbMsg, callback);
595   - }
596   - }
597   -
598   - private void saveEventToEdgeQueue(TenantId tenantId, EdgeId edgeId, EdgeQueueEntityType entityType, String type, String data, FutureCallback<Void> callback) throws IOException {
599   - log.debug("Pushing single event to edge queue. tenantId [{}], edgeId [{}], entityType [{}], type[{}], data [{}]", tenantId, edgeId, entityType, type, data);
600   -
601   - EdgeQueueEntry queueEntry = new EdgeQueueEntry();
602   - queueEntry.setEntityType(entityType);
603   - queueEntry.setType(type);
604   - queueEntry.setData(data);
605   -
606   - Event event = new Event();
607   - event.setEntityId(edgeId);
608   - event.setTenantId(tenantId);
609   - event.setType(DataConstants.EDGE_QUEUE_EVENT_TYPE);
610   - event.setBody(mapper.valueToTree(queueEntry));
611   - ListenableFuture<Event> saveFuture = eventService.saveAsync(event);
612   -
613   - addMainCallback(saveFuture, callback);
614   - }
615   -
616   - private void addMainCallback(ListenableFuture<Event> saveFuture, final FutureCallback<Void> callback) {
617   - Futures.addCallback(saveFuture, new FutureCallback<Event>() {
618   - @Override
619   - public void onSuccess(@Nullable Event result) {
620   - callback.onSuccess(null);
621   - }
622   -
623   - @Override
624   - public void onFailure(Throwable t) {
625   - callback.onFailure(t);
626   - }
627   - }, tsCallBackExecutor);
628   - }
629   -
630   - private void pushRuleChainMetadataToEdge(TenantId tenantId, EdgeId edgeId, TbMsg tbMsg, FutureCallback<Void> callback) throws IOException {
631   - RuleChain ruleChain = mapper.readValue(tbMsg.getData(), RuleChain.class);
632   - switch (tbMsg.getType()) {
633   - case DataConstants.ENTITY_ASSIGNED_TO_EDGE:
634   - case DataConstants.ENTITY_UNASSIGNED_FROM_EDGE:
635   - case DataConstants.ENTITY_UPDATED:
636   - RuleChainMetaData ruleChainMetaData = ruleChainService.loadRuleChainMetaData(tenantId, ruleChain.getId());
637   - saveEventToEdgeQueue(tenantId, edgeId, EdgeQueueEntityType.RULE_CHAIN_METADATA, tbMsg.getType(), mapper.writeValueAsString(ruleChainMetaData), callback);
638   - break;
639   - default:
640   - log.warn("Unsupported msgType [{}], tbMsg [{}]", tbMsg.getType(), tbMsg);
641   - }
642   - }
643   -
644   - @Override
645   - public TimePageData<Event> findQueueEvents(TenantId tenantId, EdgeId edgeId, TimePageLink pageLink) {
646   - return eventService.findEvents(tenantId, edgeId, DataConstants.EDGE_QUEUE_EVENT_TYPE, pageLink);
647   - }
648   -
649   - @Override
650   - public Edge setEdgeRootRuleChain(TenantId tenantId, Edge edge, RuleChainId ruleChainId) throws IOException {
651   - edge.setRootRuleChainId(ruleChainId);
652   - Edge savedEdge = saveEdge(edge);
653   - RuleChain ruleChain = ruleChainService.findRuleChainById(tenantId, ruleChainId);
654   - saveEventToEdgeQueue(tenantId, edge.getId(), EdgeQueueEntityType.RULE_CHAIN, DataConstants.ENTITY_UPDATED, mapper.writeValueAsString(ruleChain), new FutureCallback<Void>() {
655   - @Override
656   - public void onSuccess(@Nullable Void aVoid) {
657   - log.debug("Event saved successfully!");
658   - }
659   -
660   - @Override
661   - public void onFailure(Throwable t) {
662   - log.debug("Failure during event save", t);
663   - }
664   - });
665   - return savedEdge;
666   - }
667   -
668   - @Override
669 292 public void assignDefaultRuleChainsToEdge(TenantId tenantId, EdgeId edgeId) {
670 293 log.trace("Executing assignDefaultRuleChainsToEdge, tenantId [{}], edgeId [{}]", tenantId, edgeId);
671 294 ListenableFuture<List<RuleChain>> future = ruleChainService.findDefaultEdgeRuleChainsByTenantId(tenantId);
... ... @@ -713,7 +336,6 @@ public class EdgeServiceImpl extends AbstractEntityService implements EdgeServic
713 336 }, MoreExecutors.directExecutor());
714 337 }
715 338
716   -
717 339 private DataValidator<Edge> edgeValidator =
718 340 new DataValidator<Edge>() {
719 341
... ...