diff --git a/application/src/main/java/org/thingsboard/server/actors/device/DeviceActorMessageProcessor.java b/application/src/main/java/org/thingsboard/server/actors/device/DeviceActorMessageProcessor.java index 95125b331a..0b28aca7b0 100644 --- a/application/src/main/java/org/thingsboard/server/actors/device/DeviceActorMessageProcessor.java +++ b/application/src/main/java/org/thingsboard/server/actors/device/DeviceActorMessageProcessor.java @@ -35,6 +35,7 @@ import org.thingsboard.server.actors.TbActorCtx; import org.thingsboard.server.actors.shared.AbstractContextAwareMsgProcessor; 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.StringUtils; import org.thingsboard.server.common.data.edge.EdgeEvent; import org.thingsboard.server.common.data.edge.EdgeEventActionType; @@ -97,7 +98,6 @@ import javax.annotation.Nullable; import java.util.ArrayList; import java.util.Arrays; import java.util.Collections; -import java.util.ConcurrentModificationException; import java.util.HashMap; import java.util.HashSet; import java.util.LinkedHashMap; @@ -107,6 +107,7 @@ import java.util.Objects; import java.util.Optional; import java.util.Set; import java.util.UUID; +import java.util.concurrent.ExecutionException; import java.util.function.Consumer; import java.util.stream.Collectors; @@ -200,8 +201,13 @@ class DeviceActorMessageProcessor extends AbstractContextAwareMsgProcessor { boolean sent = false; if (systemContext.isEdgesEnabled() && 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; + try { + saveRpcRequestToEdgeQueue(request, rpcRequest.getRequestId()).get(); + sent = true; + } catch (InterruptedException | ExecutionException e) { + String errMsg = String.format("[%s][%s][%s] Failed to save rpc request to edge queue %s", tenantId, deviceId, edgeId.getId(), request); + log.error(errMsg, e); + } } else if (isSendNewRpcAvailable()) { sent = rpcSubscriptions.size() > 0; Set syncSessionSet = new HashSet<>(); @@ -810,13 +816,7 @@ class DeviceActorMessageProcessor extends AbstractContextAwareMsgProcessor { systemContext.getTbCoreToTransportService().process(nodeId, msg); } - private void saveRpcRequestToEdgeQueue(ToDeviceRpcRequest msg, Integer requestId) { - EdgeEvent edgeEvent = new EdgeEvent(); - edgeEvent.setTenantId(tenantId); - edgeEvent.setAction(EdgeEventActionType.RPC_CALL); - edgeEvent.setEntityId(deviceId.getId()); - edgeEvent.setType(EdgeEventType.DEVICE); - + private ListenableFuture saveRpcRequestToEdgeQueue(ToDeviceRpcRequest msg, Integer requestId) { ObjectNode body = mapper.createObjectNode(); body.put("requestId", requestId); body.put("requestUUID", msg.getId().toString()); @@ -824,11 +824,13 @@ class DeviceActorMessageProcessor extends AbstractContextAwareMsgProcessor { body.put("expirationTime", msg.getExpirationTime()); body.put("method", msg.getBody().getMethod()); body.put("params", msg.getBody().getParams()); - edgeEvent.setBody(body); - edgeEvent.setEdgeId(edgeId); - systemContext.getEdgeEventService().save(edgeEvent); - systemContext.getClusterService().onEdgeEventUpdate(tenantId, edgeId); + EdgeEvent edgeEvent = EdgeUtils.constructEdgeEvent(tenantId, edgeId, EdgeEventType.DEVICE, EdgeEventActionType.RPC_CALL, deviceId, body); + + return Futures.transform(systemContext.getEdgeEventService().saveAsync(edgeEvent), unused -> { + systemContext.getClusterService().onEdgeEventUpdate(tenantId, edgeId); + return null; + }, systemContext.getDbCallbackExecutor()); } private List toTsKvProtos(@Nullable List result) { diff --git a/application/src/main/java/org/thingsboard/server/controller/EdgeController.java b/application/src/main/java/org/thingsboard/server/controller/EdgeController.java index f3d73f13e5..a516a1e4a3 100644 --- a/application/src/main/java/org/thingsboard/server/controller/EdgeController.java +++ b/application/src/main/java/org/thingsboard/server/controller/EdgeController.java @@ -180,7 +180,7 @@ public class EdgeController extends BaseController { } } - private void onEdgeCreatedOrUpdated(TenantId tenantId, Edge edge, RuleChain edgeTemplateRootRuleChain, boolean updated, SecurityUser user) throws IOException, ThingsboardException { + private void onEdgeCreatedOrUpdated(TenantId tenantId, Edge edge, RuleChain edgeTemplateRootRuleChain, boolean updated, SecurityUser user) throws Exception { if (!updated) { ruleChainService.assignRuleChainToEdge(tenantId, edgeTemplateRootRuleChain.getId(), edge.getId()); edgeNotificationService.setEdgeRootRuleChain(tenantId, edge, edgeTemplateRootRuleChain.getId()); diff --git a/application/src/main/java/org/thingsboard/server/service/edge/DefaultEdgeNotificationService.java b/application/src/main/java/org/thingsboard/server/service/edge/DefaultEdgeNotificationService.java index 55779f19ef..6b5e519f80 100644 --- a/application/src/main/java/org/thingsboard/server/service/edge/DefaultEdgeNotificationService.java +++ b/application/src/main/java/org/thingsboard/server/service/edge/DefaultEdgeNotificationService.java @@ -16,11 +16,16 @@ package org.thingsboard.server.service.edge; import com.fasterxml.jackson.databind.JsonNode; +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.springframework.stereotype.Service; import org.thingsboard.common.util.ThingsBoardThreadFactory; import org.thingsboard.server.cluster.TbClusterService; +import org.thingsboard.server.common.data.EdgeUtils; import org.thingsboard.server.common.data.edge.Edge; import org.thingsboard.server.common.data.edge.EdgeEvent; import org.thingsboard.server.common.data.edge.EdgeEventActionType; @@ -42,7 +47,6 @@ import org.thingsboard.server.service.edge.rpc.processor.RelationEdgeProcessor; import javax.annotation.PostConstruct; import javax.annotation.PreDestroy; -import java.io.IOException; import java.util.UUID; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; @@ -76,29 +80,29 @@ public class DefaultEdgeNotificationService implements EdgeNotificationService { @Autowired private CustomerEdgeProcessor customerProcessor; - private ExecutorService tsCallBackExecutor; + private ExecutorService dbCallBackExecutor; @PostConstruct public void initExecutor() { - tsCallBackExecutor = Executors.newSingleThreadExecutor(ThingsBoardThreadFactory.forName("edge-notifications")); + dbCallBackExecutor = Executors.newSingleThreadExecutor(ThingsBoardThreadFactory.forName("edge-notifications")); } @PreDestroy public void shutdownExecutor() { - if (tsCallBackExecutor != null) { - tsCallBackExecutor.shutdownNow(); + if (dbCallBackExecutor != null) { + dbCallBackExecutor.shutdownNow(); } } @Override - public Edge setEdgeRootRuleChain(TenantId tenantId, Edge edge, RuleChainId ruleChainId) throws IOException { + public Edge setEdgeRootRuleChain(TenantId tenantId, Edge edge, RuleChainId ruleChainId) throws Exception { edge.setRootRuleChainId(ruleChainId); Edge savedEdge = edgeService.saveEdge(edge); - saveEdgeEvent(tenantId, edge.getId(), EdgeEventType.RULE_CHAIN, EdgeEventActionType.UPDATED, ruleChainId, null); + saveEdgeEvent(tenantId, edge.getId(), EdgeEventType.RULE_CHAIN, EdgeEventActionType.UPDATED, ruleChainId, null).get(); return savedEdge; } - private void saveEdgeEvent(TenantId tenantId, + private ListenableFuture saveEdgeEvent(TenantId tenantId, EdgeId edgeId, EdgeEventType type, EdgeEventActionType action, @@ -107,17 +111,12 @@ public class DefaultEdgeNotificationService implements EdgeNotificationService { log.debug("Pushing edge event to edge queue. tenantId [{}], edgeId [{}], type [{}], action[{}], entityId [{}], body [{}]", tenantId, edgeId, type, action, entityId, body); - EdgeEvent edgeEvent = new EdgeEvent(); - edgeEvent.setEdgeId(edgeId); - edgeEvent.setTenantId(tenantId); - edgeEvent.setType(type); - edgeEvent.setAction(action); - if (entityId != null) { - edgeEvent.setEntityId(entityId.getId()); - } - edgeEvent.setBody(body); - edgeEventService.save(edgeEvent); - clusterService.onEdgeEventUpdate(tenantId, edgeId); + EdgeEvent edgeEvent = EdgeUtils.constructEdgeEvent(tenantId, edgeId, type, action, entityId, body); + + return Futures.transform(edgeEventService.saveAsync(edgeEvent), unused -> { + clusterService.onEdgeEventUpdate(tenantId, edgeId); + return null; + }, dbCallBackExecutor); } @Override @@ -126,9 +125,10 @@ public class DefaultEdgeNotificationService implements EdgeNotificationService { try { TenantId tenantId = TenantId.fromUUID(new UUID(edgeNotificationMsg.getTenantIdMSB(), edgeNotificationMsg.getTenantIdLSB())); EdgeEventType type = EdgeEventType.valueOf(edgeNotificationMsg.getType()); + ListenableFuture future; switch (type) { case EDGE: - edgeProcessor.processEdgeNotification(tenantId, edgeNotificationMsg); + future = edgeProcessor.processEdgeNotification(tenantId, edgeNotificationMsg); break; case USER: case ASSET: @@ -137,33 +137,47 @@ public class DefaultEdgeNotificationService implements EdgeNotificationService { case ENTITY_VIEW: case DASHBOARD: case RULE_CHAIN: - entityProcessor.processEntityNotification(tenantId, edgeNotificationMsg); + future = entityProcessor.processEntityNotification(tenantId, edgeNotificationMsg); break; case CUSTOMER: - customerProcessor.processCustomerNotification(tenantId, edgeNotificationMsg); + future = customerProcessor.processCustomerNotification(tenantId, edgeNotificationMsg); break; case WIDGETS_BUNDLE: case WIDGET_TYPE: - entityProcessor.processEntityNotificationForAllEdges(tenantId, edgeNotificationMsg); + future = entityProcessor.processEntityNotificationForAllEdges(tenantId, edgeNotificationMsg); break; case ALARM: - alarmProcessor.processAlarmNotification(tenantId, edgeNotificationMsg); + future = alarmProcessor.processAlarmNotification(tenantId, edgeNotificationMsg); break; case RELATION: - relationProcessor.processRelationNotification(tenantId, edgeNotificationMsg); + future = relationProcessor.processRelationNotification(tenantId, edgeNotificationMsg); break; default: log.warn("Edge event type [{}] is not designed to be pushed to edge", type); + future = Futures.immediateFuture(null); } + Futures.addCallback(future, new FutureCallback<>() { + @Override + public void onSuccess(@Nullable Void unused) { + callback.onSuccess(); + } + + @Override + public void onFailure(Throwable throwable) { + callBackFailure(edgeNotificationMsg, callback, throwable); + } + }, dbCallBackExecutor); } catch (Exception e) { - callback.onFailure(e); - String errMsg = String.format("Can't push to edge updates, edgeNotificationMsg [%s]", edgeNotificationMsg); - log.error(errMsg, e); - } finally { - callback.onSuccess(); + callBackFailure(edgeNotificationMsg, callback, e); } } + private void callBackFailure(TransportProtos.EdgeNotificationMsgProto edgeNotificationMsg, TbCallback callback, Throwable throwable) { + String errMsg = String.format("Can't push to edge updates, edgeNotificationMsg [%s]", edgeNotificationMsg); + log.error(errMsg, throwable); + callback.onFailure(throwable); + } + } diff --git a/application/src/main/java/org/thingsboard/server/service/edge/EdgeNotificationService.java b/application/src/main/java/org/thingsboard/server/service/edge/EdgeNotificationService.java index be6741cef6..ee36c80454 100644 --- a/application/src/main/java/org/thingsboard/server/service/edge/EdgeNotificationService.java +++ b/application/src/main/java/org/thingsboard/server/service/edge/EdgeNotificationService.java @@ -21,11 +21,9 @@ import org.thingsboard.server.common.data.id.TenantId; import org.thingsboard.server.common.msg.queue.TbCallback; import org.thingsboard.server.gen.transport.TransportProtos; -import java.io.IOException; - public interface EdgeNotificationService { - Edge setEdgeRootRuleChain(TenantId tenantId, Edge edge, RuleChainId ruleChainId) throws IOException; + Edge setEdgeRootRuleChain(TenantId tenantId, Edge edge, RuleChainId ruleChainId) throws Exception; void pushNotificationToEdge(TransportProtos.EdgeNotificationMsgProto edgeNotificationMsg, TbCallback callback); } diff --git a/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/AlarmEdgeProcessor.java b/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/AlarmEdgeProcessor.java index ccd3f2362c..a63e377a5b 100644 --- a/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/AlarmEdgeProcessor.java +++ b/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/AlarmEdgeProcessor.java @@ -16,11 +16,9 @@ package org.thingsboard.server.service.edge.rpc.processor; import com.fasterxml.jackson.core.JsonProcessingException; -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.stereotype.Component; import org.thingsboard.server.common.data.EdgeUtils; import org.thingsboard.server.common.data.EntityType; @@ -43,6 +41,8 @@ import org.thingsboard.server.gen.edge.v1.UpdateMsgType; import org.thingsboard.server.gen.transport.TransportProtos; import org.thingsboard.server.queue.util.TbCoreComponent; +import java.util.ArrayList; +import java.util.List; import java.util.UUID; @Component @@ -148,49 +148,44 @@ public class AlarmEdgeProcessor extends BaseEdgeProcessor { return downlinkMsg; } - public void processAlarmNotification(TenantId tenantId, TransportProtos.EdgeNotificationMsgProto edgeNotificationMsg) throws JsonProcessingException { + public ListenableFuture processAlarmNotification(TenantId tenantId, TransportProtos.EdgeNotificationMsgProto edgeNotificationMsg) throws JsonProcessingException { EdgeEventActionType actionType = EdgeEventActionType.valueOf(edgeNotificationMsg.getAction()); AlarmId alarmId = new AlarmId(new UUID(edgeNotificationMsg.getEntityIdMSB(), edgeNotificationMsg.getEntityIdLSB())); switch (actionType) { case DELETED: EdgeId edgeId = new EdgeId(new UUID(edgeNotificationMsg.getEdgeIdMSB(), edgeNotificationMsg.getEdgeIdLSB())); - Alarm alarm = mapper.readValue(edgeNotificationMsg.getBody(), Alarm.class); - saveEdgeEvent(tenantId, edgeId, EdgeEventType.ALARM, actionType, alarmId, mapper.valueToTree(alarm)); - break; + Alarm deletedAlarm = mapper.readValue(edgeNotificationMsg.getBody(), Alarm.class); + return saveEdgeEvent(tenantId, edgeId, EdgeEventType.ALARM, actionType, alarmId, mapper.valueToTree(deletedAlarm)); default: ListenableFuture alarmFuture = alarmService.findAlarmByIdAsync(tenantId, alarmId); - Futures.addCallback(alarmFuture, new FutureCallback() { - @Override - public void onSuccess(@Nullable Alarm alarm) { - if (alarm != null) { - EdgeEventType type = EdgeUtils.getEdgeEventTypeByEntityType(alarm.getOriginator().getEntityType()); - if (type != null) { - PageLink pageLink = new PageLink(DEFAULT_PAGE_SIZE); - PageData pageData; - do { - pageData = edgeService.findRelatedEdgeIdsByEntityId(tenantId, alarm.getOriginator(), pageLink); - if (pageData != null && pageData.getData() != null && !pageData.getData().isEmpty()) { - for (EdgeId edgeId : pageData.getData()) { - saveEdgeEvent(tenantId, - edgeId, - EdgeEventType.ALARM, - EdgeEventActionType.valueOf(edgeNotificationMsg.getAction()), - alarmId, - null); - } - if (pageData.hasNext()) { - pageLink = pageLink.nextPageLink(); - } - } - } while (pageData != null && pageData.hasNext()); - } - } + return Futures.transformAsync(alarmFuture, alarm -> { + if (alarm == null) { + return Futures.immediateFuture(null); } - - @Override - public void onFailure(Throwable t) { - log.warn("[{}] can't find alarm by id [{}] {}", tenantId.getId(), alarmId.getId(), t); + EdgeEventType type = EdgeUtils.getEdgeEventTypeByEntityType(alarm.getOriginator().getEntityType()); + if (type == null) { + return Futures.immediateFuture(null); } + PageLink pageLink = new PageLink(DEFAULT_PAGE_SIZE); + PageData pageData; + List> futures = new ArrayList<>(); + do { + pageData = edgeService.findRelatedEdgeIdsByEntityId(tenantId, alarm.getOriginator(), pageLink); + if (pageData != null && pageData.getData() != null && !pageData.getData().isEmpty()) { + for (EdgeId relatedEdgeId : pageData.getData()) { + futures.add(saveEdgeEvent(tenantId, + relatedEdgeId, + EdgeEventType.ALARM, + EdgeEventActionType.valueOf(edgeNotificationMsg.getAction()), + alarmId, + null)); + } + if (pageData.hasNext()) { + pageLink = pageLink.nextPageLink(); + } + } + } while (pageData != null && pageData.hasNext()); + return Futures.transform(Futures.allAsList(futures), voids -> null, dbCallbackExecutorService); }, dbCallbackExecutorService); } } diff --git a/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/BaseEdgeProcessor.java b/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/BaseEdgeProcessor.java index 10d1ae88c5..09bef32af9 100644 --- a/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/BaseEdgeProcessor.java +++ b/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/BaseEdgeProcessor.java @@ -17,10 +17,13 @@ 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.Futures; +import com.google.common.util.concurrent.ListenableFuture; import lombok.extern.slf4j.Slf4j; import org.springframework.beans.factory.annotation.Autowired; import org.thingsboard.server.cluster.TbClusterService; import org.thingsboard.server.common.data.Device; +import org.thingsboard.server.common.data.EdgeUtils; import org.thingsboard.server.common.data.HasCustomerId; import org.thingsboard.server.common.data.edge.Edge; import org.thingsboard.server.common.data.edge.EdgeEvent; @@ -67,6 +70,9 @@ import org.thingsboard.server.service.executors.DbCallbackExecutorService; import org.thingsboard.server.service.profile.TbDeviceProfileCache; import org.thingsboard.server.service.state.DeviceStateService; +import java.util.ArrayList; +import java.util.List; + @Slf4j public abstract class BaseEdgeProcessor { @@ -179,27 +185,22 @@ public abstract class BaseEdgeProcessor { @Autowired protected DbCallbackExecutorService dbCallbackExecutorService; - protected void saveEdgeEvent(TenantId tenantId, - EdgeId edgeId, - EdgeEventType type, - EdgeEventActionType action, - EntityId entityId, - JsonNode body) { + protected ListenableFuture saveEdgeEvent(TenantId tenantId, + EdgeId edgeId, + EdgeEventType type, + EdgeEventActionType action, + EntityId entityId, + JsonNode body) { log.debug("Pushing event to edge queue. tenantId [{}], edgeId [{}], type[{}], " + "action [{}], entityId [{}], body [{}]", tenantId, edgeId, type, action, entityId, body); - EdgeEvent edgeEvent = new EdgeEvent(); - edgeEvent.setTenantId(tenantId); - edgeEvent.setEdgeId(edgeId); - edgeEvent.setType(type); - edgeEvent.setAction(action); - if (entityId != null) { - edgeEvent.setEntityId(entityId.getId()); - } - edgeEvent.setBody(body); - edgeEventService.save(edgeEvent); - tbClusterService.onEdgeEventUpdate(tenantId, edgeId); + EdgeEvent edgeEvent = EdgeUtils.constructEdgeEvent(tenantId, edgeId, type, action, entityId, body); + + return Futures.transform(edgeEventService.saveAsync(edgeEvent), unused -> { + tbClusterService.onEdgeEventUpdate(tenantId, edgeId); + return null; + }, dbCallbackExecutorService); } protected CustomerId getCustomerIdIfEdgeAssignedToCustomer(HasCustomerId hasCustomerIdEntity, Edge edge) { @@ -210,19 +211,21 @@ public abstract class BaseEdgeProcessor { } } - protected void processActionForAllEdges(TenantId tenantId, EdgeEventType type, EdgeEventActionType actionType, EntityId entityId) { + protected ListenableFuture processActionForAllEdges(TenantId tenantId, EdgeEventType type, EdgeEventActionType actionType, EntityId entityId) { PageLink pageLink = new PageLink(DEFAULT_PAGE_SIZE); PageData pageData; + List> futures = new ArrayList<>(); 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); + futures.add(saveEdgeEvent(tenantId, edge.getId(), type, actionType, entityId, null)); } if (pageData.hasNext()) { pageLink = pageLink.nextPageLink(); } } } while (pageData != null && pageData.hasNext()); + return Futures.transform(Futures.allAsList(futures), voids -> null, dbCallbackExecutorService); } } diff --git a/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/CustomerEdgeProcessor.java b/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/CustomerEdgeProcessor.java index 6c670e47ce..3554057482 100644 --- a/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/CustomerEdgeProcessor.java +++ b/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/CustomerEdgeProcessor.java @@ -15,6 +15,8 @@ */ package org.thingsboard.server.service.edge.rpc.processor; +import com.google.common.util.concurrent.Futures; +import com.google.common.util.concurrent.ListenableFuture; import lombok.extern.slf4j.Slf4j; import org.springframework.stereotype.Component; import org.thingsboard.server.common.data.Customer; @@ -35,6 +37,8 @@ import org.thingsboard.server.gen.edge.v1.UpdateMsgType; import org.thingsboard.server.gen.transport.TransportProtos; import org.thingsboard.server.queue.util.TbCoreComponent; +import java.util.ArrayList; +import java.util.List; import java.util.UUID; @Component @@ -70,7 +74,7 @@ public class CustomerEdgeProcessor extends BaseEdgeProcessor { return downlinkMsg; } - public void processCustomerNotification(TenantId tenantId, TransportProtos.EdgeNotificationMsgProto edgeNotificationMsg) { + public ListenableFuture processCustomerNotification(TenantId tenantId, TransportProtos.EdgeNotificationMsgProto edgeNotificationMsg) { EdgeEventActionType actionType = EdgeEventActionType.valueOf(edgeNotificationMsg.getAction()); EdgeEventType type = EdgeEventType.valueOf(edgeNotificationMsg.getType()); UUID uuid = new UUID(edgeNotificationMsg.getEntityIdMSB(), edgeNotificationMsg.getEntityIdLSB()); @@ -79,22 +83,24 @@ public class CustomerEdgeProcessor extends BaseEdgeProcessor { case UPDATED: PageLink pageLink = new PageLink(DEFAULT_PAGE_SIZE); PageData pageData; + List> futures = new ArrayList<>(); do { pageData = edgeService.findEdgesByTenantIdAndCustomerId(tenantId, customerId, pageLink); if (pageData != null && pageData.getData() != null && !pageData.getData().isEmpty()) { for (Edge edge : pageData.getData()) { - saveEdgeEvent(tenantId, edge.getId(), type, actionType, customerId, null); + futures.add(saveEdgeEvent(tenantId, edge.getId(), type, actionType, customerId, null)); } if (pageData.hasNext()) { pageLink = pageLink.nextPageLink(); } } } while (pageData != null && pageData.hasNext()); - break; + return Futures.transform(Futures.allAsList(futures), voids -> null, dbCallbackExecutorService); case DELETED: EdgeId edgeId = new EdgeId(new UUID(edgeNotificationMsg.getEdgeIdMSB(), edgeNotificationMsg.getEdgeIdLSB())); - saveEdgeEvent(tenantId, edgeId, type, actionType, customerId, null); - break; + return saveEdgeEvent(tenantId, edgeId, type, actionType, customerId, null); + default: + return Futures.immediateFuture(null); } } diff --git a/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/DeviceEdgeProcessor.java b/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/DeviceEdgeProcessor.java index f19ec25b27..0e1dee7132 100644 --- a/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/DeviceEdgeProcessor.java +++ b/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/DeviceEdgeProcessor.java @@ -84,7 +84,7 @@ public class DeviceEdgeProcessor extends BaseEdgeProcessor { if (deviceAlreadyExistsForThisEdge) { log.info("[{}] Device with name '{}' already exists on the cloud, and related to this edge [{}]. " + "deviceUpdateMsg [{}], Updating device", tenantId, deviceName, edge.getId(), deviceUpdateMsg); - updateDevice(tenantId, edge, deviceUpdateMsg); + return updateDevice(tenantId, edge, deviceUpdateMsg); } else { log.info("[{}] Device with name '{}' already exists on the cloud, but not related to this edge [{}]. deviceUpdateMsg [{}]." + "Creating a new device with random prefix and relate to this edge", tenantId, deviceName, edge.getId(), deviceUpdateMsg); @@ -99,8 +99,10 @@ public class DeviceEdgeProcessor extends BaseEdgeProcessor { } ObjectNode body = mapper.createObjectNode(); body.put("conflictName", deviceName); - saveEdgeEvent(tenantId, edge.getId(), EdgeEventType.DEVICE, EdgeEventActionType.ENTITY_MERGE_REQUEST, newDevice.getId(), body); - saveEdgeEvent(tenantId, edge.getId(), EdgeEventType.DEVICE, EdgeEventActionType.CREDENTIALS_REQUEST, newDevice.getId(), null); + ListenableFuture input = saveEdgeEvent(tenantId, edge.getId(), EdgeEventType.DEVICE, EdgeEventActionType.ENTITY_MERGE_REQUEST, newDevice.getId(), body); + return Futures.transformAsync(input, unused -> + saveEdgeEvent(tenantId, edge.getId(), EdgeEventType.DEVICE, EdgeEventActionType.CREDENTIALS_REQUEST, newDevice.getId(), null), + dbCallbackExecutorService); } } else { log.info("[{}] Creating new device and replacing device entity on the edge [{}]", tenantId, deviceUpdateMsg); @@ -111,24 +113,22 @@ public class DeviceEdgeProcessor extends BaseEdgeProcessor { log.error(errMsg, e); return Futures.immediateFuture(null); } - saveEdgeEvent(tenantId, edge.getId(), EdgeEventType.DEVICE, EdgeEventActionType.CREDENTIALS_REQUEST, device.getId(), null); + return saveEdgeEvent(tenantId, edge.getId(), EdgeEventType.DEVICE, EdgeEventActionType.CREDENTIALS_REQUEST, device.getId(), null); } - break; case ENTITY_UPDATED_RPC_MESSAGE: - updateDevice(tenantId, edge, deviceUpdateMsg); - break; + return updateDevice(tenantId, edge, deviceUpdateMsg); case ENTITY_DELETED_RPC_MESSAGE: DeviceId deviceId = new DeviceId(new UUID(deviceUpdateMsg.getIdMSB(), deviceUpdateMsg.getIdLSB())); Device deviceToDelete = deviceService.findDeviceById(tenantId, deviceId); if (deviceToDelete != null) { deviceService.unassignDeviceFromEdge(tenantId, deviceId, edge.getId()); } - break; + return Futures.immediateFuture(null); case UNRECOGNIZED: + default: log.error("Unsupported msg type {}", deviceUpdateMsg.getMsgType()); return Futures.immediateFailedFuture(new RuntimeException("Unsupported msg type " + deviceUpdateMsg.getMsgType())); } - return Futures.immediateFuture(null); } private boolean isDeviceAlreadyExistsOnCloudForThisEdge(TenantId tenantId, Edge edge, Device device) { @@ -174,7 +174,7 @@ public class DeviceEdgeProcessor extends BaseEdgeProcessor { } - private void updateDevice(TenantId tenantId, Edge edge, DeviceUpdateMsg deviceUpdateMsg) { + private ListenableFuture updateDevice(TenantId tenantId, Edge edge, DeviceUpdateMsg deviceUpdateMsg) { DeviceId deviceId = new DeviceId(new UUID(deviceUpdateMsg.getIdMSB(), deviceUpdateMsg.getIdLSB())); Device device = deviceService.findDeviceById(tenantId, deviceId); if (device != null) { @@ -194,9 +194,11 @@ public class DeviceEdgeProcessor extends BaseEdgeProcessor { } Device savedDevice = deviceService.saveDevice(device); tbClusterService.onDeviceUpdated(savedDevice, device); - saveEdgeEvent(tenantId, edge.getId(), EdgeEventType.DEVICE, EdgeEventActionType.CREDENTIALS_REQUEST, deviceId, null); + return saveEdgeEvent(tenantId, edge.getId(), EdgeEventType.DEVICE, EdgeEventActionType.CREDENTIALS_REQUEST, deviceId, null); } else { - log.warn("[{}] can't find device [{}], edge [{}]", tenantId, deviceUpdateMsg, edge.getId()); + String errMsg = String.format("[%s] can't find device [%s], edge [%s]", tenantId, deviceUpdateMsg, edge.getId()); + log.warn(errMsg); + return Futures.immediateFailedFuture(new RuntimeException(errMsg)); } } diff --git a/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/EdgeProcessor.java b/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/EdgeProcessor.java index 3d662da232..fdbebafc98 100644 --- a/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/EdgeProcessor.java +++ b/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/EdgeProcessor.java @@ -15,11 +15,9 @@ */ package org.thingsboard.server.service.edge.rpc.processor; -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.stereotype.Component; import org.thingsboard.server.common.data.User; import org.thingsboard.server.common.data.edge.Edge; @@ -33,6 +31,8 @@ import org.thingsboard.server.common.data.page.PageLink; import org.thingsboard.server.gen.transport.TransportProtos; import org.thingsboard.server.queue.util.TbCoreComponent; +import java.util.ArrayList; +import java.util.List; import java.util.UUID; @Component @@ -40,7 +40,7 @@ import java.util.UUID; @TbCoreComponent public class EdgeProcessor extends BaseEdgeProcessor { - public void processEdgeNotification(TenantId tenantId, TransportProtos.EdgeNotificationMsgProto edgeNotificationMsg) { + public ListenableFuture processEdgeNotification(TenantId tenantId, TransportProtos.EdgeNotificationMsgProto edgeNotificationMsg) { try { EdgeEventActionType actionType = EdgeEventActionType.valueOf(edgeNotificationMsg.getAction()); EdgeId edgeId = new EdgeId(new UUID(edgeNotificationMsg.getEntityIdMSB(), edgeNotificationMsg.getEntityIdLSB())); @@ -49,54 +49,43 @@ public class EdgeProcessor extends BaseEdgeProcessor { case ASSIGNED_TO_CUSTOMER: CustomerId customerId = mapper.readValue(edgeNotificationMsg.getBody(), CustomerId.class); edgeFuture = edgeService.findEdgeByIdAsync(tenantId, edgeId); - Futures.addCallback(edgeFuture, new FutureCallback() { - @Override - public void onSuccess(@Nullable Edge edge) { - if (edge != null && !customerId.isNullUid()) { - saveEdgeEvent(edge.getTenantId(), edge.getId(), EdgeEventType.CUSTOMER, EdgeEventActionType.ADDED, customerId, null); - PageLink pageLink = new PageLink(DEFAULT_PAGE_SIZE); - PageData 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()); - } - } - - @Override - public void onFailure(Throwable t) { - log.error("Can't find edge by id [{}]", edgeNotificationMsg, t); + return Futures.transformAsync(edgeFuture, edge -> { + if (edge == null || customerId.isNullUid()) { + return Futures.immediateFuture(null); } + List> futures = new ArrayList<>(); + futures.add(saveEdgeEvent(edge.getTenantId(), edge.getId(), EdgeEventType.CUSTOMER, EdgeEventActionType.ADDED, customerId, null)); + PageLink pageLink = new PageLink(DEFAULT_PAGE_SIZE); + PageData 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()) { + futures.add(saveEdgeEvent(edge.getTenantId(), edge.getId(), EdgeEventType.USER, EdgeEventActionType.ADDED, user.getId(), null)); + } + if (pageData.hasNext()) { + pageLink = pageLink.nextPageLink(); + } + } + } while (pageData != null && pageData.hasNext()); + return Futures.transform(Futures.allAsList(futures), voids -> null, dbCallbackExecutorService); }, dbCallbackExecutorService); - break; case UNASSIGNED_FROM_CUSTOMER: CustomerId customerIdToDelete = mapper.readValue(edgeNotificationMsg.getBody(), CustomerId.class); edgeFuture = edgeService.findEdgeByIdAsync(tenantId, edgeId); - Futures.addCallback(edgeFuture, new FutureCallback() { - @Override - public void onSuccess(@Nullable Edge edge) { - if (edge != null && !customerIdToDelete.isNullUid()) { - saveEdgeEvent(edge.getTenantId(), edge.getId(), EdgeEventType.CUSTOMER, EdgeEventActionType.DELETED, customerIdToDelete, null); - } - } - - @Override - public void onFailure(Throwable t) { - log.error("Can't find edge by id [{}]", edgeNotificationMsg, t); + return Futures.transformAsync(edgeFuture, edge -> { + if (edge == null || customerIdToDelete.isNullUid()) { + return Futures.immediateFuture(null); } + return saveEdgeEvent(edge.getTenantId(), edge.getId(), EdgeEventType.CUSTOMER, EdgeEventActionType.DELETED, customerIdToDelete, null); }, dbCallbackExecutorService); - break; + default: + return Futures.immediateFuture(null); } } catch (Exception e) { log.error("Exception during processing edge event", e); + return Futures.immediateFailedFuture(e); } } } diff --git a/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/EntityEdgeProcessor.java b/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/EntityEdgeProcessor.java index 89ce662467..195f932b93 100644 --- a/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/EntityEdgeProcessor.java +++ b/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/EntityEdgeProcessor.java @@ -15,11 +15,9 @@ */ package org.thingsboard.server.service.edge.rpc.processor; -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.stereotype.Component; import org.thingsboard.server.common.data.Device; import org.thingsboard.server.common.data.EdgeUtils; @@ -45,6 +43,7 @@ import org.thingsboard.server.gen.edge.v1.UpdateMsgType; import org.thingsboard.server.gen.transport.TransportProtos; import org.thingsboard.server.queue.util.TbCoreComponent; +import java.util.ArrayList; import java.util.List; import java.util.UUID; @@ -89,92 +88,107 @@ public class EntityEdgeProcessor extends BaseEdgeProcessor { return downlinkMsg; } - public void processEntityNotification(TenantId tenantId, TransportProtos.EdgeNotificationMsgProto edgeNotificationMsg) { + public ListenableFuture processEntityNotification(TenantId tenantId, TransportProtos.EdgeNotificationMsgProto edgeNotificationMsg) { EdgeEventActionType actionType = EdgeEventActionType.valueOf(edgeNotificationMsg.getAction()); EdgeEventType type = EdgeEventType.valueOf(edgeNotificationMsg.getType()); EntityId entityId = EntityIdFactory.getByEdgeEventTypeAndUuid(type, new UUID(edgeNotificationMsg.getEntityIdMSB(), edgeNotificationMsg.getEntityIdLSB())); - EdgeId edgeId = null; - if (edgeNotificationMsg.getEdgeIdMSB() != 0 && edgeNotificationMsg.getEdgeIdLSB() != 0) { - edgeId = new EdgeId(new UUID(edgeNotificationMsg.getEdgeIdMSB(), edgeNotificationMsg.getEdgeIdLSB())); - } + EdgeId edgeId = safeGetEdgeId(edgeNotificationMsg); switch (actionType) { case ADDED: // used only for USER entity case UPDATED: case CREDENTIALS_UPDATED: - pushNotificationToAllRelatedEdges(tenantId, entityId, type, actionType); - break; + return pushNotificationToAllRelatedEdges(tenantId, entityId, type, actionType); case ASSIGNED_TO_CUSTOMER: case UNASSIGNED_FROM_CUSTOMER: - PageLink pageLink = new PageLink(DEFAULT_PAGE_SIZE); - PageData pageData; - do { - pageData = edgeService.findRelatedEdgeIdsByEntityId(tenantId, entityId, pageLink); - if (pageData != null && pageData.getData() != null && !pageData.getData().isEmpty()) { - for (EdgeId relatedEdgeId : pageData.getData()) { - try { - CustomerId customerId = mapper.readValue(edgeNotificationMsg.getBody(), CustomerId.class); - ListenableFuture future = edgeService.findEdgeByIdAsync(tenantId, relatedEdgeId); - Futures.addCallback(future, new FutureCallback<>() { - @Override - public void onSuccess(@Nullable Edge edge) { - if (edge != null && edge.getCustomerId() != null && - !edge.getCustomerId().isNullUid() && edge.getCustomerId().equals(customerId)) { - saveEdgeEvent(tenantId, relatedEdgeId, type, actionType, entityId, null); - } - } - - @Override - public void onFailure(Throwable t) { - log.error("Failed to find edge by id [{}] {}", edgeNotificationMsg, t); - } - }, dbCallbackExecutorService); - } catch (Exception e) { - log.error("Can't parse customer id from entity body [{}]", edgeNotificationMsg, e); - } - } - if (pageData.hasNext()) { - pageLink = pageLink.nextPageLink(); - } - } - } while (pageData != null && pageData.hasNext()); - break; + return pushNotificationToAllRelatedCustomerEdges(tenantId, edgeNotificationMsg, entityId, actionType, type); case DELETED: if (edgeId != null) { - saveEdgeEvent(tenantId, edgeId, type, actionType, entityId, null); + return saveEdgeEvent(tenantId, edgeId, type, actionType, entityId, null); } else { - pushNotificationToAllRelatedEdges(tenantId, entityId, type, actionType); + return pushNotificationToAllRelatedEdges(tenantId, entityId, type, actionType); } - break; case ASSIGNED_TO_EDGE: case UNASSIGNED_FROM_EDGE: - saveEdgeEvent(tenantId, edgeId, type, actionType, entityId, null); - if (type.equals(EdgeEventType.RULE_CHAIN)) { - updateDependentRuleChains(tenantId, new RuleChainId(entityId.getId()), edgeId); + ListenableFuture future = saveEdgeEvent(tenantId, edgeId, type, actionType, entityId, null); + return Futures.transformAsync(future, unused -> { + if (type.equals(EdgeEventType.RULE_CHAIN)) { + return updateDependentRuleChains(tenantId, new RuleChainId(entityId.getId()), edgeId); + } else { + return Futures.immediateFuture(null); + } + }, dbCallbackExecutorService); + default: + return Futures.immediateFuture(null); + } + } + + private ListenableFuture pushNotificationToAllRelatedCustomerEdges(TenantId tenantId, + TransportProtos.EdgeNotificationMsgProto edgeNotificationMsg, + EntityId entityId, + EdgeEventActionType actionType, + EdgeEventType type) { + PageLink pageLink = new PageLink(DEFAULT_PAGE_SIZE); + PageData pageData; + List> futures = new ArrayList<>(); + do { + pageData = edgeService.findRelatedEdgeIdsByEntityId(tenantId, entityId, pageLink); + if (pageData != null && pageData.getData() != null && !pageData.getData().isEmpty()) { + for (EdgeId relatedEdgeId : pageData.getData()) { + try { + CustomerId customerId = mapper.readValue(edgeNotificationMsg.getBody(), CustomerId.class); + ListenableFuture future = edgeService.findEdgeByIdAsync(tenantId, relatedEdgeId); + futures.add(Futures.transformAsync(future, edge -> { + if (edge != null && edge.getCustomerId() != null && + !edge.getCustomerId().isNullUid() && edge.getCustomerId().equals(customerId)) { + return saveEdgeEvent(tenantId, relatedEdgeId, type, actionType, entityId, null); + } else { + return Futures.immediateFuture(null); + } + }, dbCallbackExecutorService)); + } catch (Exception e) { + log.error("Can't parse customer id from entity body [{}]", edgeNotificationMsg, e); + return Futures.immediateFailedFuture(e); + } } - break; + if (pageData.hasNext()) { + pageLink = pageLink.nextPageLink(); + } + } + } while (pageData != null && pageData.hasNext()); + return Futures.transform(Futures.allAsList(futures), voids -> null, dbCallbackExecutorService); + } + + private EdgeId safeGetEdgeId(TransportProtos.EdgeNotificationMsgProto edgeNotificationMsg) { + if (edgeNotificationMsg.getEdgeIdMSB() != 0 && edgeNotificationMsg.getEdgeIdLSB() != 0) { + return new EdgeId(new UUID(edgeNotificationMsg.getEdgeIdMSB(), edgeNotificationMsg.getEdgeIdLSB())); + } else { + return null; } } - private void pushNotificationToAllRelatedEdges(TenantId tenantId, EntityId entityId, EdgeEventType type, EdgeEventActionType actionType) { + private ListenableFuture pushNotificationToAllRelatedEdges(TenantId tenantId, EntityId entityId, EdgeEventType type, EdgeEventActionType actionType) { PageLink pageLink = new PageLink(DEFAULT_PAGE_SIZE); PageData pageData; + List> futures = new ArrayList<>(); do { pageData = edgeService.findRelatedEdgeIdsByEntityId(tenantId, entityId, pageLink); if (pageData != null && pageData.getData() != null && !pageData.getData().isEmpty()) { for (EdgeId relatedEdgeId : pageData.getData()) { - saveEdgeEvent(tenantId, relatedEdgeId, type, actionType, entityId, null); + futures.add(saveEdgeEvent(tenantId, relatedEdgeId, type, actionType, entityId, null)); } if (pageData.hasNext()) { pageLink = pageLink.nextPageLink(); } } } while (pageData != null && pageData.hasNext()); + return Futures.transform(Futures.allAsList(futures), voids -> null, dbCallbackExecutorService); } - private void updateDependentRuleChains(TenantId tenantId, RuleChainId processingRuleChainId, EdgeId edgeId) { + private ListenableFuture updateDependentRuleChains(TenantId tenantId, RuleChainId processingRuleChainId, EdgeId edgeId) { PageLink pageLink = new PageLink(DEFAULT_PAGE_SIZE); PageData pageData; + List> futures = new ArrayList<>(); do { pageData = ruleChainService.findRuleChainsByTenantIdAndEdgeId(tenantId, edgeId, pageLink); if (pageData != null && pageData.getData() != null && !pageData.getData().isEmpty()) { @@ -185,12 +199,12 @@ public class EntityEdgeProcessor extends BaseEdgeProcessor { if (connectionInfos != null && !connectionInfos.isEmpty()) { for (RuleChainConnectionInfo connectionInfo : connectionInfos) { if (connectionInfo.getTargetRuleChainId().equals(processingRuleChainId)) { - saveEdgeEvent(tenantId, + futures.add(saveEdgeEvent(tenantId, edgeId, EdgeEventType.RULE_CHAIN_METADATA, EdgeEventActionType.UPDATED, ruleChain.getId(), - null); + null)); } } } @@ -201,9 +215,10 @@ public class EntityEdgeProcessor extends BaseEdgeProcessor { } } } while (pageData != null && pageData.hasNext()); + return Futures.transform(Futures.allAsList(futures), voids -> null, dbCallbackExecutorService); } - public void processEntityNotificationForAllEdges(TenantId tenantId, TransportProtos.EdgeNotificationMsgProto edgeNotificationMsg) { + public ListenableFuture processEntityNotificationForAllEdges(TenantId tenantId, TransportProtos.EdgeNotificationMsgProto edgeNotificationMsg) { EdgeEventActionType actionType = EdgeEventActionType.valueOf(edgeNotificationMsg.getAction()); EdgeEventType type = EdgeEventType.valueOf(edgeNotificationMsg.getType()); EntityId entityId = EntityIdFactory.getByEdgeEventTypeAndUuid(type, new UUID(edgeNotificationMsg.getEntityIdMSB(), edgeNotificationMsg.getEntityIdLSB())); @@ -211,8 +226,9 @@ public class EntityEdgeProcessor extends BaseEdgeProcessor { case ADDED: case UPDATED: case DELETED: - processActionForAllEdges(tenantId, type, actionType, entityId); - break; + return processActionForAllEdges(tenantId, type, actionType, entityId); + default: + return Futures.immediateFuture(null); } } } diff --git a/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/RelationEdgeProcessor.java b/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/RelationEdgeProcessor.java index d49ac7094c..f86ea8aad2 100644 --- a/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/RelationEdgeProcessor.java +++ b/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/RelationEdgeProcessor.java @@ -126,24 +126,29 @@ public class RelationEdgeProcessor extends BaseEdgeProcessor { .build(); } - public void processRelationNotification(TenantId tenantId, TransportProtos.EdgeNotificationMsgProto edgeNotificationMsg) throws JsonProcessingException { + public ListenableFuture processRelationNotification(TenantId tenantId, TransportProtos.EdgeNotificationMsgProto edgeNotificationMsg) throws JsonProcessingException { EntityRelation relation = mapper.readValue(edgeNotificationMsg.getBody(), EntityRelation.class); - if (!relation.getFrom().getEntityType().equals(EntityType.EDGE) && - !relation.getTo().getEntityType().equals(EntityType.EDGE)) { - Set uniqueEdgeIds = new HashSet<>(); - uniqueEdgeIds.addAll(findRelatedEdgeIds(tenantId, relation.getTo())); - uniqueEdgeIds.addAll(findRelatedEdgeIds(tenantId, relation.getFrom())); - if (!uniqueEdgeIds.isEmpty()) { - for (EdgeId edgeId : uniqueEdgeIds) { - saveEdgeEvent(tenantId, - edgeId, - EdgeEventType.RELATION, - EdgeEventActionType.valueOf(edgeNotificationMsg.getAction()), - null, - mapper.valueToTree(relation)); - } - } + if (relation.getFrom().getEntityType().equals(EntityType.EDGE) || + relation.getTo().getEntityType().equals(EntityType.EDGE)) { + return Futures.immediateFuture(null); + } + + Set uniqueEdgeIds = new HashSet<>(); + uniqueEdgeIds.addAll(findRelatedEdgeIds(tenantId, relation.getTo())); + uniqueEdgeIds.addAll(findRelatedEdgeIds(tenantId, relation.getFrom())); + if (uniqueEdgeIds.isEmpty()) { + return Futures.immediateFuture(null); + } + List> futures = new ArrayList<>(); + for (EdgeId edgeId : uniqueEdgeIds) { + futures.add(saveEdgeEvent(tenantId, + edgeId, + EdgeEventType.RELATION, + EdgeEventActionType.valueOf(edgeNotificationMsg.getAction()), + null, + mapper.valueToTree(relation))); } + return Futures.transform(Futures.allAsList(futures), voids -> null, dbCallbackExecutorService); } private List findRelatedEdgeIds(TenantId tenantId, EntityId entityId) { diff --git a/application/src/main/java/org/thingsboard/server/service/edge/rpc/sync/DefaultEdgeRequestsService.java b/application/src/main/java/org/thingsboard/server/service/edge/rpc/sync/DefaultEdgeRequestsService.java index 43344f5406..9d87cabb2c 100644 --- a/application/src/main/java/org/thingsboard/server/service/edge/rpc/sync/DefaultEdgeRequestsService.java +++ b/application/src/main/java/org/thingsboard/server/service/edge/rpc/sync/DefaultEdgeRequestsService.java @@ -26,6 +26,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.cluster.TbClusterService; import org.thingsboard.server.common.data.Device; import org.thingsboard.server.common.data.DeviceProfile; import org.thingsboard.server.common.data.EdgeUtils; @@ -72,7 +73,6 @@ import org.thingsboard.server.gen.edge.v1.RuleChainMetadataRequestMsg; import org.thingsboard.server.gen.edge.v1.UserCredentialsRequestMsg; import org.thingsboard.server.gen.edge.v1.WidgetBundleTypesRequestMsg; import org.thingsboard.server.service.executors.DbCallbackExecutorService; -import org.thingsboard.server.cluster.TbClusterService; import java.util.ArrayList; import java.util.HashMap; @@ -121,13 +121,13 @@ public class DefaultEdgeRequestsService implements EdgeRequestsService { @Override public ListenableFuture processRuleChainMetadataRequestMsg(TenantId tenantId, Edge edge, RuleChainMetadataRequestMsg ruleChainMetadataRequestMsg) { log.trace("[{}] processRuleChainMetadataRequestMsg [{}][{}]", tenantId, edge.getName(), ruleChainMetadataRequestMsg); - if (ruleChainMetadataRequestMsg.getRuleChainIdMSB() != 0 && ruleChainMetadataRequestMsg.getRuleChainIdLSB() != 0) { - RuleChainId ruleChainId = - new RuleChainId(new UUID(ruleChainMetadataRequestMsg.getRuleChainIdMSB(), ruleChainMetadataRequestMsg.getRuleChainIdLSB())); - saveEdgeEvent(tenantId, edge.getId(), - EdgeEventType.RULE_CHAIN_METADATA, EdgeEventActionType.ADDED, ruleChainId, null); + if (ruleChainMetadataRequestMsg.getRuleChainIdMSB() == 0 || ruleChainMetadataRequestMsg.getRuleChainIdLSB() == 0) { + return Futures.immediateFuture(null); } - return Futures.immediateFuture(null); + RuleChainId ruleChainId = + new RuleChainId(new UUID(ruleChainMetadataRequestMsg.getRuleChainIdMSB(), ruleChainMetadataRequestMsg.getRuleChainIdLSB())); + return saveEdgeEvent(tenantId, edge.getId(), + EdgeEventType.RULE_CHAIN_METADATA, EdgeEventActionType.ADDED, ruleChainId, null); } @Override @@ -137,63 +137,72 @@ public class DefaultEdgeRequestsService implements EdgeRequestsService { EntityType.valueOf(attributesRequestMsg.getEntityType()), new UUID(attributesRequestMsg.getEntityIdMSB(), attributesRequestMsg.getEntityIdLSB())); final EdgeEventType type = EdgeUtils.getEdgeEventTypeByEntityType(entityId.getEntityType()); - if (type != null) { - SettableFuture futureToSet = SettableFuture.create(); - String scope = attributesRequestMsg.getScope(); - ListenableFuture> findAttrFuture = attributesService.findAll(tenantId, entityId, scope); - Futures.addCallback(findAttrFuture, new FutureCallback>() { - @Override - public void onSuccess(@Nullable List ssAttributes) { - if (ssAttributes != null && !ssAttributes.isEmpty()) { - try { - Map entityData = new HashMap<>(); - ObjectNode attributes = mapper.createObjectNode(); - for (AttributeKvEntry attr : ssAttributes) { - if (attr.getDataType() == DataType.BOOLEAN && attr.getBooleanValue().isPresent()) { - attributes.put(attr.getKey(), attr.getBooleanValue().get()); - } else if (attr.getDataType() == DataType.DOUBLE && attr.getDoubleValue().isPresent()) { - attributes.put(attr.getKey(), attr.getDoubleValue().get()); - } else if (attr.getDataType() == DataType.LONG && attr.getLongValue().isPresent()) { - attributes.put(attr.getKey(), attr.getLongValue().get()); - } else { - attributes.put(attr.getKey(), attr.getValueAsString()); - } - } - entityData.put("kv", attributes); - entityData.put("scope", scope); - JsonNode body = mapper.valueToTree(entityData); - log.debug("Sending attributes data msg, entityId [{}], attributes [{}]", entityId, body); - saveEdgeEvent(tenantId, - edge.getId(), - type, - EdgeEventActionType.ATTRIBUTES_UPDATED, - entityId, - body); - } catch (Exception e) { - log.error("[{}] Failed to save attribute updates to the edge", edge.getName(), e); - futureToSet.setException(new RuntimeException("[" + edge.getName() + "] Failed to send attribute updates to the edge", e)); - return; - } - } else { - log.trace("[{}][{}] No attributes found for entity {} [{}]", tenantId, - edge.getName(), - entityId.getEntityType(), - entityId.getId()); - } + if (type == null) { + log.warn("[{}] Type doesn't supported {}", tenantId, entityId.getEntityType()); + return Futures.immediateFuture(null); + } + SettableFuture futureToSet = SettableFuture.create(); + String scope = attributesRequestMsg.getScope(); + ListenableFuture> findAttrFuture = attributesService.findAll(tenantId, entityId, scope); + Futures.addCallback(findAttrFuture, new FutureCallback<>() { + @Override + public void onSuccess(@Nullable List ssAttributes) { + if (ssAttributes == null || ssAttributes.isEmpty()) { + log.trace("[{}][{}] No attributes found for entity {} [{}]", tenantId, + edge.getName(), + entityId.getEntityType(), + entityId.getId()); futureToSet.set(null); + return; } - @Override - public void onFailure(Throwable t) { - log.error("Can't find attributes [{}]", attributesRequestMsg, t); - futureToSet.setException(t); + try { + Map entityData = new HashMap<>(); + ObjectNode attributes = mapper.createObjectNode(); + for (AttributeKvEntry attr : ssAttributes) { + if (attr.getDataType() == DataType.BOOLEAN && attr.getBooleanValue().isPresent()) { + attributes.put(attr.getKey(), attr.getBooleanValue().get()); + } else if (attr.getDataType() == DataType.DOUBLE && attr.getDoubleValue().isPresent()) { + attributes.put(attr.getKey(), attr.getDoubleValue().get()); + } else if (attr.getDataType() == DataType.LONG && attr.getLongValue().isPresent()) { + attributes.put(attr.getKey(), attr.getLongValue().get()); + } else { + attributes.put(attr.getKey(), attr.getValueAsString()); + } + } + entityData.put("kv", attributes); + entityData.put("scope", scope); + JsonNode body = mapper.valueToTree(entityData); + log.debug("Sending attributes data msg, entityId [{}], attributes [{}]", entityId, body); + ListenableFuture future = saveEdgeEvent(tenantId, edge.getId(), type, EdgeEventActionType.ATTRIBUTES_UPDATED, entityId, body); + Futures.addCallback(future, new FutureCallback<>() { + @Override + public void onSuccess(@Nullable Void unused) { + futureToSet.set(null); + } + + @Override + public void onFailure(Throwable throwable) { + String errMsg = String.format("[%s] Failed to save edge event [%s]", edge.getId(), attributesRequestMsg); + log.error(errMsg, throwable); + futureToSet.setException(new RuntimeException(errMsg, throwable)); + } + }, dbCallbackExecutorService); + } catch (Exception e) { + String errMsg = String.format("[%s] Failed to save attribute updates to the edge [%s]", edge.getId(), attributesRequestMsg); + log.error(errMsg, e); + futureToSet.setException(new RuntimeException(errMsg, e)); } - }, dbCallbackExecutorService); - return futureToSet; - } else { - log.warn("[{}] Type doesn't supported {}", tenantId, entityId.getEntityType()); - return Futures.immediateFuture(null); - } + } + + @Override + public void onFailure(Throwable t) { + String errMsg = String.format("[%s] Can't find attributes [%s]", edge.getId(), attributesRequestMsg); + log.error(errMsg, t); + futureToSet.setException(new RuntimeException(errMsg, t)); + } + }, dbCallbackExecutorService); + return futureToSet; } @Override @@ -208,33 +217,49 @@ public class DefaultEdgeRequestsService implements EdgeRequestsService { futures.add(findRelationByQuery(tenantId, edge, entityId, EntitySearchDirection.TO)); ListenableFuture>> relationsListFuture = Futures.allAsList(futures); SettableFuture futureToSet = SettableFuture.create(); - Futures.addCallback(relationsListFuture, new FutureCallback>>() { + Futures.addCallback(relationsListFuture, new FutureCallback<>() { @Override public void onSuccess(@Nullable List> relationsList) { try { if (relationsList != null && !relationsList.isEmpty()) { + List> futures = new ArrayList<>(); for (List 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(tenantId, + futures.add(saveEdgeEvent(tenantId, edge.getId(), EdgeEventType.RELATION, EdgeEventActionType.ADDED, null, - mapper.valueToTree(relation)); + mapper.valueToTree(relation))); } } catch (Exception e) { - log.error("Exception during loading relation [{}] to edge on sync!", relation, e); - futureToSet.setException(e); + String errMsg = String.format("[%s] Exception during loading relation [%s] to edge on sync!", edge.getId(), relation); + log.error(errMsg, e); + futureToSet.setException(new RuntimeException(errMsg, e)); return; } } } + Futures.addCallback(Futures.allAsList(futures), new FutureCallback<>() { + @Override + public void onSuccess(@Nullable List voids) { + futureToSet.set(null); + } + + @Override + public void onFailure(Throwable throwable) { + String errMsg = String.format("[%s] Exception during saving edge events [%s]!", edge.getId(), relationRequestMsg); + log.error(errMsg, throwable); + futureToSet.setException(new RuntimeException(errMsg, throwable)); + } + }, dbCallbackExecutorService); + } else { + futureToSet.set(null); } - futureToSet.set(null); } catch (Exception e) { log.error("Exception during loading relation(s) to edge on sync!", e); futureToSet.setException(e); @@ -243,8 +268,9 @@ public class DefaultEdgeRequestsService implements EdgeRequestsService { @Override public void onFailure(Throwable t) { - log.error("[{}] Can't find relation by query. Entity id [{}]", tenantId, entityId, t); - futureToSet.setException(t); + String errMsg = String.format("[%s] Can't find relation by query. Entity id [%s]!", tenantId, entityId); + log.error(errMsg, t); + futureToSet.setException(new RuntimeException(errMsg, t)); } }, dbCallbackExecutorService); return futureToSet; @@ -260,40 +286,42 @@ public class DefaultEdgeRequestsService implements EdgeRequestsService { @Override public ListenableFuture processDeviceCredentialsRequestMsg(TenantId tenantId, Edge edge, DeviceCredentialsRequestMsg deviceCredentialsRequestMsg) { log.trace("[{}] processDeviceCredentialsRequestMsg [{}][{}]", tenantId, edge.getName(), deviceCredentialsRequestMsg); - if (deviceCredentialsRequestMsg.getDeviceIdMSB() != 0 && deviceCredentialsRequestMsg.getDeviceIdLSB() != 0) { - DeviceId deviceId = new DeviceId(new UUID(deviceCredentialsRequestMsg.getDeviceIdMSB(), deviceCredentialsRequestMsg.getDeviceIdLSB())); - saveEdgeEvent(tenantId, edge.getId(), EdgeEventType.DEVICE, - EdgeEventActionType.CREDENTIALS_UPDATED, deviceId, null); + if (deviceCredentialsRequestMsg.getDeviceIdMSB() == 0 || deviceCredentialsRequestMsg.getDeviceIdLSB() == 0) { + return Futures.immediateFuture(null); } - return Futures.immediateFuture(null); + DeviceId deviceId = new DeviceId(new UUID(deviceCredentialsRequestMsg.getDeviceIdMSB(), deviceCredentialsRequestMsg.getDeviceIdLSB())); + return saveEdgeEvent(tenantId, edge.getId(), EdgeEventType.DEVICE, + EdgeEventActionType.CREDENTIALS_UPDATED, deviceId, null); } @Override public ListenableFuture processUserCredentialsRequestMsg(TenantId tenantId, Edge edge, UserCredentialsRequestMsg userCredentialsRequestMsg) { log.trace("[{}] processUserCredentialsRequestMsg [{}][{}]", tenantId, edge.getName(), userCredentialsRequestMsg); - if (userCredentialsRequestMsg.getUserIdMSB() != 0 && userCredentialsRequestMsg.getUserIdLSB() != 0) { - UserId userId = new UserId(new UUID(userCredentialsRequestMsg.getUserIdMSB(), userCredentialsRequestMsg.getUserIdLSB())); - saveEdgeEvent(tenantId, edge.getId(), EdgeEventType.USER, - EdgeEventActionType.CREDENTIALS_UPDATED, userId, null); + if (userCredentialsRequestMsg.getUserIdMSB() == 0 || userCredentialsRequestMsg.getUserIdLSB() == 0) { + return Futures.immediateFuture(null); } - return Futures.immediateFuture(null); + UserId userId = new UserId(new UUID(userCredentialsRequestMsg.getUserIdMSB(), userCredentialsRequestMsg.getUserIdLSB())); + return saveEdgeEvent(tenantId, edge.getId(), EdgeEventType.USER, + EdgeEventActionType.CREDENTIALS_UPDATED, userId, null); } @Override public ListenableFuture processDeviceProfileDevicesRequestMsg(TenantId tenantId, Edge edge, DeviceProfileDevicesRequestMsg deviceProfileDevicesRequestMsg) { log.trace("[{}] processDeviceProfileDevicesRequestMsg [{}][{}]", tenantId, edge.getName(), deviceProfileDevicesRequestMsg); - if (deviceProfileDevicesRequestMsg.getDeviceProfileIdMSB() != 0 && deviceProfileDevicesRequestMsg.getDeviceProfileIdLSB() != 0) { - DeviceProfileId deviceProfileId = new DeviceProfileId(new UUID(deviceProfileDevicesRequestMsg.getDeviceProfileIdMSB(), deviceProfileDevicesRequestMsg.getDeviceProfileIdLSB())); - DeviceProfile deviceProfileById = deviceProfileService.findDeviceProfileById(tenantId, deviceProfileId); - if (deviceProfileById != null) { - syncDevices(tenantId, edge, deviceProfileById.getName()); - } + if (deviceProfileDevicesRequestMsg.getDeviceProfileIdMSB() == 0 || deviceProfileDevicesRequestMsg.getDeviceProfileIdLSB() == 0) { + return Futures.immediateFuture(null); + } + DeviceProfileId deviceProfileId = new DeviceProfileId(new UUID(deviceProfileDevicesRequestMsg.getDeviceProfileIdMSB(), deviceProfileDevicesRequestMsg.getDeviceProfileIdLSB())); + DeviceProfile deviceProfileById = deviceProfileService.findDeviceProfileById(tenantId, deviceProfileId); + if (deviceProfileById == null) { + return Futures.immediateFuture(null); } - return Futures.immediateFuture(null); + return syncDevices(tenantId, edge, deviceProfileById.getName()); } - private void syncDevices(TenantId tenantId, Edge edge, String deviceType) { + private ListenableFuture syncDevices(TenantId tenantId, Edge edge, String deviceType) { log.trace("[{}] syncDevices [{}][{}]", tenantId, edge.getName(), deviceType); + List> futures = new ArrayList<>(); try { PageLink pageLink = new PageLink(DEFAULT_PAGE_SIZE); PageData pageData; @@ -302,7 +330,7 @@ public class DefaultEdgeRequestsService implements EdgeRequestsService { 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(tenantId, edge.getId(), EdgeEventType.DEVICE, EdgeEventActionType.ADDED, device.getId(), null); + futures.add(saveEdgeEvent(tenantId, edge.getId(), EdgeEventType.DEVICE, EdgeEventActionType.ADDED, device.getId(), null)); } if (pageData.hasNext()) { pageLink = pageLink.nextPageLink(); @@ -312,25 +340,26 @@ public class DefaultEdgeRequestsService implements EdgeRequestsService { } catch (Exception e) { log.error("Exception during loading edge device(s) on sync!", e); } + return Futures.transform(Futures.allAsList(futures), voids -> null, dbCallbackExecutorService); } @Override public ListenableFuture processWidgetBundleTypesRequestMsg(TenantId tenantId, Edge edge, WidgetBundleTypesRequestMsg widgetBundleTypesRequestMsg) { log.trace("[{}] processWidgetBundleTypesRequestMsg [{}][{}]", tenantId, edge.getName(), widgetBundleTypesRequestMsg); + List> futures = new ArrayList<>(); if (widgetBundleTypesRequestMsg.getWidgetBundleIdMSB() != 0 && widgetBundleTypesRequestMsg.getWidgetBundleIdLSB() != 0) { WidgetsBundleId widgetsBundleId = new WidgetsBundleId(new UUID(widgetBundleTypesRequestMsg.getWidgetBundleIdMSB(), widgetBundleTypesRequestMsg.getWidgetBundleIdLSB())); WidgetsBundle widgetsBundleById = widgetsBundleService.findWidgetsBundleById(tenantId, widgetsBundleId); if (widgetsBundleById != null) { List widgetTypesToPush = widgetTypeService.findWidgetTypesByTenantIdAndBundleAlias(widgetsBundleById.getTenantId(), widgetsBundleById.getAlias()); - for (WidgetType widgetType : widgetTypesToPush) { - saveEdgeEvent(tenantId, edge.getId(), EdgeEventType.WIDGET_TYPE, EdgeEventActionType.ADDED, widgetType.getId(), null); + futures.add(saveEdgeEvent(tenantId, edge.getId(), EdgeEventType.WIDGET_TYPE, EdgeEventActionType.ADDED, widgetType.getId(), null)); } } } - return Futures.immediateFuture(null); + return Futures.transform(Futures.allAsList(futures), voids -> null, dbCallbackExecutorService); } @Override @@ -343,46 +372,35 @@ public class DefaultEdgeRequestsService implements EdgeRequestsService { Futures.addCallback(entityViewService.findEntityViewsByTenantIdAndEntityIdAsync(tenantId, entityId), new FutureCallback<>() { @Override public void onSuccess(@Nullable List entityViews) { - try { - if (entityViews != null && !entityViews.isEmpty()) { - List> futures = new ArrayList<>(); - for (EntityView entityView : entityViews) { - ListenableFuture future = relationService.checkRelation(tenantId, edge.getId(), entityView.getId(), - EntityRelation.CONTAINS_TYPE, RelationTypeGroup.EDGE); - futures.add(future); - Futures.addCallback(future, new FutureCallback<>() { - @Override - public void onSuccess(@Nullable Boolean result) { - if (Boolean.TRUE.equals(result)) { - saveEdgeEvent(tenantId, edge.getId(), EdgeEventType.ENTITY_VIEW, - EdgeEventActionType.ADDED, entityView.getId(), null); - } - } - @Override - public void onFailure(Throwable t) { - // Do nothing - error handles in allAsList - } - }, dbCallbackExecutorService); + if (entityViews == null || entityViews.isEmpty()) { + futureToSet.set(null); + return; + } + List> futures = new ArrayList<>(); + for (EntityView entityView : entityViews) { + ListenableFuture future = relationService.checkRelation(tenantId, edge.getId(), entityView.getId(), + EntityRelation.CONTAINS_TYPE, RelationTypeGroup.EDGE); + futures.add(Futures.transformAsync(future, result -> { + if (Boolean.TRUE.equals(result)) { + return saveEdgeEvent(tenantId, edge.getId(), EdgeEventType.ENTITY_VIEW, + EdgeEventActionType.ADDED, entityView.getId(), null); + } else { + return Futures.immediateFuture(null); } - Futures.addCallback(Futures.allAsList(futures), new FutureCallback<>() { - @Override - public void onSuccess(@Nullable List result) { - futureToSet.set(null); - } - - @Override - public void onFailure(Throwable t) { - log.error("Exception during loading relation [{}] to edge on sync!", t, t); - futureToSet.setException(t); - } - }, dbCallbackExecutorService); - } else { + }, dbCallbackExecutorService)); + } + Futures.addCallback(Futures.allAsList(futures), new FutureCallback<>() { + @Override + public void onSuccess(@Nullable List result) { futureToSet.set(null); } - } catch (Exception e) { - log.error("Exception during loading relation(s) to edge on sync!", e); - futureToSet.setException(e); - } + + @Override + public void onFailure(Throwable t) { + log.error("Exception during loading relation to edge on sync!", t); + futureToSet.setException(t); + } + }, dbCallbackExecutorService); } @Override @@ -394,7 +412,7 @@ public class DefaultEdgeRequestsService implements EdgeRequestsService { return futureToSet; } - private void saveEdgeEvent(TenantId tenantId, + private ListenableFuture saveEdgeEvent(TenantId tenantId, EdgeId edgeId, EdgeEventType type, EdgeEventActionType action, @@ -405,8 +423,10 @@ public class DefaultEdgeRequestsService implements EdgeRequestsService { EdgeEvent edgeEvent = EdgeUtils.constructEdgeEvent(tenantId, edgeId, type, action, entityId, body); - edgeEventService.save(edgeEvent); - tbClusterService.onEdgeEventUpdate(tenantId, edgeId); + return Futures.transform(edgeEventService.saveAsync(edgeEvent), unused -> { + tbClusterService.onEdgeEventUpdate(tenantId, edgeId); + return null; + }, dbCallbackExecutorService); } } diff --git a/application/src/main/resources/thingsboard.yml b/application/src/main/resources/thingsboard.yml index 4eb82e93dc..a1c7bc765e 100644 --- a/application/src/main/resources/thingsboard.yml +++ b/application/src/main/resources/thingsboard.yml @@ -283,6 +283,11 @@ sql: batch_max_delay: "${SQL_EVENTS_BATCH_MAX_DELAY_MS:100}" stats_print_interval_ms: "${SQL_EVENTS_BATCH_STATS_PRINT_MS:10000}" batch_threads: "${SQL_EVENTS_BATCH_THREADS:3}" # batch thread count have to be a prime number like 3 or 5 to gain perfect hash distribution + edge_events: + batch_size: "${SQL_EDGE_EVENTS_BATCH_SIZE:1000}" + batch_max_delay: "${SQL_EDGE_EVENTS_BATCH_MAX_DELAY_MS:100}" + stats_print_interval_ms: "${SQL_EDGE_EVENTS_BATCH_STATS_PRINT_MS:10000}" + batch_threads: "${SQL_EDGE_EVENTS_BATCH_THREADS:3}" # batch thread count have to be a prime number like 3 or 5 to gain perfect hash distribution # Specify whether to sort entities before batch update. Should be enabled for cluster mode to avoid deadlocks batch_sort: "${SQL_BATCH_SORT:false}" # Specify whether to remove null characters from strValue of attributes and timeseries before insert diff --git a/application/src/test/java/org/thingsboard/server/edge/BaseEdgeTest.java b/application/src/test/java/org/thingsboard/server/edge/BaseEdgeTest.java index b2bb56a989..e815378d27 100644 --- a/application/src/test/java/org/thingsboard/server/edge/BaseEdgeTest.java +++ b/application/src/test/java/org/thingsboard/server/edge/BaseEdgeTest.java @@ -976,7 +976,7 @@ abstract public class BaseEdgeTest extends AbstractControllerTest { String timeseriesData = "{\"data\":{\"temperature\":25},\"ts\":" + System.currentTimeMillis() + "}"; JsonNode timeseriesEntityData = mapper.readTree(timeseriesData); EdgeEvent edgeEvent = constructEdgeEvent(tenantId, edge.getId(), EdgeEventActionType.TIMESERIES_UPDATED, device.getId().getId(), EdgeEventType.DEVICE, timeseriesEntityData); - edgeEventService.save(edgeEvent); + edgeEventService.saveAsync(edgeEvent).get(); clusterService.onEdgeEventUpdate(tenantId, edge.getId()); Assert.assertTrue(edgeImitator.waitForMessages()); @@ -1007,12 +1007,12 @@ abstract public class BaseEdgeTest extends AbstractControllerTest { testAttributesDeleteMsg(device); } - private void testAttributesUpdatedMsg(Device device) throws JsonProcessingException, InterruptedException { + private void testAttributesUpdatedMsg(Device device) throws Exception { String attributesData = "{\"scope\":\"SERVER_SCOPE\",\"kv\":{\"key1\":\"value1\"}}"; JsonNode attributesEntityData = mapper.readTree(attributesData); EdgeEvent edgeEvent1 = constructEdgeEvent(tenantId, edge.getId(), EdgeEventActionType.ATTRIBUTES_UPDATED, device.getId().getId(), EdgeEventType.DEVICE, attributesEntityData); edgeImitator.expectMessageAmount(1); - edgeEventService.save(edgeEvent1); + edgeEventService.saveAsync(edgeEvent1).get(); clusterService.onEdgeEventUpdate(tenantId, edge.getId()); Assert.assertTrue(edgeImitator.waitForMessages()); @@ -1032,12 +1032,12 @@ abstract public class BaseEdgeTest extends AbstractControllerTest { Assert.assertEquals("value1", keyValueProto.getStringV()); } - private void testPostAttributesMsg(Device device) throws JsonProcessingException, InterruptedException { + private void testPostAttributesMsg(Device device) throws Exception { String postAttributesData = "{\"scope\":\"SERVER_SCOPE\",\"kv\":{\"key2\":\"value2\"}}"; JsonNode postAttributesEntityData = mapper.readTree(postAttributesData); EdgeEvent edgeEvent = constructEdgeEvent(tenantId, edge.getId(), EdgeEventActionType.POST_ATTRIBUTES, device.getId().getId(), EdgeEventType.DEVICE, postAttributesEntityData); edgeImitator.expectMessageAmount(1); - edgeEventService.save(edgeEvent); + edgeEventService.saveAsync(edgeEvent).get(); clusterService.onEdgeEventUpdate(tenantId, edge.getId()); Assert.assertTrue(edgeImitator.waitForMessages()); @@ -1057,12 +1057,12 @@ abstract public class BaseEdgeTest extends AbstractControllerTest { Assert.assertEquals("value2", keyValueProto.getStringV()); } - private void testAttributesDeleteMsg(Device device) throws JsonProcessingException, InterruptedException { + private void testAttributesDeleteMsg(Device device) throws Exception { String deleteAttributesData = "{\"scope\":\"SERVER_SCOPE\",\"keys\":[\"key1\",\"key2\"]}"; JsonNode deleteAttributesEntityData = mapper.readTree(deleteAttributesData); EdgeEvent edgeEvent = constructEdgeEvent(tenantId, edge.getId(), EdgeEventActionType.ATTRIBUTES_DELETED, device.getId().getId(), EdgeEventType.DEVICE, deleteAttributesEntityData); edgeImitator.expectMessageAmount(1); - edgeEventService.save(edgeEvent); + edgeEventService.saveAsync(edgeEvent).get(); clusterService.onEdgeEventUpdate(tenantId, edge.getId()); Assert.assertTrue(edgeImitator.waitForMessages()); @@ -1097,7 +1097,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.save(edgeEvent); + edgeEventService.saveAsync(edgeEvent).get(); clusterService.onEdgeEventUpdate(tenantId, edge.getId()); Assert.assertTrue(edgeImitator.waitForMessages()); @@ -1122,7 +1122,7 @@ abstract public class BaseEdgeTest extends AbstractControllerTest { JsonNode timeseriesEntityData = mapper.readTree(timeseriesData); EdgeEvent edgeEvent = constructEdgeEvent(tenantId, edge.getId(), EdgeEventActionType.TIMESERIES_UPDATED, device.getId().getId(), EdgeEventType.DEVICE, timeseriesEntityData); - edgeEventService.save(edgeEvent); + edgeEventService.saveAsync(edgeEvent).get(); clusterService.onEdgeEventUpdate(tenantId, edge.getId()); } diff --git a/common/dao-api/src/main/java/org/thingsboard/server/dao/edge/EdgeEventService.java b/common/dao-api/src/main/java/org/thingsboard/server/dao/edge/EdgeEventService.java index e8ec430bcf..e9950447db 100644 --- a/common/dao-api/src/main/java/org/thingsboard/server/dao/edge/EdgeEventService.java +++ b/common/dao-api/src/main/java/org/thingsboard/server/dao/edge/EdgeEventService.java @@ -15,6 +15,7 @@ */ package org.thingsboard.server.dao.edge; +import com.google.common.util.concurrent.ListenableFuture; import org.thingsboard.server.common.data.edge.EdgeEvent; import org.thingsboard.server.common.data.id.EdgeId; import org.thingsboard.server.common.data.id.TenantId; @@ -23,7 +24,7 @@ import org.thingsboard.server.common.data.page.TimePageLink; public interface EdgeEventService { - EdgeEvent save(EdgeEvent edgeEvent); + ListenableFuture saveAsync(EdgeEvent edgeEvent); PageData findEdgeEvents(TenantId tenantId, EdgeId edgeId, TimePageLink pageLink, boolean withTsUpdate); diff --git a/dao/src/main/java/org/thingsboard/server/dao/edge/BaseEdgeEventService.java b/dao/src/main/java/org/thingsboard/server/dao/edge/BaseEdgeEventService.java index 9d18146ec2..f9e94af613 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/edge/BaseEdgeEventService.java +++ b/dao/src/main/java/org/thingsboard/server/dao/edge/BaseEdgeEventService.java @@ -15,6 +15,7 @@ */ package org.thingsboard.server.dao.edge; +import com.google.common.util.concurrent.ListenableFuture; import lombok.extern.slf4j.Slf4j; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.stereotype.Service; @@ -36,9 +37,9 @@ public class BaseEdgeEventService implements EdgeEventService { private DataValidator edgeEventValidator; @Override - public EdgeEvent save(EdgeEvent edgeEvent) { + public ListenableFuture saveAsync(EdgeEvent edgeEvent) { edgeEventValidator.validate(edgeEvent, EdgeEvent::getTenantId); - return edgeEventDao.save(edgeEvent); + return edgeEventDao.saveAsync(edgeEvent); } @Override diff --git a/dao/src/main/java/org/thingsboard/server/dao/edge/EdgeEventDao.java b/dao/src/main/java/org/thingsboard/server/dao/edge/EdgeEventDao.java index 5ea81eb1f9..7a43d6c065 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/edge/EdgeEventDao.java +++ b/dao/src/main/java/org/thingsboard/server/dao/edge/EdgeEventDao.java @@ -35,7 +35,7 @@ public interface EdgeEventDao extends Dao { * @param edgeEvent the event object * @return saved edge event object future */ - EdgeEvent save(EdgeEvent edgeEvent); + ListenableFuture saveAsync(EdgeEvent edgeEvent); /** diff --git a/dao/src/main/java/org/thingsboard/server/dao/sql/edge/EdgeEventInsertRepository.java b/dao/src/main/java/org/thingsboard/server/dao/sql/edge/EdgeEventInsertRepository.java new file mode 100644 index 0000000000..c67aaabbb1 --- /dev/null +++ b/dao/src/main/java/org/thingsboard/server/dao/sql/edge/EdgeEventInsertRepository.java @@ -0,0 +1,79 @@ +/** + * Copyright © 2016-2022 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.dao.sql.edge; + +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.jdbc.core.BatchPreparedStatementSetter; +import org.springframework.jdbc.core.JdbcTemplate; +import org.springframework.stereotype.Repository; +import org.springframework.transaction.TransactionStatus; +import org.springframework.transaction.annotation.Transactional; +import org.springframework.transaction.support.TransactionCallbackWithoutResult; +import org.springframework.transaction.support.TransactionTemplate; +import org.thingsboard.server.dao.model.sql.EdgeEventEntity; +import org.thingsboard.server.dao.util.PsqlDao; + +import java.sql.PreparedStatement; +import java.sql.SQLException; +import java.util.List; + +@PsqlDao +@Repository +@Transactional +public class EdgeEventInsertRepository { + + private static final String INSERT = + "INSERT INTO edge_event (id, created_time, edge_id, edge_event_type, edge_event_uid, entity_id, edge_event_action, body, tenant_id, ts) " + + "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?) " + + "ON CONFLICT DO NOTHING;"; + + @Autowired + protected JdbcTemplate jdbcTemplate; + + @Autowired + private TransactionTemplate transactionTemplate; + + protected void save(List entities) { + transactionTemplate.execute(new TransactionCallbackWithoutResult() { + @Override + protected void doInTransactionWithoutResult(TransactionStatus status) { + jdbcTemplate.batchUpdate(INSERT, new BatchPreparedStatementSetter() { + @Override + public void setValues(PreparedStatement ps, int i) throws SQLException { + EdgeEventEntity edgeEvent = entities.get(i); + ps.setObject(1, edgeEvent.getId()); + ps.setLong(2, edgeEvent.getCreatedTime()); + ps.setObject(3, edgeEvent.getEdgeId()); + ps.setString(4, edgeEvent.getEdgeEventType().name()); + ps.setString(5, edgeEvent.getEdgeEventUid()); + ps.setObject(6, edgeEvent.getEntityId()); + ps.setString(7, edgeEvent.getEdgeEventAction().name()); + ps.setString(8, edgeEvent.getEntityBody() != null + ? edgeEvent.getEntityBody().toString() + : null); + ps.setObject(9, edgeEvent.getTenantId()); + ps.setLong(10, edgeEvent.getTs()); + } + + @Override + public int getBatchSize() { + return entities.size(); + } + }); + } + }); + } +} diff --git a/dao/src/main/java/org/thingsboard/server/dao/sql/edge/JpaBaseEdgeEventDao.java b/dao/src/main/java/org/thingsboard/server/dao/sql/edge/JpaBaseEdgeEventDao.java index b327ef4edf..5ecb60e744 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/sql/edge/JpaBaseEdgeEventDao.java +++ b/dao/src/main/java/org/thingsboard/server/dao/sql/edge/JpaBaseEdgeEventDao.java @@ -16,9 +16,11 @@ package org.thingsboard.server.dao.sql.edge; import com.datastax.oss.driver.api.core.uuid.Uuids; +import com.google.common.util.concurrent.ListenableFuture; import lombok.extern.slf4j.Slf4j; import org.apache.commons.lang3.StringUtils; import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.beans.factory.annotation.Value; import org.springframework.data.jpa.repository.JpaRepository; import org.springframework.stereotype.Component; import org.thingsboard.server.common.data.edge.EdgeEvent; @@ -26,23 +28,26 @@ import org.thingsboard.server.common.data.id.EdgeEventId; import org.thingsboard.server.common.data.id.EdgeId; import org.thingsboard.server.common.data.page.PageData; import org.thingsboard.server.common.data.page.TimePageLink; +import org.thingsboard.server.common.stats.StatsFactory; import org.thingsboard.server.dao.DaoUtil; import org.thingsboard.server.dao.edge.EdgeEventDao; import org.thingsboard.server.dao.model.sql.EdgeEventEntity; import org.thingsboard.server.dao.sql.JpaAbstractSearchTextDao; +import org.thingsboard.server.dao.sql.ScheduledLogExecutorComponent; +import org.thingsboard.server.dao.sql.TbSqlBlockingQueueParams; +import org.thingsboard.server.dao.sql.TbSqlBlockingQueueWrapper; +import javax.annotation.PostConstruct; +import javax.annotation.PreDestroy; import java.sql.Connection; import java.sql.PreparedStatement; import java.sql.ResultSet; import java.sql.SQLException; +import java.util.Comparator; import java.util.Objects; -import java.util.Optional; import java.util.UUID; -import java.util.concurrent.ConcurrentHashMap; -import java.util.concurrent.ConcurrentMap; import java.util.concurrent.TimeUnit; -import java.util.concurrent.locks.Lock; -import java.util.concurrent.locks.ReentrantLock; +import java.util.function.Function; import static org.thingsboard.server.dao.model.ModelConstants.NULL_UUID; @@ -52,11 +57,32 @@ public class JpaBaseEdgeEventDao extends JpaAbstractSearchTextDao readWriteLocks = new ConcurrentHashMap<>(); + @Autowired + ScheduledLogExecutorComponent logExecutor; + + @Autowired + private StatsFactory statsFactory; + + @Value("${sql.edge_events.batch_size:1000}") + private int batchSize; + + @Value("${sql.edge_events.batch_max_delay:100}") + private long maxDelay; + + @Value("${sql.edge_events.stats_print_interval_ms:10000}") + private long statsPrintIntervalMs; + + @Value("${sql.edge_events.batch_threads:3}") + private int batchThreads; + + private TbSqlBlockingQueueWrapper queue; @Autowired private EdgeEventRepository edgeEventRepository; + @Autowired + private EdgeEventInsertRepository edgeEventInsertRepository; + @Override protected Class getEntityClass() { return EdgeEventEntity.class; @@ -67,66 +93,58 @@ public class JpaBaseEdgeEventDao extends JpaAbstractSearchTextDao new ReentrantLock()); - readWriteLock.lock(); - try { - log.debug("Save edge event [{}] ", edgeEvent); - if (edgeEvent.getId() == null) { - UUID timeBased = Uuids.timeBased(); - edgeEvent.setId(new EdgeEventId(timeBased)); - edgeEvent.setCreatedTime(Uuids.unixTimestamp(timeBased)); - } else if (edgeEvent.getCreatedTime() == 0L) { - UUID eventId = edgeEvent.getId().getId(); - if (eventId.version() == 1) { - edgeEvent.setCreatedTime(Uuids.unixTimestamp(eventId)); - } else { - edgeEvent.setCreatedTime(System.currentTimeMillis()); - } - } - if (StringUtils.isEmpty(edgeEvent.getUid())) { - edgeEvent.setUid(edgeEvent.getId().toString()); + @PostConstruct + private void init() { + TbSqlBlockingQueueParams params = TbSqlBlockingQueueParams.builder() + .logName("Edge Events") + .batchSize(batchSize) + .maxDelay(maxDelay) + .statsPrintIntervalMs(statsPrintIntervalMs) + .statsNamePrefix("edge.events") + .batchSortEnabled(true) + .build(); + Function hashcodeFunction = entity -> { + if (entity.getEntityId() != null) { + return entity.getEntityId().hashCode(); + } else { + return NULL_UUID.hashCode(); } - return save(new EdgeEventEntity(edgeEvent)).orElse(null); - } finally { - readWriteLock.unlock(); + }; + queue = new TbSqlBlockingQueueWrapper<>(params, hashcodeFunction, batchThreads, statsFactory); + queue.init(logExecutor, v -> edgeEventInsertRepository.save(v), + Comparator.comparing(EdgeEventEntity::getTs) + ); + } + + @PreDestroy + private void destroy() { + if (queue != null) { + queue.destroy(); } } @Override - public PageData findEdgeEvents(UUID tenantId, EdgeId edgeId, TimePageLink pageLink, boolean withTsUpdate) { - final Lock readWriteLock = readWriteLocks.computeIfAbsent(edgeId, id -> new ReentrantLock()); - readWriteLock.lock(); - try { - if (withTsUpdate) { - return DaoUtil.toPageData( - edgeEventRepository - .findEdgeEventsByTenantIdAndEdgeId( - tenantId, - edgeId.getId(), - Objects.toString(pageLink.getTextSearch(), ""), - pageLink.getStartTime(), - pageLink.getEndTime(), - DaoUtil.toPageable(pageLink))); + public ListenableFuture saveAsync(EdgeEvent edgeEvent) { + log.debug("Save edge event [{}] ", edgeEvent); + if (edgeEvent.getId() == null) { + UUID timeBased = Uuids.timeBased(); + edgeEvent.setId(new EdgeEventId(timeBased)); + edgeEvent.setCreatedTime(Uuids.unixTimestamp(timeBased)); + } else if (edgeEvent.getCreatedTime() == 0L) { + UUID eventId = edgeEvent.getId().getId(); + if (eventId.version() == 1) { + edgeEvent.setCreatedTime(Uuids.unixTimestamp(eventId)); } else { - return DaoUtil.toPageData( - edgeEventRepository - .findEdgeEventsByTenantIdAndEdgeIdWithoutTimeseriesUpdated( - tenantId, - edgeId.getId(), - Objects.toString(pageLink.getTextSearch(), ""), - pageLink.getStartTime(), - pageLink.getEndTime(), - DaoUtil.toPageable(pageLink))); - + edgeEvent.setCreatedTime(System.currentTimeMillis()); } - } finally { - readWriteLock.unlock(); } + if (StringUtils.isEmpty(edgeEvent.getUid())) { + edgeEvent.setUid(edgeEvent.getId().toString()); + } + return save(new EdgeEventEntity(edgeEvent)); } - public Optional save(EdgeEventEntity entity) { + private ListenableFuture save(EdgeEventEntity entity) { log.debug("Save edge event [{}] ", entity); if (entity.getTenantId() == null) { log.trace("Save system edge event with predefined id {}", systemTenantId); @@ -135,7 +153,39 @@ public class JpaBaseEdgeEventDao extends JpaAbstractSearchTextDao addToQueue(EdgeEventEntity entity) { + return queue.add(entity); + } + + + @Override + public PageData findEdgeEvents(UUID tenantId, EdgeId edgeId, TimePageLink pageLink, boolean withTsUpdate) { + if (withTsUpdate) { + return DaoUtil.toPageData( + edgeEventRepository + .findEdgeEventsByTenantIdAndEdgeId( + tenantId, + edgeId.getId(), + Objects.toString(pageLink.getTextSearch(), ""), + pageLink.getStartTime(), + pageLink.getEndTime(), + DaoUtil.toPageable(pageLink))); + } else { + return DaoUtil.toPageData( + edgeEventRepository + .findEdgeEventsByTenantIdAndEdgeIdWithoutTimeseriesUpdated( + tenantId, + edgeId.getId(), + Objects.toString(pageLink.getTextSearch(), ""), + pageLink.getStartTime(), + pageLink.getEndTime(), + DaoUtil.toPageable(pageLink))); + + } } @Override diff --git a/dao/src/test/java/org/thingsboard/server/dao/service/BaseEdgeEventServiceTest.java b/dao/src/test/java/org/thingsboard/server/dao/service/BaseEdgeEventServiceTest.java index aa180a779d..33cb7e1bc1 100644 --- a/dao/src/test/java/org/thingsboard/server/dao/service/BaseEdgeEventServiceTest.java +++ b/dao/src/test/java/org/thingsboard/server/dao/service/BaseEdgeEventServiceTest.java @@ -16,6 +16,8 @@ package org.thingsboard.server.dao.service; import com.datastax.oss.driver.api.core.uuid.Uuids; +import com.google.common.util.concurrent.Futures; +import com.google.common.util.concurrent.ListenableFuture; import org.junit.Assert; import org.junit.Test; import org.thingsboard.server.common.data.edge.EdgeEvent; @@ -34,6 +36,8 @@ import java.io.IOException; import java.time.LocalDateTime; import java.time.Month; import java.time.ZoneOffset; +import java.util.ArrayList; +import java.util.List; public abstract class BaseEdgeEventServiceTest extends AbstractServiceTest { @@ -41,8 +45,14 @@ public abstract class BaseEdgeEventServiceTest extends AbstractServiceTest { public void saveEdgeEvent() throws Exception { EdgeId edgeId = new EdgeId(Uuids.timeBased()); DeviceId deviceId = new DeviceId(Uuids.timeBased()); - EdgeEvent edgeEvent = generateEdgeEvent(null, edgeId, deviceId, EdgeEventActionType.ADDED); - EdgeEvent saved = edgeEventService.save(edgeEvent); + TenantId tenantId = new TenantId(Uuids.timeBased()); + EdgeEvent edgeEvent = generateEdgeEvent(tenantId, edgeId, deviceId, EdgeEventActionType.ADDED); + edgeEventService.saveAsync(edgeEvent).get(); + + PageData edgeEvents = edgeEventService.findEdgeEvents(tenantId, edgeId, new TimePageLink(1), false); + Assert.assertFalse(edgeEvents.getData().isEmpty()); + + EdgeEvent saved = edgeEvents.getData().get(0); Assert.assertEquals(saved.getTenantId(), edgeEvent.getTenantId()); Assert.assertEquals(saved.getEdgeId(), edgeEvent.getEdgeId()); Assert.assertEquals(saved.getEntityId(), edgeEvent.getEntityId()); @@ -77,27 +87,31 @@ public abstract class BaseEdgeEventServiceTest extends AbstractServiceTest { EdgeId edgeId = new EdgeId(Uuids.timeBased()); DeviceId deviceId = new DeviceId(Uuids.timeBased()); TenantId tenantId = TenantId.fromUUID(Uuids.timeBased()); - saveEdgeEventWithProvidedTime(timeBeforeStartTime, edgeId, deviceId, tenantId); - EdgeEvent savedEdgeEvent = saveEdgeEventWithProvidedTime(eventTime, edgeId, deviceId, tenantId); - EdgeEvent savedEdgeEvent2 = saveEdgeEventWithProvidedTime(eventTime + 1, edgeId, deviceId, tenantId); - EdgeEvent savedEdgeEvent3 = saveEdgeEventWithProvidedTime(eventTime + 2, edgeId, deviceId, tenantId); - saveEdgeEventWithProvidedTime(timeAfterEndTime, edgeId, deviceId, tenantId); + + List> futures = new ArrayList<>(); + futures.add(saveEdgeEventWithProvidedTime(timeBeforeStartTime, edgeId, deviceId, tenantId)); + futures.add(saveEdgeEventWithProvidedTime(eventTime, edgeId, deviceId, tenantId)); + futures.add(saveEdgeEventWithProvidedTime(eventTime + 1, edgeId, deviceId, tenantId)); + futures.add(saveEdgeEventWithProvidedTime(eventTime + 2, edgeId, deviceId, tenantId)); + futures.add(saveEdgeEventWithProvidedTime(timeAfterEndTime, edgeId, deviceId, tenantId)); + + Futures.allAsList(futures).get(); TimePageLink pageLink = new TimePageLink(2, 0, "", new SortOrder("createdTime", SortOrder.Direction.DESC), startTime, endTime); PageData edgeEvents = edgeEventService.findEdgeEvents(tenantId, edgeId, pageLink, true); Assert.assertNotNull(edgeEvents.getData()); - Assert.assertTrue(edgeEvents.getData().size() == 2); - Assert.assertTrue(edgeEvents.getData().get(0).getUuidId().equals(savedEdgeEvent3.getUuidId())); - Assert.assertTrue(edgeEvents.getData().get(1).getUuidId().equals(savedEdgeEvent2.getUuidId())); + Assert.assertEquals(2, edgeEvents.getData().size()); + Assert.assertEquals(Uuids.startOf(eventTime + 2), edgeEvents.getData().get(0).getUuidId()); + Assert.assertEquals(Uuids.startOf(eventTime + 1), edgeEvents.getData().get(1).getUuidId()); Assert.assertTrue(edgeEvents.hasNext()); Assert.assertNotNull(pageLink.nextPageLink()); edgeEvents = edgeEventService.findEdgeEvents(tenantId, edgeId, pageLink.nextPageLink(), true); Assert.assertNotNull(edgeEvents.getData()); - Assert.assertTrue(edgeEvents.getData().size() == 1); - Assert.assertTrue(edgeEvents.getData().get(0).getUuidId().equals(savedEdgeEvent.getUuidId())); + Assert.assertEquals(1, edgeEvents.getData().size()); + Assert.assertEquals(Uuids.startOf(eventTime), edgeEvents.getData().get(0).getUuidId()); Assert.assertFalse(edgeEvents.hasNext()); } @@ -109,7 +123,7 @@ public abstract class BaseEdgeEventServiceTest extends AbstractServiceTest { TimePageLink pageLink = new TimePageLink(1, 0, null, new SortOrder("createdTime", SortOrder.Direction.ASC)); EdgeEvent edgeEventWithTsUpdate = generateEdgeEvent(tenantId, edgeId, deviceId, EdgeEventActionType.TIMESERIES_UPDATED); - edgeEventService.save(edgeEventWithTsUpdate); + edgeEventService.saveAsync(edgeEventWithTsUpdate).get(); PageData allEdgeEvents = edgeEventService.findEdgeEvents(tenantId, edgeId, pageLink, true); PageData edgeEventsWithoutTsUpdate = edgeEventService.findEdgeEvents(tenantId, edgeId, pageLink, false); @@ -121,9 +135,9 @@ public abstract class BaseEdgeEventServiceTest extends AbstractServiceTest { Assert.assertTrue(edgeEventsWithoutTsUpdate.getData().isEmpty()); } - private EdgeEvent saveEdgeEventWithProvidedTime(long time, EdgeId edgeId, EntityId entityId, TenantId tenantId) throws Exception { + private ListenableFuture saveEdgeEventWithProvidedTime(long time, EdgeId edgeId, EntityId entityId, TenantId tenantId) throws Exception { EdgeEvent edgeEvent = generateEdgeEvent(tenantId, edgeId, entityId, EdgeEventActionType.ADDED); edgeEvent.setId(new EdgeEventId(Uuids.startOf(time))); - return edgeEventService.save(edgeEvent); + return edgeEventService.saveAsync(edgeEvent); } } \ No newline at end of file diff --git a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/edge/AbstractTbMsgPushNode.java b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/edge/AbstractTbMsgPushNode.java index 93257ac48c..eb30a236af 100644 --- a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/edge/AbstractTbMsgPushNode.java +++ b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/edge/AbstractTbMsgPushNode.java @@ -80,10 +80,6 @@ public abstract class AbstractTbMsgPushNode entityBody = new HashMap<>(); Map metadata = msg.getMetaData().getData(); @@ -107,7 +103,11 @@ public abstract class AbstractTbMsgPushNode future = notifyEdge(ctx, edgeEvent, edgeId); + FutureCallback futureCallback = new FutureCallback<>() { + @Override + public void onSuccess(@Nullable Void result) { + ctx.tellSuccess(msg); + } + + @Override + public void onFailure(Throwable t) { + ctx.tellFailure(msg, t); + } + }; + Futures.addCallback(future, futureCallback, ctx.getDbCallbackExecutor()); } else { - tellFailure(ctx, msg); - } - } else { - PageLink pageLink = new PageLink(DEFAULT_PAGE_SIZE); - PageData pageData; - boolean edgeNotified = false; - do { - pageData = ctx.getEdgeService().findRelatedEdgeIdsByEntityId(ctx.getTenantId(), msg.getOriginator(), pageLink); - if (pageData != null && pageData.getData() != null && !pageData.getData().isEmpty()) { - for (EdgeId edgeId : pageData.getData()) { - EdgeEvent edgeEvent = buildEvent(msg, ctx); - if (edgeEvent != null) { - notifyEdge(ctx, msg, edgeEvent, edgeId); - edgeNotified = true; - } else { - tellFailure(ctx, msg); + PageLink pageLink = new PageLink(DEFAULT_PAGE_SIZE); + PageData pageData; + List> futures = new ArrayList<>(); + do { + pageData = ctx.getEdgeService().findRelatedEdgeIdsByEntityId(ctx.getTenantId(), msg.getOriginator(), pageLink); + if (pageData != null && pageData.getData() != null && !pageData.getData().isEmpty()) { + for (EdgeId edgeId : pageData.getData()) { + EdgeEvent edgeEvent = buildEvent(msg, ctx); + futures.add(notifyEdge(ctx, edgeEvent, edgeId)); + } + if (pageData.hasNext()) { + pageLink = pageLink.nextPageLink(); } } - if (pageData.hasNext()) { - pageLink = pageLink.nextPageLink(); - } - } - } while (pageData != null && pageData.hasNext()); + } while (pageData != null && pageData.hasNext()); + + if (futures.isEmpty()) { + // ack in case no edges are related to provided entity + ctx.ack(msg); + } else { + Futures.addCallback(Futures.allAsList(futures), new FutureCallback<>() { + @Override + public void onSuccess(@Nullable List voids) { + ctx.tellSuccess(msg); + } - if (!edgeNotified) { - // ack in case no edges are related to provided entity - ctx.ack(msg); + @Override + public void onFailure(Throwable t) { + ctx.tellFailure(msg, t); + } + }, ctx.getDbCallbackExecutor()); + } } + } catch (Exception e) { + log.error("Failed to build edge event", e); + ctx.tellFailure(msg, e); } } - private void tellFailure(TbContext ctx, TbMsg msg) { - String errMsg = String.format("Edge event type is null. Entity Type %s", msg.getOriginator().getEntityType()); - log.warn(errMsg); - ctx.tellFailure(msg, new RuntimeException(errMsg)); - } - - private void notifyEdge(TbContext ctx, TbMsg msg, EdgeEvent edgeEvent, EdgeId edgeId) { + private ListenableFuture notifyEdge(TbContext ctx, EdgeEvent edgeEvent, EdgeId edgeId) { edgeEvent.setEdgeId(edgeId); - ctx.getEdgeEventService().save(edgeEvent); - ctx.tellNext(msg, SUCCESS); - ctx.onEdgeEventUpdate(ctx.getTenantId(), edgeId); + ListenableFuture future = ctx.getEdgeEventService().saveAsync(edgeEvent); + return Futures.transform(future, result -> { + ctx.onEdgeEventUpdate(ctx.getTenantId(), edgeId); + return null; + }, ctx.getDbCallbackExecutor()); } }