diff --git a/.gitignore b/.gitignore index da8569baa3..c0628fd2e3 100644 --- a/.gitignore +++ b/.gitignore @@ -33,3 +33,4 @@ pom.xml.versionsBackup **/.env .instance_id rebuild-docker.sh +.run/ \ No newline at end of file diff --git a/application/src/main/data/upgrade/2.6.0/schema_update_ttl.sql b/application/src/main/data/upgrade/2.6.0/schema_update_ttl.sql new file mode 100644 index 0000000000..7c1075059f --- /dev/null +++ b/application/src/main/data/upgrade/2.6.0/schema_update_ttl.sql @@ -0,0 +1,32 @@ +-- +-- Copyright © 2016-2020 The Thingsboard Authors +-- +-- Licensed under the Apache License, Version 2.0 (the "License"); +-- you may not use this file except in compliance with the License. +-- You may obtain a copy of the License at +-- +-- http://www.apache.org/licenses/LICENSE-2.0 +-- +-- Unless required by applicable law or agreed to in writing, software +-- distributed under the License is distributed on an "AS IS" BASIS, +-- WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +-- See the License for the specific language governing permissions and +-- limitations under the License. +-- + +CREATE OR REPLACE PROCEDURE cleanup_edge_events_by_ttl(IN ttl bigint, INOUT deleted bigint) + LANGUAGE plpgsql AS +$$ +DECLARE + ttl_ts bigint; + ttl_deleted_count bigint DEFAULT 0; +BEGIN + IF ttl > 0 THEN + ttl_ts := (EXTRACT(EPOCH FROM current_timestamp) * 1000 - ttl::bigint * 1000)::bigint; + EXECUTE format( + 'WITH deleted AS (DELETE FROM edge_event WHERE ts < %L::bigint RETURNING *) SELECT count(*) FROM deleted', ttl_ts) into ttl_deleted_count; + END IF; + RAISE NOTICE 'Edge events removed by ttl: %', ttl_deleted_count; + deleted := ttl_deleted_count; +END +$$; diff --git a/application/src/main/java/org/thingsboard/server/controller/AlarmController.java b/application/src/main/java/org/thingsboard/server/controller/AlarmController.java index 01c5ea28ae..4aaeb58fd5 100644 --- a/application/src/main/java/org/thingsboard/server/controller/AlarmController.java +++ b/application/src/main/java/org/thingsboard/server/controller/AlarmController.java @@ -92,7 +92,8 @@ public class AlarmController extends BaseController { getCurrentUser().getCustomerId(), alarm.getId() == null ? ActionType.ADDED : ActionType.UPDATED, null); - sendNotificationMsgToEdgeService(getTenantId(), savedAlarm.getId(), alarm.getId() == null ? ActionType.ADDED : ActionType.UPDATED); + sendNotificationMsgToEdgeService(getTenantId(), savedAlarm.getId(), + alarm.getId() == null ? ActionType.ADDED : ActionType.UPDATED); return savedAlarm; } catch (Exception e) { diff --git a/application/src/main/java/org/thingsboard/server/controller/AssetController.java b/application/src/main/java/org/thingsboard/server/controller/AssetController.java index 20fae1e4c6..0877937afa 100644 --- a/application/src/main/java/org/thingsboard/server/controller/AssetController.java +++ b/application/src/main/java/org/thingsboard/server/controller/AssetController.java @@ -88,13 +88,13 @@ public class AssetController extends BaseController { Asset savedAsset = checkNotNull(assetService.saveAsset(asset)); - sendNotificationMsgToEdgeService(savedAsset.getTenantId(), null, - savedAsset.getId(), EdgeEventType.ASSET, asset.getId() == null ? ActionType.ADDED : ActionType.UPDATED); - logEntityAction(savedAsset.getId(), savedAsset, savedAsset.getCustomerId(), asset.getId() == null ? ActionType.ADDED : ActionType.UPDATED, null); + sendNotificationMsgToEdgeService(getTenantId(), savedAsset.getId(), + asset.getId() == null ? ActionType.ADDED : ActionType.UPDATED); + return savedAsset; } catch (Exception e) { logEntityAction(emptyId(EntityType.ASSET), asset, @@ -117,7 +117,7 @@ public class AssetController extends BaseController { asset.getCustomerId(), ActionType.DELETED, null, strAssetId); - sendNotificationMsgToEdgeService(getTenantId(), null, assetId, EdgeEventType.ASSET, ActionType.DELETED); + sendNotificationMsgToEdgeService(getTenantId(), assetId, ActionType.DELETED); } catch (Exception e) { logEntityAction(emptyId(EntityType.ASSET), null, @@ -360,7 +360,7 @@ public class AssetController extends BaseController { savedAsset.getCustomerId(), ActionType.ASSIGNED_TO_EDGE, null, strAssetId, strEdgeId, edge.getName()); - sendNotificationMsgToEdgeService(getTenantId(), edgeId, savedAsset.getId(), EdgeEventType.ASSET, ActionType.ASSIGNED_TO_EDGE); + sendNotificationMsgToEdgeService(getTenantId(), edgeId, savedAsset.getId(), ActionType.ASSIGNED_TO_EDGE); return savedAsset; } catch (Exception e) { @@ -393,8 +393,7 @@ public class AssetController extends BaseController { asset.getCustomerId(), ActionType.UNASSIGNED_FROM_EDGE, null, strAssetId, edge.getId().toString(), edge.getName()); - sendNotificationMsgToEdgeService(getTenantId(), edgeId, savedAsset.getId(), - EdgeEventType.ASSET, ActionType.UNASSIGNED_FROM_EDGE); + sendNotificationMsgToEdgeService(getTenantId(), edgeId, savedAsset.getId(), ActionType.UNASSIGNED_FROM_EDGE); return savedAsset; } catch (Exception e) { diff --git a/application/src/main/java/org/thingsboard/server/controller/AuthController.java b/application/src/main/java/org/thingsboard/server/controller/AuthController.java index e999320e43..e00ee733b1 100644 --- a/application/src/main/java/org/thingsboard/server/controller/AuthController.java +++ b/application/src/main/java/org/thingsboard/server/controller/AuthController.java @@ -126,7 +126,7 @@ public class AuthController extends BaseController { userCredentials.setPassword(passwordEncoder.encode(newPassword)); userService.replaceUserCredentials(securityUser.getTenantId(), userCredentials); - sendNotificationMsgToEdgeService(getTenantId(), null, userCredentials.getUserId(), EdgeEventType.USER, ActionType.CREDENTIALS_UPDATED); + sendNotificationMsgToEdgeService(getTenantId(), userCredentials.getUserId(), ActionType.CREDENTIALS_UPDATED); } catch (Exception e) { throw handleException(e); diff --git a/application/src/main/java/org/thingsboard/server/controller/BaseController.java b/application/src/main/java/org/thingsboard/server/controller/BaseController.java index e435ad6bbb..390b2187fa 100644 --- a/application/src/main/java/org/thingsboard/server/controller/BaseController.java +++ b/application/src/main/java/org/thingsboard/server/controller/BaseController.java @@ -718,6 +718,14 @@ public abstract class BaseController { return result; } + protected void sendNotificationMsgToEdgeService(TenantId tenantId, EdgeId edgeId, CustomerId customerId, ActionType edgeEventAction) { + try { + sendNotificationMsgToEdgeService(tenantId, edgeId, null, json.writeValueAsString(customerId), EdgeEventType.EDGE, edgeEventAction); + } catch (Exception e) { + log.warn("Failed to push assign/unassign to/from customer to core: {}", customerId, e); + } + } + protected void sendNotificationMsgToEdgeService(TenantId tenantId, EntityRelation relation, ActionType edgeEventAction) { try { if (!relation.getFrom().getEntityType().equals(EntityType.EDGE) && @@ -730,16 +738,16 @@ public abstract class BaseController { } protected void sendNotificationMsgToEdgeService(TenantId tenantId, EntityId entityId, ActionType edgeEventAction) { + sendNotificationMsgToEdgeService(tenantId, null, entityId, edgeEventAction); + } + + protected void sendNotificationMsgToEdgeService(TenantId tenantId, EdgeId edgeId, EntityId entityId, ActionType edgeEventAction) { EdgeEventType edgeEventType = EdgeUtils.getEdgeEventTypeByEntityType(entityId.getEntityType()); if (edgeEventType != null) { - sendNotificationMsgToEdgeService(tenantId, null, entityId, null, edgeEventType, edgeEventAction); + sendNotificationMsgToEdgeService(tenantId, edgeId, entityId, null, edgeEventType, edgeEventAction); } } - protected void sendNotificationMsgToEdgeService(TenantId tenantId, EdgeId edgeId, EntityId entityId, EdgeEventType edgeEventType, ActionType edgeEventAction) { - sendNotificationMsgToEdgeService(tenantId, edgeId, entityId, null, edgeEventType, edgeEventAction); - } - private void sendNotificationMsgToEdgeService(TenantId tenantId, EdgeId edgeId, EntityId entityId, String entityBody, EdgeEventType edgeEventType, ActionType edgeEventAction) { TransportProtos.EdgeNotificationMsgProto.Builder builder = TransportProtos.EdgeNotificationMsgProto.newBuilder(); builder.setTenantIdMSB(tenantId.getId().getMostSignificantBits()); diff --git a/application/src/main/java/org/thingsboard/server/controller/DashboardController.java b/application/src/main/java/org/thingsboard/server/controller/DashboardController.java index 22674b6ced..0ae455d8f4 100644 --- a/application/src/main/java/org/thingsboard/server/controller/DashboardController.java +++ b/application/src/main/java/org/thingsboard/server/controller/DashboardController.java @@ -117,8 +117,8 @@ public class DashboardController extends BaseController { null, dashboard.getId() == null ? ActionType.ADDED : ActionType.UPDATED, null); - sendNotificationMsgToEdgeService(savedDashboard.getTenantId(), null, savedDashboard.getId(), - EdgeEventType.DASHBOARD, savedDashboard.getId() == null ? ActionType.ADDED : ActionType.UPDATED); + sendNotificationMsgToEdgeService(savedDashboard.getTenantId(), savedDashboard.getId(), + dashboard.getId() == null ? ActionType.ADDED : ActionType.UPDATED); return savedDashboard; } catch (Exception e) { @@ -143,7 +143,7 @@ public class DashboardController extends BaseController { null, ActionType.DELETED, null, strDashboardId); - sendNotificationMsgToEdgeService(getTenantId(), null, dashboardId, EdgeEventType.DASHBOARD, ActionType.DELETED); + sendNotificationMsgToEdgeService(getTenantId(), dashboardId, ActionType.DELETED); } catch (Exception e) { logEntityAction(emptyId(EntityType.DASHBOARD), @@ -500,8 +500,7 @@ public class DashboardController extends BaseController { null, ActionType.ASSIGNED_TO_EDGE, null, strDashboardId, strEdgeId, edge.getName()); - sendNotificationMsgToEdgeService(getTenantId(), edgeId, savedDashboard.getId(), - EdgeEventType.DASHBOARD, ActionType.ASSIGNED_TO_EDGE); + sendNotificationMsgToEdgeService(getTenantId(), edgeId, savedDashboard.getId(), ActionType.ASSIGNED_TO_EDGE); return savedDashboard; } catch (Exception e) { @@ -533,8 +532,7 @@ public class DashboardController extends BaseController { null, ActionType.UNASSIGNED_FROM_EDGE, null, strDashboardId, edge.getId().toString(), edge.getName()); - sendNotificationMsgToEdgeService(getTenantId(), edgeId, savedDashboard.getId(), - EdgeEventType.DASHBOARD, ActionType.UNASSIGNED_FROM_EDGE); + sendNotificationMsgToEdgeService(getTenantId(), edgeId, savedDashboard.getId(), ActionType.UNASSIGNED_FROM_EDGE); return savedDashboard; } catch (Exception e) { diff --git a/application/src/main/java/org/thingsboard/server/controller/DeviceController.java b/application/src/main/java/org/thingsboard/server/controller/DeviceController.java index 6c7391127d..b65d6387c2 100644 --- a/application/src/main/java/org/thingsboard/server/controller/DeviceController.java +++ b/application/src/main/java/org/thingsboard/server/controller/DeviceController.java @@ -108,8 +108,8 @@ public class DeviceController extends BaseController { tbClusterService.pushMsgToCore(new DeviceNameOrTypeUpdateMsg(savedDevice.getTenantId(), savedDevice.getId(), savedDevice.getName(), savedDevice.getType()), null); - sendNotificationMsgToEdgeService(savedDevice.getTenantId(), null, savedDevice.getId(), - EdgeEventType.DEVICE, device.getId() == null ? ActionType.ADDED : ActionType.UPDATED); + sendNotificationMsgToEdgeService(savedDevice.getTenantId(), savedDevice.getId(), + device.getId() == null ? ActionType.ADDED : ActionType.UPDATED); logEntityAction(savedDevice.getId(), savedDevice, savedDevice.getCustomerId(), @@ -142,7 +142,7 @@ public class DeviceController extends BaseController { device.getCustomerId(), ActionType.DELETED, null, strDeviceId); - sendNotificationMsgToEdgeService(getTenantId(), null, deviceId, EdgeEventType.DEVICE, ActionType.DELETED); + sendNotificationMsgToEdgeService(getTenantId(), deviceId, ActionType.DELETED); deviceStateService.onDeviceDeleted(device); } catch (Exception e) { @@ -267,7 +267,7 @@ public class DeviceController extends BaseController { tbClusterService.pushMsgToCore(new DeviceCredentialsUpdateNotificationMsg(getCurrentUser().getTenantId(), deviceCredentials.getDeviceId()), null); - sendNotificationMsgToEdgeService(getTenantId(), null, device.getId(), EdgeEventType.DEVICE, ActionType.CREDENTIALS_UPDATED); + sendNotificationMsgToEdgeService(getTenantId(), device.getId(), ActionType.CREDENTIALS_UPDATED); logEntityAction(device.getId(), device, device.getCustomerId(), @@ -518,7 +518,7 @@ public class DeviceController extends BaseController { savedDevice.getCustomerId(), ActionType.ASSIGNED_TO_EDGE, null, strDeviceId, strEdgeId, edge.getName()); - sendNotificationMsgToEdgeService(getTenantId(), edgeId, savedDevice.getId(), EdgeEventType.DEVICE, ActionType.ASSIGNED_TO_EDGE); + sendNotificationMsgToEdgeService(getTenantId(), edgeId, savedDevice.getId(), ActionType.ASSIGNED_TO_EDGE); return savedDevice; } catch (Exception e) { @@ -549,7 +549,7 @@ public class DeviceController extends BaseController { device.getCustomerId(), ActionType.UNASSIGNED_FROM_EDGE, null, strDeviceId, edge.getId().toString(), edge.getName()); - sendNotificationMsgToEdgeService(getTenantId(), edgeId, savedDevice.getId(), EdgeEventType.DEVICE, ActionType.UNASSIGNED_FROM_EDGE); + sendNotificationMsgToEdgeService(getTenantId(), edgeId, savedDevice.getId(), ActionType.UNASSIGNED_FROM_EDGE); return savedDevice; } catch (Exception e) { 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 3c17176b88..19b82b3b85 100644 --- a/application/src/main/java/org/thingsboard/server/controller/EdgeController.java +++ b/application/src/main/java/org/thingsboard/server/controller/EdgeController.java @@ -178,6 +178,9 @@ public class EdgeController extends BaseController { savedEdge.getCustomerId(), ActionType.ASSIGNED_TO_CUSTOMER, null, strEdgeId, strCustomerId, customer.getName()); + sendNotificationMsgToEdgeService(savedEdge.getTenantId(), savedEdge.getId(), + customerId, ActionType.ASSIGNED_TO_CUSTOMER); + return savedEdge; } catch (Exception e) { logEntityAction(emptyId(EntityType.EDGE), null, @@ -206,6 +209,9 @@ public class EdgeController extends BaseController { edge.getCustomerId(), ActionType.UNASSIGNED_FROM_CUSTOMER, null, strEdgeId, customer.getId().toString(), customer.getName()); + sendNotificationMsgToEdgeService(savedEdge.getTenantId(), savedEdge.getId(), + edge.getCustomerId(), ActionType.UNASSIGNED_FROM_CUSTOMER); + return savedEdge; } catch (Exception e) { logEntityAction(emptyId(EntityType.EDGE), null, diff --git a/application/src/main/java/org/thingsboard/server/controller/EntityViewController.java b/application/src/main/java/org/thingsboard/server/controller/EntityViewController.java index f2b95963f5..821dbcbd70 100644 --- a/application/src/main/java/org/thingsboard/server/controller/EntityViewController.java +++ b/application/src/main/java/org/thingsboard/server/controller/EntityViewController.java @@ -118,8 +118,8 @@ public class EntityViewController extends BaseController { logEntityAction(savedEntityView.getId(), savedEntityView, null, entityView.getId() == null ? ActionType.ADDED : ActionType.UPDATED, null); - sendNotificationMsgToEdgeService(getTenantId(), null, savedEntityView.getId(), - EdgeEventType.ENTITY_VIEW, entityView.getId() == null ? ActionType.ADDED : ActionType.UPDATED); + sendNotificationMsgToEdgeService(getTenantId(), savedEntityView.getId(), + entityView.getId() == null ? ActionType.ADDED : ActionType.UPDATED); return savedEntityView; } catch (Exception e) { logEntityAction(emptyId(EntityType.ENTITY_VIEW), entityView, null, @@ -190,7 +190,7 @@ public class EntityViewController extends BaseController { logEntityAction(entityViewId, entityView, entityView.getCustomerId(), ActionType.DELETED, null, strEntityViewId); - sendNotificationMsgToEdgeService(getTenantId(), null, entityViewId, EdgeEventType.ENTITY_VIEW, ActionType.DELETED); + sendNotificationMsgToEdgeService(getTenantId(), entityViewId, ActionType.DELETED); } catch (Exception e) { logEntityAction(emptyId(EntityType.ENTITY_VIEW), null, @@ -396,8 +396,7 @@ public class EntityViewController extends BaseController { savedEntityView.getCustomerId(), ActionType.ASSIGNED_TO_EDGE, null, strEntityViewId, strEdgeId, edge.getName()); - sendNotificationMsgToEdgeService(getTenantId(), edgeId, savedEntityView.getId(), - EdgeEventType.ENTITY_VIEW, ActionType.ASSIGNED_TO_EDGE); + sendNotificationMsgToEdgeService(getTenantId(), edgeId, savedEntityView.getId(), ActionType.ASSIGNED_TO_EDGE); return savedEntityView; } catch (Exception e) { @@ -427,8 +426,7 @@ public class EntityViewController extends BaseController { entityView.getCustomerId(), ActionType.UNASSIGNED_FROM_EDGE, null, strEntityViewId, edge.getId().toString(), edge.getName()); - sendNotificationMsgToEdgeService(getTenantId(), edgeId, savedEntityView.getId(), - EdgeEventType.ENTITY_VIEW, ActionType.UNASSIGNED_FROM_EDGE); + sendNotificationMsgToEdgeService(getTenantId(), edgeId, savedEntityView.getId(), ActionType.UNASSIGNED_FROM_EDGE); return savedEntityView; } catch (Exception e) { diff --git a/application/src/main/java/org/thingsboard/server/controller/RuleChainController.java b/application/src/main/java/org/thingsboard/server/controller/RuleChainController.java index 463934b4ab..4d2e78bc67 100644 --- a/application/src/main/java/org/thingsboard/server/controller/RuleChainController.java +++ b/application/src/main/java/org/thingsboard/server/controller/RuleChainController.java @@ -145,9 +145,8 @@ public class RuleChainController extends BaseController { created ? ActionType.ADDED : ActionType.UPDATED, null); if (RuleChainType.EDGE.equals(savedRuleChain.getType())) { - sendNotificationMsgToEdgeService(savedRuleChain.getTenantId(), null, - savedRuleChain.getId(), EdgeEventType.RULE_CHAIN, - savedRuleChain.getId() == null ? ActionType.ADDED : ActionType.UPDATED); + sendNotificationMsgToEdgeService(savedRuleChain.getTenantId(), savedRuleChain.getId(), + created ? ActionType.ADDED : ActionType.UPDATED); } return savedRuleChain; @@ -226,9 +225,7 @@ public class RuleChainController extends BaseController { if (RuleChainType.EDGE.equals(ruleChain.getType())) { sendNotificationMsgToEdgeService(ruleChain.getTenantId(), - null, - ruleChain.getId(), EdgeEventType.RULE_CHAIN, - ActionType.UPDATED); + ruleChain.getId(), ActionType.UPDATED); } return savedRuleChainMetaData; @@ -293,8 +290,7 @@ public class RuleChainController extends BaseController { ActionType.DELETED, null, strRuleChainId); if (RuleChainType.EDGE.equals(ruleChain.getType())) { - sendNotificationMsgToEdgeService(ruleChain.getTenantId(), null, - ruleChain.getId(), EdgeEventType.RULE_CHAIN, ActionType.DELETED); + sendNotificationMsgToEdgeService(ruleChain.getTenantId(), ruleChain.getId(), ActionType.DELETED); } } catch (Exception e) { @@ -426,8 +422,7 @@ public class RuleChainController extends BaseController { null, ActionType.ASSIGNED_TO_EDGE, null, strRuleChainId, strEdgeId, edge.getName()); - sendNotificationMsgToEdgeService(getTenantId(), edgeId, savedRuleChain.getId(), - EdgeEventType.RULE_CHAIN, ActionType.ASSIGNED_TO_EDGE); + sendNotificationMsgToEdgeService(getTenantId(), edgeId, savedRuleChain.getId(), ActionType.ASSIGNED_TO_EDGE); return savedRuleChain; } catch (Exception e) { @@ -459,8 +454,7 @@ public class RuleChainController extends BaseController { null, ActionType.UNASSIGNED_FROM_EDGE, null, strRuleChainId, edge.getId().toString(), edge.getName()); - sendNotificationMsgToEdgeService(getTenantId(), edgeId, savedRuleChain.getId(), - EdgeEventType.RULE_CHAIN, ActionType.UNASSIGNED_FROM_EDGE); + sendNotificationMsgToEdgeService(getTenantId(), edgeId, savedRuleChain.getId(), ActionType.UNASSIGNED_FROM_EDGE); return savedRuleChain; } catch (Exception e) { diff --git a/application/src/main/java/org/thingsboard/server/controller/UserController.java b/application/src/main/java/org/thingsboard/server/controller/UserController.java index 8dedc59098..25e68d3571 100644 --- a/application/src/main/java/org/thingsboard/server/controller/UserController.java +++ b/application/src/main/java/org/thingsboard/server/controller/UserController.java @@ -162,7 +162,8 @@ public class UserController extends BaseController { savedUser.getCustomerId(), user.getId() == null ? ActionType.ADDED : ActionType.UPDATED, null); - sendNotificationMsgToEdgeService(getTenantId(), null, user.getId(), EdgeEventType.USER, user.getId() == null ? ActionType.ADDED : ActionType.UPDATED); + sendNotificationMsgToEdgeService(getTenantId(), savedUser.getId(), + user.getId() == null ? ActionType.ADDED : ActionType.UPDATED); return savedUser; } catch (Exception e) { @@ -239,7 +240,7 @@ public class UserController extends BaseController { user.getCustomerId(), ActionType.DELETED, null, strUserId); - sendNotificationMsgToEdgeService(getTenantId(), null, user.getId(), EdgeEventType.USER, ActionType.DELETED); + sendNotificationMsgToEdgeService(getTenantId(), user.getId(), ActionType.DELETED); } catch (Exception e) { logEntityAction(emptyId(EntityType.USER), diff --git a/application/src/main/java/org/thingsboard/server/controller/WidgetTypeController.java b/application/src/main/java/org/thingsboard/server/controller/WidgetTypeController.java index ffff90536f..45316f77c5 100644 --- a/application/src/main/java/org/thingsboard/server/controller/WidgetTypeController.java +++ b/application/src/main/java/org/thingsboard/server/controller/WidgetTypeController.java @@ -15,6 +15,7 @@ */ package org.thingsboard.server.controller; +import lombok.extern.slf4j.Slf4j; import org.springframework.http.HttpStatus; import org.springframework.security.access.prepost.PreAuthorize; import org.springframework.web.bind.annotation.PathVariable; @@ -25,6 +26,7 @@ import org.springframework.web.bind.annotation.RequestParam; import org.springframework.web.bind.annotation.ResponseBody; import org.springframework.web.bind.annotation.ResponseStatus; import org.springframework.web.bind.annotation.RestController; +import org.thingsboard.server.common.data.audit.ActionType; import org.thingsboard.server.common.data.exception.ThingsboardException; import org.thingsboard.server.common.data.id.TenantId; import org.thingsboard.server.common.data.id.WidgetTypeId; @@ -37,6 +39,7 @@ import org.thingsboard.server.service.security.permission.Resource; import java.util.List; +@Slf4j @RestController @TbCoreComponent @RequestMapping("/api") @@ -67,8 +70,12 @@ public class WidgetTypeController extends BaseController { } checkEntity(widgetType.getId(), widgetType, Resource.WIDGET_TYPE); + WidgetType savedWidgetType = widgetTypeService.saveWidgetType(widgetType); - return checkNotNull(widgetTypeService.saveWidgetType(widgetType)); + sendNotificationMsgToEdgeService(getTenantId(), savedWidgetType.getId(), + widgetType.getId() == null ? ActionType.ADDED : ActionType.UPDATED); + + return checkNotNull(savedWidgetType); } catch (Exception e) { throw handleException(e); } @@ -83,6 +90,9 @@ public class WidgetTypeController extends BaseController { WidgetTypeId widgetTypeId = new WidgetTypeId(toUUID(strWidgetTypeId)); checkWidgetTypeId(widgetTypeId, Operation.DELETE); widgetTypeService.deleteWidgetType(getCurrentUser().getTenantId(), widgetTypeId); + + sendNotificationMsgToEdgeService(getTenantId(), widgetTypeId, ActionType.DELETED); + } catch (Exception e) { throw handleException(e); } diff --git a/application/src/main/java/org/thingsboard/server/controller/WidgetsBundleController.java b/application/src/main/java/org/thingsboard/server/controller/WidgetsBundleController.java index 1b6b371297..d6442a3dc2 100644 --- a/application/src/main/java/org/thingsboard/server/controller/WidgetsBundleController.java +++ b/application/src/main/java/org/thingsboard/server/controller/WidgetsBundleController.java @@ -25,6 +25,7 @@ import org.springframework.web.bind.annotation.RequestParam; import org.springframework.web.bind.annotation.ResponseBody; import org.springframework.web.bind.annotation.ResponseStatus; import org.springframework.web.bind.annotation.RestController; +import org.thingsboard.server.common.data.audit.ActionType; import org.thingsboard.server.common.data.exception.ThingsboardException; import org.thingsboard.server.common.data.id.TenantId; import org.thingsboard.server.common.data.id.WidgetsBundleId; @@ -68,7 +69,12 @@ public class WidgetsBundleController extends BaseController { } checkEntity(widgetsBundle.getId(), widgetsBundle, Resource.WIDGETS_BUNDLE); - return checkNotNull(widgetsBundleService.saveWidgetsBundle(widgetsBundle)); + WidgetsBundle savedWidgetsBundle = widgetsBundleService.saveWidgetsBundle(widgetsBundle); + + sendNotificationMsgToEdgeService(getTenantId(), savedWidgetsBundle.getId(), + widgetsBundle.getId() == null ? ActionType.ADDED : ActionType.UPDATED); + + return checkNotNull(savedWidgetsBundle); } catch (Exception e) { throw handleException(e); } @@ -83,6 +89,9 @@ public class WidgetsBundleController extends BaseController { WidgetsBundleId widgetsBundleId = new WidgetsBundleId(toUUID(strWidgetsBundleId)); checkWidgetsBundleId(widgetsBundleId, Operation.DELETE); widgetsBundleService.deleteWidgetsBundle(getTenantId(), widgetsBundleId); + + sendNotificationMsgToEdgeService(getTenantId(), widgetsBundleId, ActionType.DELETED); + } catch (Exception e) { throw handleException(e); } 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 c90c05746f..279ec96064 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 @@ -18,9 +18,11 @@ package org.thingsboard.server.service.edge; import com.fasterxml.jackson.core.JsonProcessingException; import com.fasterxml.jackson.databind.JsonNode; import com.fasterxml.jackson.databind.ObjectMapper; +import com.google.common.util.concurrent.FutureCallback; import com.google.common.util.concurrent.Futures; import com.google.common.util.concurrent.ListenableFuture; import lombok.extern.slf4j.Slf4j; +import org.checkerframework.checker.nullness.qual.Nullable; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.stereotype.Service; import org.thingsboard.server.common.data.EntityType; @@ -46,12 +48,15 @@ import org.thingsboard.server.common.data.page.TimePageData; import org.thingsboard.server.common.data.page.TimePageLink; import org.thingsboard.server.common.data.relation.EntityRelation; import org.thingsboard.server.common.data.relation.RelationTypeGroup; +import org.thingsboard.server.common.data.rule.RuleChain; +import org.thingsboard.server.common.data.rule.RuleChainConnectionInfo; import org.thingsboard.server.common.msg.queue.TbCallback; import org.thingsboard.server.dao.alarm.AlarmService; import org.thingsboard.server.dao.edge.EdgeEventService; import org.thingsboard.server.dao.edge.EdgeService; import org.thingsboard.server.dao.model.ModelConstants; import org.thingsboard.server.dao.relation.RelationService; +import org.thingsboard.server.dao.rule.RuleChainService; import org.thingsboard.server.dao.user.UserService; import org.thingsboard.server.gen.transport.TransportProtos; import org.thingsboard.server.queue.util.TbCoreComponent; @@ -86,6 +91,9 @@ public class DefaultEdgeNotificationService implements EdgeNotificationService { @Autowired private UserService userService; + @Autowired + private RuleChainService ruleChainService; + @Autowired private RelationService relationService; @@ -149,14 +157,17 @@ public class DefaultEdgeNotificationService implements EdgeNotificationService { TenantId tenantId = new TenantId(new UUID(edgeNotificationMsg.getTenantIdMSB(), edgeNotificationMsg.getTenantIdLSB())); EdgeEventType edgeEventType = EdgeEventType.valueOf(edgeNotificationMsg.getEdgeEventType()); switch (edgeEventType) { - // TODO: voba - handle edge updates - // case EDGE: + case EDGE: + processEdge(tenantId, edgeNotificationMsg); + break; case USER: case ASSET: case DEVICE: case ENTITY_VIEW: case DASHBOARD: case RULE_CHAIN: + case WIDGETS_BUNDLE: + case WIDGET_TYPE: processEntity(tenantId, edgeNotificationMsg); break; case ALARM: @@ -176,29 +187,76 @@ public class DefaultEdgeNotificationService implements EdgeNotificationService { } } + private void processEdge(TenantId tenantId, TransportProtos.EdgeNotificationMsgProto edgeNotificationMsg) { + // TODO: voba - handle edge updates + try { + ActionType edgeEventActionType = ActionType.valueOf(edgeNotificationMsg.getEdgeEventAction()); + EdgeId edgeId = new EdgeId(new UUID(edgeNotificationMsg.getEdgeIdMSB(), edgeNotificationMsg.getEdgeIdLSB())); + switch (edgeEventActionType) { + case ASSIGNED_TO_CUSTOMER: + case UNASSIGNED_FROM_CUSTOMER: + CustomerId customerId = mapper.readValue(edgeNotificationMsg.getEntityBody(), CustomerId.class); + ListenableFuture edgeFuture = edgeService.findEdgeByIdAsync(tenantId, edgeId); + Futures.addCallback(edgeFuture, new FutureCallback() { + @Override + public void onSuccess(@Nullable Edge edge) { + if (edge != null && customerId != null && !EntityId.NULL_UUID.equals(customerId.getId())) { + ActionType actionType = ActionType.ASSIGNED_TO_CUSTOMER.equals(edgeEventActionType) ? ActionType.ADDED : ActionType.DELETED; + saveEdgeEvent(edge.getTenantId(), edge.getId(), EdgeEventType.CUSTOMER, actionType, customerId, null); + TextPageData pageData = userService.findCustomerUsers(tenantId, customerId, new TextPageLink(Integer.MAX_VALUE)); + if (pageData != null && pageData.getData() != null && !pageData.getData().isEmpty()) { + log.trace("[{}] [{}] user(s) are going to be {} to edge.", edge.getId(), pageData.getData().size(), actionType.name()); + for (User user : pageData.getData()) { + saveEdgeEvent(edge.getTenantId(), edge.getId(), EdgeEventType.USER, actionType, user.getId(), null); + } + } + } + } + + @Override + public void onFailure(Throwable t) { + log.error("Can't find edge by id [{}]", edgeNotificationMsg, t); + } + }, dbCallbackExecutorService); + break; + } + } catch (Exception e) { + log.error("Exception during processing edge event", e); + } + } + private void processEntity(TenantId tenantId, TransportProtos.EdgeNotificationMsgProto edgeNotificationMsg) { ActionType edgeEventActionType = ActionType.valueOf(edgeNotificationMsg.getEdgeEventAction()); EdgeEventType edgeEventType = EdgeEventType.valueOf(edgeNotificationMsg.getEdgeEventType()); EntityId entityId = EntityIdFactory.getByEdgeEventTypeAndUuid(edgeEventType, new UUID(edgeNotificationMsg.getEntityIdMSB(), edgeNotificationMsg.getEntityIdLSB())); switch (edgeEventActionType) { // TODO: voba - ADDED is not required for CE version ? - // case ADDED: + case ADDED: case UPDATED: case CREDENTIALS_UPDATED: - ListenableFuture> edgeIdsFuture = findRelatedEdgeIdsByEntityId(tenantId, entityId); - Futures.transform(edgeIdsFuture, edgeIds -> { - if (edgeIds != null && !edgeIds.isEmpty()) { - for (EdgeId edgeId : edgeIds) { - try { - saveEdgeEvent(tenantId, edgeId, edgeEventType, edgeEventActionType, entityId, null); - } catch (Exception e) { - log.error("[{}] Failed to push event to edge, edgeId [{}], edgeEventType [{}], edgeEventActionType [{}], entityId [{}]", - tenantId, edgeId, edgeEventType, edgeEventActionType, entityId, e); - } + if (edgeEventType.equals(EdgeEventType.WIDGETS_BUNDLE) || edgeEventType.equals(EdgeEventType.WIDGET_TYPE)) { + TextPageData edgesByTenantId = edgeService.findEdgesByTenantId(tenantId, new TextPageLink(Integer.MAX_VALUE)); + if (edgesByTenantId != null && edgesByTenantId.getData() != null && !edgesByTenantId.getData().isEmpty()) { + for (Edge edge : edgesByTenantId.getData()) { + saveEdgeEvent(tenantId, edge.getId(), edgeEventType, edgeEventActionType, entityId, null); } } - return null; - }, dbCallbackExecutorService); + } else { + ListenableFuture> edgeIdsFuture = findRelatedEdgeIdsByEntityId(tenantId, entityId); + Futures.transform(edgeIdsFuture, edgeIds -> { + if (edgeIds != null && !edgeIds.isEmpty()) { + for (EdgeId edgeId : edgeIds) { + try { + saveEdgeEvent(tenantId, edgeId, edgeEventType, edgeEventActionType, entityId, null); + } catch (Exception e) { + log.error("[{}] Failed to push event to edge, edgeId [{}], edgeEventType [{}], edgeEventActionType [{}], entityId [{}]", + tenantId, edgeId, edgeEventType, edgeEventActionType, entityId, e); + } + } + } + return null; + }, dbCallbackExecutorService); + } break; case DELETED: TextPageData edgesByTenantId = edgeService.findEdgesByTenantId(tenantId, new TextPageLink(Integer.MAX_VALUE)); @@ -212,6 +270,9 @@ public class DefaultEdgeNotificationService implements EdgeNotificationService { case UNASSIGNED_FROM_EDGE: EdgeId edgeId = new EdgeId(new UUID(edgeNotificationMsg.getEdgeIdMSB(), edgeNotificationMsg.getEdgeIdLSB())); saveEdgeEvent(tenantId, edgeId, edgeEventType, edgeEventActionType, entityId, null); + if (edgeEventType.equals(EdgeEventType.RULE_CHAIN)) { + updateDependentRuleChains(tenantId, new RuleChainId(entityId.getId()), edgeId); + } break; case RELATIONS_DELETED: // TODO: voba - add support for relations deleted @@ -219,6 +280,40 @@ public class DefaultEdgeNotificationService implements EdgeNotificationService { } } + private void updateDependentRuleChains(TenantId tenantId, RuleChainId processingRuleChainId, EdgeId edgeId) { + ListenableFuture> future = ruleChainService.findRuleChainsByTenantIdAndEdgeId(tenantId, edgeId, new TimePageLink(Integer.MAX_VALUE)); + Futures.addCallback(future, new FutureCallback>() { + @Override + public void onSuccess(@Nullable TimePageData pageData) { + if (pageData != null && pageData.getData() != null && !pageData.getData().isEmpty()) { + for (RuleChain ruleChain : pageData.getData()) { + if (!ruleChain.getId().equals(processingRuleChainId)) { + List connectionInfos = + ruleChainService.loadRuleChainMetaData(ruleChain.getTenantId(), ruleChain.getId()).getRuleChainConnections(); + if (connectionInfos != null && !connectionInfos.isEmpty()) { + for (RuleChainConnectionInfo connectionInfo : connectionInfos) { + if (connectionInfo.getTargetRuleChainId().equals(processingRuleChainId)) { + saveEdgeEvent(tenantId, + edgeId, + EdgeEventType.RULE_CHAIN_METADATA, + ActionType.UPDATED, + ruleChain.getId(), + null); + } + } + } + } + } + } + } + + @Override + public void onFailure(Throwable t) { + log.error("Exception during updating dependent rule chains on sync!", t); + } + }, dbCallbackExecutorService); + } + private void processAlarm(TenantId tenantId, TransportProtos.EdgeNotificationMsgProto edgeNotificationMsg) { AlarmId alarmId = new AlarmId(new UUID(edgeNotificationMsg.getEntityIdMSB(), edgeNotificationMsg.getEntityIdLSB())); ListenableFuture alarmFuture = alarmService.findAlarmByIdAsync(tenantId, alarmId); diff --git a/application/src/main/java/org/thingsboard/server/service/edge/EdgeContextComponent.java b/application/src/main/java/org/thingsboard/server/service/edge/EdgeContextComponent.java index c21c1ebc9c..f5a97e2326 100644 --- a/application/src/main/java/org/thingsboard/server/service/edge/EdgeContextComponent.java +++ b/application/src/main/java/org/thingsboard/server/service/edge/EdgeContextComponent.java @@ -33,12 +33,16 @@ import org.thingsboard.server.dao.entityview.EntityViewService; import org.thingsboard.server.dao.relation.RelationService; import org.thingsboard.server.dao.rule.RuleChainService; import org.thingsboard.server.dao.user.UserService; +import org.thingsboard.server.dao.widget.WidgetTypeService; +import org.thingsboard.server.dao.widget.WidgetsBundleService; import org.thingsboard.server.queue.discovery.PartitionService; import org.thingsboard.server.queue.provider.TbQueueProducerProvider; import org.thingsboard.server.queue.util.TbCoreComponent; import org.thingsboard.server.service.edge.rpc.EdgeEventStorageSettings; +import org.thingsboard.server.service.edge.rpc.constructor.AdminSettingsUpdateMsgConstructor; import org.thingsboard.server.service.edge.rpc.constructor.AlarmUpdateMsgConstructor; import org.thingsboard.server.service.edge.rpc.constructor.AssetUpdateMsgConstructor; +import org.thingsboard.server.service.edge.rpc.constructor.CustomerUpdateMsgConstructor; import org.thingsboard.server.service.edge.rpc.constructor.DashboardUpdateMsgConstructor; import org.thingsboard.server.service.edge.rpc.constructor.DeviceUpdateMsgConstructor; import org.thingsboard.server.service.edge.rpc.constructor.EntityDataMsgConstructor; @@ -46,6 +50,8 @@ import org.thingsboard.server.service.edge.rpc.constructor.EntityViewUpdateMsgCo import org.thingsboard.server.service.edge.rpc.constructor.RelationUpdateMsgConstructor; import org.thingsboard.server.service.edge.rpc.constructor.RuleChainUpdateMsgConstructor; import org.thingsboard.server.service.edge.rpc.constructor.UserUpdateMsgConstructor; +import org.thingsboard.server.service.edge.rpc.constructor.WidgetTypeUpdateMsgConstructor; +import org.thingsboard.server.service.edge.rpc.constructor.WidgetsBundleUpdateMsgConstructor; import org.thingsboard.server.service.edge.rpc.init.SyncEdgeService; import org.thingsboard.server.service.executors.DbCallbackExecutorService; import org.thingsboard.server.service.queue.TbClusterService; @@ -119,6 +125,14 @@ public class EdgeContextComponent { @Autowired private ActorService actorService; + @Lazy + @Autowired + private WidgetsBundleService widgetsBundleService; + + @Lazy + @Autowired + private WidgetTypeService widgetTypeService; + @Lazy @Autowired private DeviceStateService deviceStateService; @@ -155,6 +169,10 @@ public class EdgeContextComponent { @Autowired private DashboardUpdateMsgConstructor dashboardUpdateMsgConstructor; + @Lazy + @Autowired + private CustomerUpdateMsgConstructor customerUpdateMsgConstructor; + @Lazy @Autowired private UserUpdateMsgConstructor userUpdateMsgConstructor; @@ -163,6 +181,18 @@ public class EdgeContextComponent { @Autowired private RelationUpdateMsgConstructor relationUpdateMsgConstructor; + @Lazy + @Autowired + private WidgetsBundleUpdateMsgConstructor widgetsBundleUpdateMsgConstructor; + + @Lazy + @Autowired + private WidgetTypeUpdateMsgConstructor widgetTypeUpdateMsgConstructor; + + @Lazy + @Autowired + private AdminSettingsUpdateMsgConstructor adminSettingsUpdateMsgConstructor; + @Lazy @Autowired private EntityDataMsgConstructor entityDataMsgConstructor; diff --git a/application/src/main/java/org/thingsboard/server/service/edge/rpc/EdgeGrpcSession.java b/application/src/main/java/org/thingsboard/server/service/edge/rpc/EdgeGrpcSession.java index 72014cf1c9..121772a01b 100644 --- a/application/src/main/java/org/thingsboard/server/service/edge/rpc/EdgeGrpcSession.java +++ b/application/src/main/java/org/thingsboard/server/service/edge/rpc/EdgeGrpcSession.java @@ -17,11 +17,14 @@ package org.thingsboard.server.service.edge.rpc; import com.datastax.driver.core.utils.UUIDs; import com.fasterxml.jackson.core.JsonProcessingException; +import com.fasterxml.jackson.databind.JsonNode; import com.fasterxml.jackson.databind.ObjectMapper; +import com.fasterxml.jackson.databind.node.NullNode; import com.fasterxml.jackson.databind.node.ObjectNode; import com.google.common.util.concurrent.FutureCallback; import com.google.common.util.concurrent.Futures; import com.google.common.util.concurrent.ListenableFuture; +import com.google.common.util.concurrent.MoreExecutors; import com.google.gson.Gson; import com.google.gson.JsonElement; import com.google.gson.JsonObject; @@ -30,6 +33,8 @@ import lombok.Data; import lombok.extern.slf4j.Slf4j; import org.apache.commons.lang.RandomStringUtils; import org.checkerframework.checker.nullness.qual.Nullable; +import org.thingsboard.server.common.data.AdminSettings; +import org.thingsboard.server.common.data.Customer; import org.thingsboard.server.common.data.Dashboard; import org.thingsboard.server.common.data.DataConstants; import org.thingsboard.server.common.data.Device; @@ -43,6 +48,8 @@ import org.thingsboard.server.common.data.asset.Asset; import org.thingsboard.server.common.data.audit.ActionType; import org.thingsboard.server.common.data.edge.Edge; import org.thingsboard.server.common.data.edge.EdgeEvent; +import org.thingsboard.server.common.data.exception.ThingsboardErrorCode; +import org.thingsboard.server.common.data.exception.ThingsboardException; import org.thingsboard.server.common.data.id.AlarmId; import org.thingsboard.server.common.data.id.AssetId; import org.thingsboard.server.common.data.id.CustomerId; @@ -50,10 +57,13 @@ import org.thingsboard.server.common.data.id.DashboardId; import org.thingsboard.server.common.data.id.DeviceId; import org.thingsboard.server.common.data.id.EdgeId; import org.thingsboard.server.common.data.id.EntityId; +import org.thingsboard.server.common.data.id.EntityIdFactory; import org.thingsboard.server.common.data.id.EntityViewId; import org.thingsboard.server.common.data.id.RuleChainId; import org.thingsboard.server.common.data.id.TenantId; import org.thingsboard.server.common.data.id.UserId; +import org.thingsboard.server.common.data.id.WidgetTypeId; +import org.thingsboard.server.common.data.id.WidgetsBundleId; import org.thingsboard.server.common.data.kv.AttributeKvEntry; import org.thingsboard.server.common.data.kv.BaseAttributeKvEntry; import org.thingsboard.server.common.data.kv.LongDataEntry; @@ -63,31 +73,37 @@ import org.thingsboard.server.common.data.relation.EntityRelation; import org.thingsboard.server.common.data.relation.RelationTypeGroup; import org.thingsboard.server.common.data.rule.RuleChain; import org.thingsboard.server.common.data.rule.RuleChainMetaData; +import org.thingsboard.server.common.data.security.Authority; import org.thingsboard.server.common.data.security.DeviceCredentials; import org.thingsboard.server.common.data.security.DeviceCredentialsType; import org.thingsboard.server.common.data.security.UserCredentials; +import org.thingsboard.server.common.data.widget.WidgetType; +import org.thingsboard.server.common.data.widget.WidgetsBundle; import org.thingsboard.server.common.msg.TbMsg; import org.thingsboard.server.common.msg.TbMsgMetaData; import org.thingsboard.server.common.msg.queue.ServiceType; import org.thingsboard.server.common.msg.queue.TopicPartitionInfo; import org.thingsboard.server.common.msg.session.SessionMsgType; import org.thingsboard.server.common.transport.util.JsonUtils; +import org.thingsboard.server.gen.edge.AdminSettingsUpdateMsg; import org.thingsboard.server.gen.edge.AlarmUpdateMsg; import org.thingsboard.server.gen.edge.AssetUpdateMsg; import org.thingsboard.server.gen.edge.AttributesRequestMsg; import org.thingsboard.server.gen.edge.ConnectRequestMsg; import org.thingsboard.server.gen.edge.ConnectResponseCode; import org.thingsboard.server.gen.edge.ConnectResponseMsg; +import org.thingsboard.server.gen.edge.CustomerUpdateMsg; import org.thingsboard.server.gen.edge.DashboardUpdateMsg; import org.thingsboard.server.gen.edge.DeviceCredentialsRequestMsg; import org.thingsboard.server.gen.edge.DeviceCredentialsUpdateMsg; import org.thingsboard.server.gen.edge.DeviceUpdateMsg; import org.thingsboard.server.gen.edge.DownlinkMsg; +import org.thingsboard.server.gen.edge.DownlinkResponseMsg; import org.thingsboard.server.gen.edge.EdgeConfiguration; import org.thingsboard.server.gen.edge.EntityDataProto; -import org.thingsboard.server.gen.edge.EntityUpdateMsg; import org.thingsboard.server.gen.edge.EntityViewUpdateMsg; import org.thingsboard.server.gen.edge.RelationRequestMsg; +import org.thingsboard.server.gen.edge.RelationUpdateMsg; import org.thingsboard.server.gen.edge.RequestMsg; import org.thingsboard.server.gen.edge.RequestMsgType; import org.thingsboard.server.gen.edge.ResponseMsg; @@ -99,6 +115,8 @@ import org.thingsboard.server.gen.edge.UplinkMsg; import org.thingsboard.server.gen.edge.UplinkResponseMsg; import org.thingsboard.server.gen.edge.UserCredentialsRequestMsg; import org.thingsboard.server.gen.edge.UserCredentialsUpdateMsg; +import org.thingsboard.server.gen.edge.WidgetTypeUpdateMsg; +import org.thingsboard.server.gen.edge.WidgetsBundleUpdateMsg; import org.thingsboard.server.gen.transport.TransportProtos; import org.thingsboard.server.queue.TbQueueCallback; import org.thingsboard.server.queue.TbQueueMsgMetadata; @@ -108,11 +126,14 @@ import org.thingsboard.server.service.edge.EdgeContextComponent; import java.io.Closeable; import java.io.IOException; +import java.util.ArrayList; import java.util.Collections; import java.util.List; import java.util.Optional; import java.util.UUID; +import java.util.concurrent.CountDownLatch; import java.util.concurrent.ExecutionException; +import java.util.concurrent.TimeUnit; import java.util.concurrent.locks.ReentrantLock; import java.util.function.BiConsumer; import java.util.function.Consumer; @@ -125,6 +146,8 @@ public final class EdgeGrpcSession implements Closeable { private static final ReentrantLock deviceCreationLock = new ReentrantLock(); + private static final ReentrantLock responseMsgLock = new ReentrantLock(); + private final Gson gson = new Gson(); private static final String QUEUE_START_TS_ATTR_KEY = "queueStartTs"; @@ -140,6 +163,8 @@ public final class EdgeGrpcSession implements Closeable { private StreamObserver outputStream; private boolean connected; + private CountDownLatch latch; + private TbQueueProducer> ruleEngineMsgProducer; EdgeGrpcSession(EdgeContextComponent ctx, StreamObserver outputStream, BiConsumer sessionOpenListener, @@ -160,7 +185,7 @@ public final class EdgeGrpcSession implements Closeable { public void onNext(RequestMsg requestMsg) { if (!connected && requestMsg.getMsgType().equals(RequestMsgType.CONNECT_RPC_MESSAGE)) { ConnectResponseMsg responseMsg = processConnect(requestMsg.getConnectRequestMsg()); - outputStream.onNext(ResponseMsg.newBuilder() + sendResponseMsg(ResponseMsg.newBuilder() .setConnectResponseMsg(responseMsg) .build()); if (ConnectResponseCode.ACCEPTED != responseMsg.getResponseCode()) { @@ -172,9 +197,10 @@ public final class EdgeGrpcSession implements Closeable { } if (connected) { if (requestMsg.getMsgType().equals(RequestMsgType.UPLINK_RPC_MESSAGE) && requestMsg.hasUplinkMsg()) { - outputStream.onNext(ResponseMsg.newBuilder() - .setUplinkResponseMsg(processUplinkMsg(requestMsg.getUplinkMsg())) - .build()); + onUplinkMsg(requestMsg.getUplinkMsg()); + } + if (requestMsg.getMsgType().equals(RequestMsgType.UPLINK_RPC_MESSAGE) && requestMsg.hasDownlinkResponseMsg()) { + onDownlinkResponse(requestMsg.getDownlinkResponseMsg()); } } } @@ -192,11 +218,56 @@ public final class EdgeGrpcSession implements Closeable { }; } + private void onUplinkMsg(UplinkMsg uplinkMsg) { + ListenableFuture> future = processUplinkMsg(uplinkMsg); + Futures.addCallback(future, new FutureCallback>() { + @Override + public void onSuccess(@Nullable List result) { + UplinkResponseMsg uplinkResponseMsg = UplinkResponseMsg.newBuilder().setSuccess(true).build(); + sendResponseMsg(ResponseMsg.newBuilder() + .setUplinkResponseMsg(uplinkResponseMsg) + .build()); + } + + @Override + public void onFailure(Throwable t) { + UplinkResponseMsg uplinkResponseMsg = UplinkResponseMsg.newBuilder().setSuccess(false).setErrorMsg(t.getMessage()).build(); + sendResponseMsg(ResponseMsg.newBuilder() + .setUplinkResponseMsg(uplinkResponseMsg) + .build()); + } + }, MoreExecutors.directExecutor()); + } + + private void onDownlinkResponse(DownlinkResponseMsg msg) { + try { + if (msg.getSuccess()) { + log.debug("[{}] Msg has been processed successfully! {}", edge.getRoutingKey(), msg); + } else { + log.error("[{}] Msg processing failed! Error msg: {}", edge.getRoutingKey(), msg.getErrorMsg()); + } + latch.countDown(); + } catch (Exception e) { + log.error("Can't process downlink response message [{}]", msg, e); + } + } + + private void sendResponseMsg(ResponseMsg responseMsg) { + if (isConnected()) { + try { + responseMsgLock.lock(); + outputStream.onNext(responseMsg); + } finally { + responseMsgLock.unlock(); + } + } + } + void onConfigurationUpdate(Edge edge) { try { this.edge = edge; // TODO: voba - push edge configuration update to edge -// outputStream.onNext(org.thingsboard.server.gen.integration.ResponseMsg.newBuilder() +// sendResponseMsg(org.thingsboard.server.gen.integration.ResponseMsg.newBuilder() // .setIntegrationUpdateMsg(IntegrationUpdateMsg.newBuilder() // .setConfiguration(constructIntegrationConfigProto(configuration, defaultConverterProto, downLinkConverterProto)) // .build()) @@ -211,46 +282,39 @@ public final class EdgeGrpcSession implements Closeable { TimePageLink pageLink = new TimePageLink(ctx.getEdgeEventStorageSettings().getMaxReadRecordsCount(), queueStartTs, null, true); TimePageData pageData; UUID ifOffset = null; + boolean success = true; do { pageData = ctx.getEdgeNotificationService().findEdgeEvents(edge.getTenantId(), edge.getId(), pageLink); if (isConnected() && !pageData.getData().isEmpty()) { log.trace("[{}] [{}] event(s) are going to be processed.", this.sessionId, pageData.getData().size()); - for (EdgeEvent edgeEvent : pageData.getData()) { - log.trace("[{}] Processing edge event [{}]", this.sessionId, edgeEvent); - try { - ActionType edgeEventAction = ActionType.valueOf(edgeEvent.getEdgeEventAction()); - switch (edgeEventAction) { - case UPDATED: - case ADDED: - case ASSIGNED_TO_EDGE: - case DELETED: - case UNASSIGNED_FROM_EDGE: - case ALARM_ACK: - case ALARM_CLEAR: - case CREDENTIALS_UPDATED: - processEntityMessage(edgeEvent, edgeEventAction); - break; - case ATTRIBUTES_UPDATED: - case ATTRIBUTES_DELETED: - case TIMESERIES_UPDATED: - processTelemetryMessage(edgeEvent); - break; - } - } catch (Exception e) { - log.error("Exception during processing records from queue", e); - } - ifOffset = edgeEvent.getUuidId(); + List downlinkMsgsPack = convertToDownlinkMsgsPack(pageData.getData()); + log.trace("[{}] downlink msg(s) are going to be send.", downlinkMsgsPack.size()); + + latch = new CountDownLatch(downlinkMsgsPack.size()); + for (DownlinkMsg downlinkMsg : downlinkMsgsPack) { + sendResponseMsg(ResponseMsg.newBuilder() + .setDownlinkMsg(downlinkMsg) + .build()); + } + + ifOffset = pageData.getData().get(pageData.getData().size() - 1).getUuidId(); + + success = latch.await(10, TimeUnit.SECONDS); + if (!success) { + log.warn("Failed to deliver the batch: {}", downlinkMsgsPack); } } - if (isConnected() && pageData.hasNext()) { - pageLink = pageData.getNextPageLink(); + if (isConnected() && (!success || pageData.hasNext())) { try { Thread.sleep(ctx.getEdgeEventStorageSettings().getSleepIntervalBetweenBatches()); } catch (InterruptedException e) { log.error("Error during sleep between batches", e); } + if (success) { + pageLink = pageData.getNextPageLink(); + } } - } while (isConnected() && pageData.hasNext()); + } while (isConnected() && (!success || pageData.hasNext())); if (ifOffset != null) { Long newStartTs = UUIDs.unixTimestamp(ifOffset); @@ -263,6 +327,42 @@ public final class EdgeGrpcSession implements Closeable { } } + private List convertToDownlinkMsgsPack(List edgeEvents) { + List result = new ArrayList<>(); + for (EdgeEvent edgeEvent : edgeEvents) { + log.trace("Processing edge event [{}]", edgeEvent); + try { + DownlinkMsg downlinkMsg = null; + ActionType edgeEventAction = ActionType.valueOf(edgeEvent.getEdgeEventAction()); + switch (edgeEventAction) { + case UPDATED: + case ADDED: + case ASSIGNED_TO_EDGE: + case DELETED: + case UNASSIGNED_FROM_EDGE: + case ALARM_ACK: + case ALARM_CLEAR: + case CREDENTIALS_UPDATED: + case RELATION_ADD_OR_UPDATE: + case RELATION_DELETED: + downlinkMsg = processEntityMessage(edgeEvent, edgeEventAction); + break; + case ATTRIBUTES_UPDATED: + case ATTRIBUTES_DELETED: + case TIMESERIES_UPDATED: + downlinkMsg = processTelemetryMessage(edgeEvent); + break; + } + if (downlinkMsg != null) { + result.add(downlinkMsg); + } + } catch (Exception e) { + log.error("Exception during processing records from queue", e); + } + } + return result; + } + private ListenableFuture getQueueStartTs() { ListenableFuture> future = ctx.getAttributesService().find(edge.getTenantId(), edge.getId(), DataConstants.SERVER_SCOPE, QUEUE_START_TS_ATTR_KEY); @@ -282,7 +382,7 @@ public final class EdgeGrpcSession implements Closeable { ctx.getAttributesService().save(edge.getTenantId(), edge.getId(), DataConstants.SERVER_SCOPE, attributes); } - private void processTelemetryMessage(EdgeEvent edgeEvent) throws IOException { + private DownlinkMsg processTelemetryMessage(EdgeEvent edgeEvent) throws IOException { log.trace("Executing processTelemetryMessage, edgeEvent [{}]", edgeEvent); EntityId entityId = null; switch (edgeEvent.getEdgeEventType()) { @@ -298,63 +398,68 @@ public final class EdgeGrpcSession implements Closeable { case DASHBOARD: entityId = new DashboardId(edgeEvent.getEntityId()); break; + case TENANT: + entityId = new TenantId(edgeEvent.getEntityId()); + break; + case CUSTOMER: + entityId = new CustomerId(edgeEvent.getEntityId()); + break; } + DownlinkMsg downlinkMsg = null; if (entityId != null) { log.debug("Sending telemetry data msg, entityId [{}], body [{}]", edgeEvent.getEntityId(), edgeEvent.getEntityBody()); - DownlinkMsg downlinkMsg; try { ActionType actionType = ActionType.valueOf(edgeEvent.getEdgeEventAction()); downlinkMsg = constructEntityDataProtoMsg(entityId, actionType, JsonUtils.parse(mapper.writeValueAsString(edgeEvent.getEntityBody()))); - outputStream.onNext(ResponseMsg.newBuilder() - .setDownlinkMsg(downlinkMsg) - .build()); } catch (Exception e) { log.warn("Can't send telemetry data msg, entityId [{}], body [{}]", edgeEvent.getEntityId(), edgeEvent.getEntityBody(), e); } - } + return downlinkMsg; } - private void processEntityMessage(EdgeEvent edgeEvent, ActionType edgeEventAction) { + private DownlinkMsg processEntityMessage(EdgeEvent edgeEvent, ActionType edgeEventAction) { UpdateMsgType msgType = getResponseMsgType(ActionType.valueOf(edgeEvent.getEdgeEventAction())); log.trace("Executing processEntityMessage, edgeEvent [{}], edgeEventAction [{}], msgType [{}]", edgeEvent, edgeEventAction, msgType); switch (edgeEvent.getEdgeEventType()) { case EDGE: // TODO: voba - add edge update logic - break; + return null; case DEVICE: - processDevice(edgeEvent, msgType, edgeEventAction); - break; + return processDevice(edgeEvent, msgType, edgeEventAction); case ASSET: - processAsset(edgeEvent, msgType, edgeEventAction); - break; + return processAsset(edgeEvent, msgType, edgeEventAction); case ENTITY_VIEW: - processEntityView(edgeEvent, msgType, edgeEventAction); - break; + return processEntityView(edgeEvent, msgType, edgeEventAction); case DASHBOARD: - processDashboard(edgeEvent, msgType, edgeEventAction); - break; + return processDashboard(edgeEvent, msgType, edgeEventAction); + case CUSTOMER: + return processCustomer(edgeEvent, msgType, edgeEventAction); case RULE_CHAIN: - processRuleChain(edgeEvent, msgType, edgeEventAction); - break; + return processRuleChain(edgeEvent, msgType, edgeEventAction); case RULE_CHAIN_METADATA: - processRuleChainMetadata(edgeEvent, msgType); - break; + return processRuleChainMetadata(edgeEvent, msgType); case ALARM: - processAlarm(edgeEvent, msgType); - break; + return processAlarm(edgeEvent, msgType); case USER: - processUser(edgeEvent, msgType, edgeEventAction); - break; + return processUser(edgeEvent, msgType, edgeEventAction); case RELATION: - processRelation(edgeEvent, msgType); - break; + return processRelation(edgeEvent, msgType); + case WIDGETS_BUNDLE: + return processWidgetsBundle(edgeEvent, msgType, edgeEventAction); + case WIDGET_TYPE: + return processWidgetType(edgeEvent, msgType, edgeEventAction); + case ADMIN_SETTINGS: + return processAdminSettings(edgeEvent); + default: + log.warn("Unsupported edge event type [{}]", edgeEvent); + return null; } } - private void processDevice(EdgeEvent edgeEvent, UpdateMsgType msgType, ActionType edgeActionType) { + private DownlinkMsg processDevice(EdgeEvent edgeEvent, UpdateMsgType msgType, ActionType edgeActionType) { DeviceId deviceId = new DeviceId(edgeEvent.getEntityId()); - EntityUpdateMsg entityUpdateMsg = null; + DownlinkMsg downlinkMsg = null; switch (edgeActionType) { case ADDED: case UPDATED: @@ -363,8 +468,8 @@ public final class EdgeGrpcSession implements Closeable { if (device != null) { DeviceUpdateMsg deviceUpdateMsg = ctx.getDeviceUpdateMsgConstructor().constructDeviceUpdatedMsg(msgType, device); - entityUpdateMsg = EntityUpdateMsg.newBuilder() - .setDeviceUpdateMsg(deviceUpdateMsg) + downlinkMsg = DownlinkMsg.newBuilder() + .addAllDeviceUpdateMsg(Collections.singletonList(deviceUpdateMsg)) .build(); } break; @@ -372,8 +477,8 @@ public final class EdgeGrpcSession implements Closeable { case UNASSIGNED_FROM_EDGE: DeviceUpdateMsg deviceUpdateMsg = ctx.getDeviceUpdateMsgConstructor().constructDeviceDeleteMsg(deviceId); - entityUpdateMsg = EntityUpdateMsg.newBuilder() - .setDeviceUpdateMsg(deviceUpdateMsg) + downlinkMsg = DownlinkMsg.newBuilder() + .addAllDeviceUpdateMsg(Collections.singletonList(deviceUpdateMsg)) .build(); break; case CREDENTIALS_UPDATED: @@ -381,22 +486,18 @@ public final class EdgeGrpcSession implements Closeable { if (deviceCredentials != null) { DeviceCredentialsUpdateMsg deviceCredentialsUpdateMsg = ctx.getDeviceUpdateMsgConstructor().constructDeviceCredentialsUpdatedMsg(deviceCredentials); - entityUpdateMsg = EntityUpdateMsg.newBuilder() - .setDeviceCredentialsUpdateMsg(deviceCredentialsUpdateMsg) + downlinkMsg = DownlinkMsg.newBuilder() + .addAllDeviceCredentialsUpdateMsg(Collections.singletonList(deviceCredentialsUpdateMsg)) .build(); } break; } - if (entityUpdateMsg != null) { - outputStream.onNext(ResponseMsg.newBuilder() - .setEntityUpdateMsg(entityUpdateMsg) - .build()); - } + return downlinkMsg; } - private void processAsset(EdgeEvent edgeEvent, UpdateMsgType msgType, ActionType edgeEventAction) { + private DownlinkMsg processAsset(EdgeEvent edgeEvent, UpdateMsgType msgType, ActionType edgeEventAction) { AssetId assetId = new AssetId(edgeEvent.getEntityId()); - EntityUpdateMsg entityUpdateMsg = null; + DownlinkMsg downlinkMsg = null; switch (edgeEventAction) { case ADDED: case UPDATED: @@ -405,8 +506,8 @@ public final class EdgeGrpcSession implements Closeable { if (asset != null) { AssetUpdateMsg assetUpdateMsg = ctx.getAssetUpdateMsgConstructor().constructAssetUpdatedMsg(msgType, asset); - entityUpdateMsg = EntityUpdateMsg.newBuilder() - .setAssetUpdateMsg(assetUpdateMsg) + downlinkMsg = DownlinkMsg.newBuilder() + .addAllAssetUpdateMsg(Collections.singletonList(assetUpdateMsg)) .build(); } break; @@ -414,21 +515,17 @@ public final class EdgeGrpcSession implements Closeable { case UNASSIGNED_FROM_EDGE: AssetUpdateMsg assetUpdateMsg = ctx.getAssetUpdateMsgConstructor().constructAssetDeleteMsg(assetId); - entityUpdateMsg = EntityUpdateMsg.newBuilder() - .setAssetUpdateMsg(assetUpdateMsg) + downlinkMsg = DownlinkMsg.newBuilder() + .addAllAssetUpdateMsg(Collections.singletonList(assetUpdateMsg)) .build(); break; } - if (entityUpdateMsg != null) { - outputStream.onNext(ResponseMsg.newBuilder() - .setEntityUpdateMsg(entityUpdateMsg) - .build()); - } + return downlinkMsg; } - private void processEntityView(EdgeEvent edgeEvent, UpdateMsgType msgType, ActionType edgeEventAction) { + private DownlinkMsg processEntityView(EdgeEvent edgeEvent, UpdateMsgType msgType, ActionType edgeEventAction) { EntityViewId entityViewId = new EntityViewId(edgeEvent.getEntityId()); - EntityUpdateMsg entityUpdateMsg = null; + DownlinkMsg downlinkMsg = null; switch (edgeEventAction) { case ADDED: case UPDATED: @@ -437,8 +534,8 @@ public final class EdgeGrpcSession implements Closeable { if (entityView != null) { EntityViewUpdateMsg entityViewUpdateMsg = ctx.getEntityViewUpdateMsgConstructor().constructEntityViewUpdatedMsg(msgType, entityView); - entityUpdateMsg = EntityUpdateMsg.newBuilder() - .setEntityViewUpdateMsg(entityViewUpdateMsg) + downlinkMsg = DownlinkMsg.newBuilder() + .addAllEntityViewUpdateMsg(Collections.singletonList(entityViewUpdateMsg)) .build(); } break; @@ -446,21 +543,17 @@ public final class EdgeGrpcSession implements Closeable { case UNASSIGNED_FROM_EDGE: EntityViewUpdateMsg entityViewUpdateMsg = ctx.getEntityViewUpdateMsgConstructor().constructEntityViewDeleteMsg(entityViewId); - entityUpdateMsg = EntityUpdateMsg.newBuilder() - .setEntityViewUpdateMsg(entityViewUpdateMsg) + downlinkMsg = DownlinkMsg.newBuilder() + .addAllEntityViewUpdateMsg(Collections.singletonList(entityViewUpdateMsg)) .build(); break; } - if (entityUpdateMsg != null) { - outputStream.onNext(ResponseMsg.newBuilder() - .setEntityUpdateMsg(entityUpdateMsg) - .build()); - } + return downlinkMsg; } - private void processDashboard(EdgeEvent edgeEvent, UpdateMsgType msgType, ActionType edgeEventAction) { + private DownlinkMsg processDashboard(EdgeEvent edgeEvent, UpdateMsgType msgType, ActionType edgeEventAction) { DashboardId dashboardId = new DashboardId(edgeEvent.getEntityId()); - EntityUpdateMsg entityUpdateMsg = null; + DownlinkMsg downlinkMsg = null; switch (edgeEventAction) { case ADDED: case UPDATED: @@ -469,8 +562,8 @@ public final class EdgeGrpcSession implements Closeable { if (dashboard != null) { DashboardUpdateMsg dashboardUpdateMsg = ctx.getDashboardUpdateMsgConstructor().constructDashboardUpdatedMsg(msgType, dashboard); - entityUpdateMsg = EntityUpdateMsg.newBuilder() - .setDashboardUpdateMsg(dashboardUpdateMsg) + downlinkMsg = DownlinkMsg.newBuilder() + .addAllDashboardUpdateMsg(Collections.singletonList(dashboardUpdateMsg)) .build(); } break; @@ -478,21 +571,43 @@ public final class EdgeGrpcSession implements Closeable { case UNASSIGNED_FROM_EDGE: DashboardUpdateMsg dashboardUpdateMsg = ctx.getDashboardUpdateMsgConstructor().constructDashboardDeleteMsg(dashboardId); - entityUpdateMsg = EntityUpdateMsg.newBuilder() - .setDashboardUpdateMsg(dashboardUpdateMsg) + downlinkMsg = DownlinkMsg.newBuilder() + .addAllDashboardUpdateMsg(Collections.singletonList(dashboardUpdateMsg)) .build(); break; } - if (entityUpdateMsg != null) { - outputStream.onNext(ResponseMsg.newBuilder() - .setEntityUpdateMsg(entityUpdateMsg) - .build()); + return downlinkMsg; + } + + private DownlinkMsg processCustomer(EdgeEvent edgeEvent, UpdateMsgType msgType, ActionType edgeEventAction) { + CustomerId customerId = new CustomerId(edgeEvent.getEntityId()); + DownlinkMsg downlinkMsg = null; + switch (edgeEventAction) { + case ADDED: + case UPDATED: + Customer customer = ctx.getCustomerService().findCustomerById(edgeEvent.getTenantId(), customerId); + if (customer != null) { + CustomerUpdateMsg customerUpdateMsg = + ctx.getCustomerUpdateMsgConstructor().constructCustomerUpdatedMsg(msgType, customer); + downlinkMsg = DownlinkMsg.newBuilder() + .addAllCustomerUpdateMsg(Collections.singletonList(customerUpdateMsg)) + .build(); + } + break; + case DELETED: + CustomerUpdateMsg customerUpdateMsg = + ctx.getCustomerUpdateMsgConstructor().constructCustomerDeleteMsg(customerId); + downlinkMsg = DownlinkMsg.newBuilder() + .addAllCustomerUpdateMsg(Collections.singletonList(customerUpdateMsg)) + .build(); + break; } + return downlinkMsg; } - private void processRuleChain(EdgeEvent edgeEvent, UpdateMsgType msgType, ActionType edgeEventAction) { + private DownlinkMsg processRuleChain(EdgeEvent edgeEvent, UpdateMsgType msgType, ActionType edgeEventAction) { RuleChainId ruleChainId = new RuleChainId(edgeEvent.getEntityId()); - EntityUpdateMsg entityUpdateMsg = null; + DownlinkMsg downlinkMsg = null; switch (edgeEventAction) { case ADDED: case UPDATED: @@ -501,105 +616,165 @@ public final class EdgeGrpcSession implements Closeable { if (ruleChain != null) { RuleChainUpdateMsg ruleChainUpdateMsg = ctx.getRuleChainUpdateMsgConstructor().constructRuleChainUpdatedMsg(edge.getRootRuleChainId(), msgType, ruleChain); - entityUpdateMsg = EntityUpdateMsg.newBuilder() - .setRuleChainUpdateMsg(ruleChainUpdateMsg) + downlinkMsg = DownlinkMsg.newBuilder() + .addAllRuleChainUpdateMsg(Collections.singletonList(ruleChainUpdateMsg)) .build(); } break; case DELETED: case UNASSIGNED_FROM_EDGE: - entityUpdateMsg = EntityUpdateMsg.newBuilder() - .setRuleChainUpdateMsg(ctx.getRuleChainUpdateMsgConstructor().constructRuleChainDeleteMsg(ruleChainId)) + downlinkMsg = DownlinkMsg.newBuilder() + .addAllRuleChainUpdateMsg(Collections.singletonList(ctx.getRuleChainUpdateMsgConstructor().constructRuleChainDeleteMsg(ruleChainId))) .build(); break; } - if (entityUpdateMsg != null) { - outputStream.onNext(ResponseMsg.newBuilder() - .setEntityUpdateMsg(entityUpdateMsg) - .build()); - } + return downlinkMsg; } - private void processRuleChainMetadata(EdgeEvent edgeEvent, UpdateMsgType msgType) { + private DownlinkMsg processRuleChainMetadata(EdgeEvent edgeEvent, UpdateMsgType msgType) { RuleChainId ruleChainId = new RuleChainId(edgeEvent.getEntityId()); RuleChain ruleChain = ctx.getRuleChainService().findRuleChainById(edgeEvent.getTenantId(), ruleChainId); + DownlinkMsg downlinkMsg = null; if (ruleChain != null) { RuleChainMetaData ruleChainMetaData = ctx.getRuleChainService().loadRuleChainMetaData(edgeEvent.getTenantId(), ruleChainId); RuleChainMetadataUpdateMsg ruleChainMetadataUpdateMsg = ctx.getRuleChainUpdateMsgConstructor().constructRuleChainMetadataUpdatedMsg(msgType, ruleChainMetaData); if (ruleChainMetadataUpdateMsg != null) { - EntityUpdateMsg entityUpdateMsg = EntityUpdateMsg.newBuilder() - .setRuleChainMetadataUpdateMsg(ruleChainMetadataUpdateMsg) + downlinkMsg = DownlinkMsg.newBuilder() + .addAllRuleChainMetadataUpdateMsg(Collections.singletonList(ruleChainMetadataUpdateMsg)) .build(); - outputStream.onNext(ResponseMsg.newBuilder() - .setEntityUpdateMsg(entityUpdateMsg) - .build()); } } + return downlinkMsg; } - private void processUser(EdgeEvent edgeEvent, UpdateMsgType msgType, ActionType edgeActionType) { + private DownlinkMsg processUser(EdgeEvent edgeEvent, UpdateMsgType msgType, ActionType edgeActionType) { UserId userId = new UserId(edgeEvent.getEntityId()); - EntityUpdateMsg entityUpdateMsg = null; + DownlinkMsg downlinkMsg = null; switch (edgeActionType) { case ADDED: case UPDATED: case ASSIGNED_TO_EDGE: User user = ctx.getUserService().findUserById(edgeEvent.getTenantId(), userId); if (user != null) { - entityUpdateMsg = EntityUpdateMsg.newBuilder() - .setUserUpdateMsg(ctx.getUserUpdateMsgConstructor().constructUserUpdatedMsg(msgType, user)) + boolean fullAccess = Authority.TENANT_ADMIN.equals(user.getAuthority()); + setFullAccess(user, fullAccess); + + downlinkMsg = DownlinkMsg.newBuilder() + .addAllUserUpdateMsg(Collections.singletonList(ctx.getUserUpdateMsgConstructor().constructUserUpdatedMsg(msgType, user))) .build(); } break; case DELETED: case UNASSIGNED_FROM_EDGE: - entityUpdateMsg = EntityUpdateMsg.newBuilder() - .setUserUpdateMsg(ctx.getUserUpdateMsgConstructor().constructUserDeleteMsg(userId)) + downlinkMsg = DownlinkMsg.newBuilder() + .addAllUserUpdateMsg(Collections.singletonList(ctx.getUserUpdateMsgConstructor().constructUserDeleteMsg(userId))) .build(); break; case CREDENTIALS_UPDATED: UserCredentials userCredentialsByUserId = ctx.getUserService().findUserCredentialsByUserId(edge.getTenantId(), userId); - if (userCredentialsByUserId != null) { + if (userCredentialsByUserId != null && userCredentialsByUserId.isEnabled()) { UserCredentialsUpdateMsg userCredentialsUpdateMsg = ctx.getUserUpdateMsgConstructor().constructUserCredentialsUpdatedMsg(userCredentialsByUserId); - entityUpdateMsg = EntityUpdateMsg.newBuilder() - .setUserCredentialsUpdateMsg(userCredentialsUpdateMsg) + downlinkMsg = DownlinkMsg.newBuilder() + .addAllUserCredentialsUpdateMsg(Collections.singletonList(userCredentialsUpdateMsg)) .build(); } } - if (entityUpdateMsg != null) { - outputStream.onNext(ResponseMsg.newBuilder() - .setEntityUpdateMsg(entityUpdateMsg) - .build()); + return downlinkMsg; + } + + private void setFullAccess(User user, boolean isFullAccess) { + JsonNode additionalInfo = user.getAdditionalInfo(); + if (additionalInfo == null || additionalInfo instanceof NullNode) { + additionalInfo = mapper.createObjectNode(); } + ((ObjectNode) additionalInfo).put("isFullAccess", isFullAccess); + user.setAdditionalInfo(additionalInfo); } - private void processRelation(EdgeEvent edgeEvent, UpdateMsgType msgType) { + private DownlinkMsg processRelation(EdgeEvent edgeEvent, UpdateMsgType msgType) { EntityRelation entityRelation = mapper.convertValue(edgeEvent.getEntityBody(), EntityRelation.class); - EntityUpdateMsg entityUpdateMsg = EntityUpdateMsg.newBuilder() - .setRelationUpdateMsg(ctx.getRelationUpdateMsgConstructor().constructRelationUpdatedMsg(msgType, entityRelation)) + RelationUpdateMsg r = ctx.getRelationUpdateMsgConstructor().constructRelationUpdatedMsg(msgType, entityRelation); + return DownlinkMsg.newBuilder() + .addAllRelationUpdateMsg(Collections.singletonList(r)) .build(); - outputStream.onNext(ResponseMsg.newBuilder() - .setEntityUpdateMsg(entityUpdateMsg) - .build()); } - private void processAlarm(EdgeEvent edgeEvent, UpdateMsgType msgType) { + private DownlinkMsg processAlarm(EdgeEvent edgeEvent, UpdateMsgType msgType) { + DownlinkMsg downlinkMsg = null; try { AlarmId alarmId = new AlarmId(edgeEvent.getEntityId()); Alarm alarm = ctx.getAlarmService().findAlarmByIdAsync(edgeEvent.getTenantId(), alarmId).get(); if (alarm != null) { - EntityUpdateMsg entityUpdateMsg = EntityUpdateMsg.newBuilder() - .setAlarmUpdateMsg(ctx.getAlarmUpdateMsgConstructor().constructAlarmUpdatedMsg(edge.getTenantId(), msgType, alarm)) + downlinkMsg = DownlinkMsg.newBuilder() + .addAllAlarmUpdateMsg(Collections.singletonList(ctx.getAlarmUpdateMsgConstructor().constructAlarmUpdatedMsg(edge.getTenantId(), msgType, alarm))) .build(); - outputStream.onNext(ResponseMsg.newBuilder() - .setEntityUpdateMsg(entityUpdateMsg) - .build()); } } catch (Exception e) { log.error("Can't process alarm msg [{}] [{}]", edgeEvent, msgType, e); } + return downlinkMsg; + } + + private DownlinkMsg processWidgetsBundle(EdgeEvent edgeEvent, UpdateMsgType msgType, ActionType edgeActionType) { + WidgetsBundleId widgetsBundleId = new WidgetsBundleId(edgeEvent.getEntityId()); + DownlinkMsg downlinkMsg = null; + switch (edgeActionType) { + case ADDED: + case UPDATED: + WidgetsBundle widgetsBundle = ctx.getWidgetsBundleService().findWidgetsBundleById(edgeEvent.getTenantId(), widgetsBundleId); + if (widgetsBundle != null) { + WidgetsBundleUpdateMsg widgetsBundleUpdateMsg = + ctx.getWidgetsBundleUpdateMsgConstructor().constructWidgetsBundleUpdateMsg(msgType, widgetsBundle); + downlinkMsg = DownlinkMsg.newBuilder() + .addAllWidgetsBundleUpdateMsg(Collections.singletonList(widgetsBundleUpdateMsg)) + .build(); + } + break; + case DELETED: + WidgetsBundleUpdateMsg widgetsBundleUpdateMsg = + ctx.getWidgetsBundleUpdateMsgConstructor().constructWidgetsBundleDeleteMsg(widgetsBundleId); + downlinkMsg = DownlinkMsg.newBuilder() + .addAllWidgetsBundleUpdateMsg(Collections.singletonList(widgetsBundleUpdateMsg)) + .build(); + break; + } + return downlinkMsg; + } + + private DownlinkMsg processWidgetType(EdgeEvent edgeEvent, UpdateMsgType msgType, ActionType edgeActionType) { + WidgetTypeId widgetTypeId = new WidgetTypeId(edgeEvent.getEntityId()); + DownlinkMsg downlinkMsg = null; + switch (edgeActionType) { + case ADDED: + case UPDATED: + WidgetType widgetType = ctx.getWidgetTypeService().findWidgetTypeById(edgeEvent.getTenantId(), widgetTypeId); + if (widgetType != null) { + WidgetTypeUpdateMsg widgetTypeUpdateMsg = + ctx.getWidgetTypeUpdateMsgConstructor().constructWidgetTypeUpdateMsg(msgType, widgetType); + downlinkMsg = DownlinkMsg.newBuilder() + .addAllWidgetTypeUpdateMsg(Collections.singletonList(widgetTypeUpdateMsg)) + .build(); + } + break; + case DELETED: + WidgetTypeUpdateMsg widgetTypeUpdateMsg = + ctx.getWidgetTypeUpdateMsgConstructor().constructWidgetTypeDeleteMsg(widgetTypeId); + downlinkMsg = DownlinkMsg.newBuilder() + .addAllWidgetTypeUpdateMsg(Collections.singletonList(widgetTypeUpdateMsg)) + .build(); + break; + } + return downlinkMsg; + } + + private DownlinkMsg processAdminSettings(EdgeEvent edgeEvent) { + AdminSettings adminSettings = mapper.convertValue(edgeEvent.getEntityBody(), AdminSettings.class); + AdminSettingsUpdateMsg t = ctx.getAdminSettingsUpdateMsgConstructor().constructAdminSettingsUpdateMsg(adminSettings); + return DownlinkMsg.newBuilder() + .addAllAdminSettingsUpdateMsg(Collections.singletonList(t)) + .build(); } private UpdateMsgType getResponseMsgType(ActionType actionType) { @@ -609,9 +784,11 @@ public final class EdgeGrpcSession implements Closeable { return UpdateMsgType.ENTITY_UPDATED_RPC_MESSAGE; case ADDED: case ASSIGNED_TO_EDGE: + case RELATION_ADD_OR_UPDATE: return ENTITY_CREATED_RPC_MESSAGE; case DELETED: case UNASSIGNED_FROM_EDGE: + case RELATION_DELETED: return UpdateMsgType.ENTITY_DELETED_RPC_MESSAGE; case ALARM_ACK: return UpdateMsgType.ALARM_ACK_RPC_MESSAGE; @@ -629,112 +806,106 @@ public final class EdgeGrpcSession implements Closeable { return builder.build(); } - private UplinkResponseMsg processUplinkMsg(UplinkMsg uplinkMsg) { + private ListenableFuture> processUplinkMsg(UplinkMsg uplinkMsg) { + List> result = new ArrayList<>(); try { if (uplinkMsg.getEntityDataList() != null && !uplinkMsg.getEntityDataList().isEmpty()) { for (EntityDataProto entityData : uplinkMsg.getEntityDataList()) { EntityId entityId = constructEntityId(entityData); if ((entityData.hasPostAttributesMsg() || entityData.hasPostTelemetryMsg()) && entityId != null) { - ListenableFuture metaDataFuture = constructBaseMsgMetadata(entityId); - Futures.transform(metaDataFuture, metaData -> { - if (metaData != null) { - metaData.putValue(DataConstants.MSG_SOURCE_KEY, DataConstants.EDGE_MSG_SOURCE); - if (entityData.hasPostAttributesMsg()) { - processPostAttributes(entityId, entityData.getPostAttributesMsg(), metaData); - } - if (entityData.hasPostTelemetryMsg()) { - processPostTelemetry(entityId, entityData.getPostTelemetryMsg(), metaData); - } - } - return null; - }, ctx.getDbCallbackExecutor()); + TbMsgMetaData metaData = constructBaseMsgMetadata(entityId); + metaData.putValue(DataConstants.MSG_SOURCE_KEY, DataConstants.EDGE_MSG_SOURCE); + if (entityData.hasPostAttributesMsg()) { + metaData.putValue("scope", entityData.getPostAttributeScope()); + result.add(processPostAttributes(entityId, entityData.getPostAttributesMsg(), metaData)); + } + if (entityData.hasPostTelemetryMsg()) { + result.add(processPostTelemetry(entityId, entityData.getPostTelemetryMsg(), metaData)); + } } } } if (uplinkMsg.getDeviceUpdateMsgList() != null && !uplinkMsg.getDeviceUpdateMsgList().isEmpty()) { for (DeviceUpdateMsg deviceUpdateMsg : uplinkMsg.getDeviceUpdateMsgList()) { - onDeviceUpdate(deviceUpdateMsg); + result.add(onDeviceUpdate(deviceUpdateMsg)); } } if (uplinkMsg.getDeviceCredentialsUpdateMsgList() != null && !uplinkMsg.getDeviceCredentialsUpdateMsgList().isEmpty()) { for (DeviceCredentialsUpdateMsg deviceCredentialsUpdateMsg : uplinkMsg.getDeviceCredentialsUpdateMsgList()) { - onDeviceCredentialsUpdate(deviceCredentialsUpdateMsg); + result.add(onDeviceCredentialsUpdate(deviceCredentialsUpdateMsg)); } } if (uplinkMsg.getAlarmUpdateMsgList() != null && !uplinkMsg.getAlarmUpdateMsgList().isEmpty()) { for (AlarmUpdateMsg alarmUpdateMsg : uplinkMsg.getAlarmUpdateMsgList()) { - onAlarmUpdate(alarmUpdateMsg); + result.add(onAlarmUpdate(alarmUpdateMsg)); + } + } + if (uplinkMsg.getRelationUpdateMsgList() != null && !uplinkMsg.getRelationUpdateMsgList().isEmpty()) { + for (RelationUpdateMsg relationUpdateMsg: uplinkMsg.getRelationUpdateMsgList()) { + onRelationUpdate(relationUpdateMsg); } } if (uplinkMsg.getRuleChainMetadataRequestMsgList() != null && !uplinkMsg.getRuleChainMetadataRequestMsgList().isEmpty()) { for (RuleChainMetadataRequestMsg ruleChainMetadataRequestMsg : uplinkMsg.getRuleChainMetadataRequestMsgList()) { - ctx.getSyncEdgeService().processRuleChainMetadataRequestMsg(edge, ruleChainMetadataRequestMsg); + result.add(ctx.getSyncEdgeService().processRuleChainMetadataRequestMsg(edge, ruleChainMetadataRequestMsg)); } } if (uplinkMsg.getAttributesRequestMsgList() != null && !uplinkMsg.getAttributesRequestMsgList().isEmpty()) { for (AttributesRequestMsg attributesRequestMsg : uplinkMsg.getAttributesRequestMsgList()) { - ctx.getSyncEdgeService().processAttributesRequestMsg(edge, attributesRequestMsg); + result.add(ctx.getSyncEdgeService().processAttributesRequestMsg(edge, attributesRequestMsg)); } } if (uplinkMsg.getRelationRequestMsgList() != null && !uplinkMsg.getRelationRequestMsgList().isEmpty()) { for (RelationRequestMsg relationRequestMsg : uplinkMsg.getRelationRequestMsgList()) { - ctx.getSyncEdgeService().processRelationRequestMsg(edge, relationRequestMsg); + result.add(ctx.getSyncEdgeService().processRelationRequestMsg(edge, relationRequestMsg)); } } if (uplinkMsg.getUserCredentialsRequestMsgList() != null && !uplinkMsg.getUserCredentialsRequestMsgList().isEmpty()) { for (UserCredentialsRequestMsg userCredentialsRequestMsg : uplinkMsg.getUserCredentialsRequestMsgList()) { - ctx.getSyncEdgeService().processUserCredentialsRequestMsg(edge, userCredentialsRequestMsg); + result.add(ctx.getSyncEdgeService().processUserCredentialsRequestMsg(edge, userCredentialsRequestMsg)); } } if (uplinkMsg.getDeviceCredentialsRequestMsgList() != null && !uplinkMsg.getDeviceCredentialsRequestMsgList().isEmpty()) { for (DeviceCredentialsRequestMsg deviceCredentialsRequestMsg : uplinkMsg.getDeviceCredentialsRequestMsgList()) { - ctx.getSyncEdgeService().processDeviceCredentialsRequestMsg(edge, deviceCredentialsRequestMsg); + result.add(ctx.getSyncEdgeService().processDeviceCredentialsRequestMsg(edge, deviceCredentialsRequestMsg)); } } } catch (Exception e) { - return UplinkResponseMsg.newBuilder().setSuccess(false).setErrorMsg(e.getMessage()).build(); + log.error("Can't process uplink msg [{}]", uplinkMsg, e); } - - return UplinkResponseMsg.newBuilder().setSuccess(true).build(); + return Futures.allAsList(result); } - private ListenableFuture constructBaseMsgMetadata(EntityId entityId) { + private TbMsgMetaData constructBaseMsgMetadata(EntityId entityId) { + TbMsgMetaData metaData = new TbMsgMetaData(); switch (entityId.getEntityType()) { case DEVICE: - ListenableFuture deviceFuture = ctx.getDeviceService().findDeviceByIdAsync(edge.getTenantId(), new DeviceId(entityId.getId())); - return Futures.transform(deviceFuture, device -> { - TbMsgMetaData metaData = new TbMsgMetaData(); - if (device != null) { - metaData.putValue("deviceName", device.getName()); - metaData.putValue("deviceType", device.getType()); - } - return metaData; - }, ctx.getDbCallbackExecutor()); + Device device = ctx.getDeviceService().findDeviceById(edge.getTenantId(), new DeviceId(entityId.getId())); + if (device != null) { + metaData.putValue("deviceName", device.getName()); + metaData.putValue("deviceType", device.getType()); + } + break; case ASSET: - ListenableFuture assetFuture = ctx.getAssetService().findAssetByIdAsync(edge.getTenantId(), new AssetId(entityId.getId())); - return Futures.transform(assetFuture, asset -> { - TbMsgMetaData metaData = new TbMsgMetaData(); - if (asset != null) { - metaData.putValue("assetName", asset.getName()); - metaData.putValue("assetType", asset.getType()); - } - return metaData; - }, ctx.getDbCallbackExecutor()); + Asset asset = ctx.getAssetService().findAssetById(edge.getTenantId(), new AssetId(entityId.getId())); + if (asset != null) { + metaData.putValue("assetName", asset.getName()); + metaData.putValue("assetType", asset.getType()); + } + break; case ENTITY_VIEW: - ListenableFuture entityViewFuture = ctx.getEntityViewService().findEntityViewByIdAsync(edge.getTenantId(), new EntityViewId(entityId.getId())); - return Futures.transform(entityViewFuture, entityView -> { - TbMsgMetaData metaData = new TbMsgMetaData(); - if (entityView != null) { - metaData.putValue("entityViewName", entityView.getName()); - metaData.putValue("entityViewType", entityView.getType()); - } - return metaData; - }, ctx.getDbCallbackExecutor()); + EntityView entityView = ctx.getEntityViewService().findEntityViewById(edge.getTenantId(), new EntityViewId(entityId.getId())); + if (entityView != null) { + metaData.putValue("entityViewName", entityView.getName()); + metaData.putValue("entityViewType", entityView.getType()); + } + break; default: - log.debug("Constructing empty metadata for entityId [{}]", entityId); - return Futures.immediateFuture(new TbMsgMetaData()); + log.debug("Using empty metadata for entityId [{}]", entityId); + break; } + return metaData; } private EntityId constructEntityId(EntityDataProto entityData) { @@ -748,13 +919,17 @@ public final class EdgeGrpcSession implements Closeable { return new EntityViewId(new UUID(entityData.getEntityIdMSB(), entityData.getEntityIdLSB())); case DASHBOARD: return new DashboardId(new UUID(entityData.getEntityIdMSB(), entityData.getEntityIdLSB())); + case TENANT: + return new TenantId(new UUID(entityData.getEntityIdMSB(), entityData.getEntityIdLSB())); + case CUSTOMER: + return new CustomerId(new UUID(entityData.getEntityIdMSB(), entityData.getEntityIdLSB())); default: log.warn("Unsupported entity type [{}] during construct of entity id. EntityDataProto [{}]", entityData.getEntityType(), entityData); return null; } } - private void processPostTelemetry(EntityId entityId, TransportProtos.PostTelemetryMsg msg, TbMsgMetaData metaData) { + private ListenableFuture processPostTelemetry(EntityId entityId, TransportProtos.PostTelemetryMsg msg, TbMsgMetaData metaData) { for (TransportProtos.TsKvListProto tsKv : msg.getTsKvListList()) { JsonObject json = JsonUtils.getJsonObject(tsKv.getKvList()); metaData.putValue("ts", tsKv.getTs() + ""); @@ -762,17 +937,18 @@ public final class EdgeGrpcSession implements Closeable { // TODO: voba - verify that null callback is OK ctx.getTbClusterService().pushMsgToRuleEngine(edge.getTenantId(), tbMsg.getOriginator(), tbMsg, null); } + return Futures.immediateFuture(null); } - private void processPostAttributes(EntityId entityId, TransportProtos.PostAttributeMsg msg, TbMsgMetaData metaData) { + private ListenableFuture processPostAttributes(EntityId entityId, TransportProtos.PostAttributeMsg msg, TbMsgMetaData metaData) { JsonObject json = JsonUtils.getJsonObject(msg.getKvList()); TbMsg tbMsg = TbMsg.newMsg(SessionMsgType.POST_ATTRIBUTES_REQUEST.name(), entityId, metaData, gson.toJson(json)); // TODO: voba - verify that null callback is OK ctx.getTbClusterService().pushMsgToRuleEngine(edge.getTenantId(), tbMsg.getOriginator(), tbMsg, null); + return Futures.immediateFuture(null); } - private void onDeviceUpdate(DeviceUpdateMsg deviceUpdateMsg) { - log.info("onDeviceUpdate {}", deviceUpdateMsg); + private ListenableFuture onDeviceUpdate(DeviceUpdateMsg deviceUpdateMsg) { DeviceId edgeDeviceId = new DeviceId(new UUID(deviceUpdateMsg.getIdMSB(), deviceUpdateMsg.getIdLSB())); switch (deviceUpdateMsg.getMsgType()) { case ENTITY_CREATED_RPC_MESSAGE: @@ -781,11 +957,12 @@ public final class EdgeGrpcSession implements Closeable { if (device != null) { // device with this name already exists on the cloud - update ID on the edge if (!device.getId().equals(edgeDeviceId)) { - EntityUpdateMsg entityUpdateMsg = EntityUpdateMsg.newBuilder() - .setDeviceUpdateMsg(ctx.getDeviceUpdateMsgConstructor().constructDeviceUpdatedMsg(UpdateMsgType.DEVICE_CONFLICT_RPC_MESSAGE, device)) + DeviceUpdateMsg d = ctx.getDeviceUpdateMsgConstructor().constructDeviceUpdatedMsg(UpdateMsgType.DEVICE_CONFLICT_RPC_MESSAGE, device); + DownlinkMsg downlinkMsg = DownlinkMsg.newBuilder() + .addAllDeviceUpdateMsg(Collections.singletonList(d)) .build(); - outputStream.onNext(ResponseMsg.newBuilder() - .setEntityUpdateMsg(entityUpdateMsg) + sendResponseMsg(ResponseMsg.newBuilder() + .setDownlinkMsg(downlinkMsg) .build()); } } else { @@ -793,11 +970,12 @@ public final class EdgeGrpcSession implements Closeable { if (deviceById != null) { // this ID already used by other device - create new device and update ID on the edge device = createDevice(deviceUpdateMsg); - EntityUpdateMsg entityUpdateMsg = EntityUpdateMsg.newBuilder() - .setDeviceUpdateMsg(ctx.getDeviceUpdateMsgConstructor().constructDeviceUpdatedMsg(UpdateMsgType.DEVICE_CONFLICT_RPC_MESSAGE, device)) + DeviceUpdateMsg d = ctx.getDeviceUpdateMsgConstructor().constructDeviceUpdatedMsg(UpdateMsgType.DEVICE_CONFLICT_RPC_MESSAGE, device); + DownlinkMsg downlinkMsg = DownlinkMsg.newBuilder() + .addAllDeviceUpdateMsg(Collections.singletonList(d)) .build(); - outputStream.onNext(ResponseMsg.newBuilder() - .setEntityUpdateMsg(entityUpdateMsg) + sendResponseMsg(ResponseMsg.newBuilder() + .setDownlinkMsg(downlinkMsg) .build()); } else { device = createDevice(deviceUpdateMsg); @@ -816,8 +994,10 @@ public final class EdgeGrpcSession implements Closeable { } break; case UNRECOGNIZED: - log.error("Unsupported msg type"); + log.error("Unsupported msg type {}", deviceUpdateMsg.getMsgType()); + return Futures.immediateFailedFuture(new RuntimeException("Unsupported msg type " + deviceUpdateMsg.getMsgType())); } + return Futures.immediateFuture(null); } private void updateDevice(DeviceUpdateMsg deviceUpdateMsg) { @@ -831,33 +1011,26 @@ public final class EdgeGrpcSession implements Closeable { requestDeviceCredentialsFromEdge(device); } - private void onDeviceCredentialsUpdate(DeviceCredentialsUpdateMsg deviceCredentialsUpdateMsg) { + private ListenableFuture onDeviceCredentialsUpdate(DeviceCredentialsUpdateMsg deviceCredentialsUpdateMsg) { log.debug("Executing onDeviceCredentialsUpdate, deviceCredentialsUpdateMsg [{}]", deviceCredentialsUpdateMsg); DeviceId deviceId = new DeviceId(new UUID(deviceCredentialsUpdateMsg.getDeviceIdMSB(), deviceCredentialsUpdateMsg.getDeviceIdLSB())); ListenableFuture deviceFuture = ctx.getDeviceService().findDeviceByIdAsync(edge.getTenantId(), deviceId); - - Futures.addCallback(deviceFuture, new FutureCallback() { - @Override - public void onSuccess(@Nullable Device device) { - if (device != null) { - log.debug("Updating device credentials for device [{}]. New device credentials Id [{}], value [{}]", - device.getName(), deviceCredentialsUpdateMsg.getCredentialsId(), deviceCredentialsUpdateMsg.getCredentialsValue()); - try { - DeviceCredentials deviceCredentials = ctx.getDeviceCredentialsService().findDeviceCredentialsByDeviceId(edge.getTenantId(), device.getId()); - deviceCredentials.setCredentialsType(DeviceCredentialsType.valueOf(deviceCredentialsUpdateMsg.getCredentialsType())); - deviceCredentials.setCredentialsId(deviceCredentialsUpdateMsg.getCredentialsId()); - deviceCredentials.setCredentialsValue(deviceCredentialsUpdateMsg.getCredentialsValue()); - ctx.getDeviceCredentialsService().updateDeviceCredentials(edge.getTenantId(), deviceCredentials); - } catch (Exception e) { - log.error("Can't update device credentials for device [{}], deviceCredentialsUpdateMsg [{}]", device.getName(), deviceCredentialsUpdateMsg, e); - } + return Futures.transform(deviceFuture, device -> { + if (device != null) { + log.debug("Updating device credentials for device [{}]. New device credentials Id [{}], value [{}]", + device.getName(), deviceCredentialsUpdateMsg.getCredentialsId(), deviceCredentialsUpdateMsg.getCredentialsValue()); + try { + DeviceCredentials deviceCredentials = ctx.getDeviceCredentialsService().findDeviceCredentialsByDeviceId(edge.getTenantId(), device.getId()); + deviceCredentials.setCredentialsType(DeviceCredentialsType.valueOf(deviceCredentialsUpdateMsg.getCredentialsType())); + deviceCredentials.setCredentialsId(deviceCredentialsUpdateMsg.getCredentialsId()); + deviceCredentials.setCredentialsValue(deviceCredentialsUpdateMsg.getCredentialsValue()); + ctx.getDeviceCredentialsService().updateDeviceCredentials(edge.getTenantId(), deviceCredentials); + } catch (Exception e) { + log.error("Can't update device credentials for device [{}], deviceCredentialsUpdateMsg [{}]", device.getName(), deviceCredentialsUpdateMsg, e); + throw new RuntimeException(e); } } - - @Override - public void onFailure(Throwable t) { - log.error("Can't update device credentials for deviceCredentialsUpdateMsg [{}]", deviceCredentialsUpdateMsg, t); - } + return null; }, ctx.getDbCallbackExecutor()); } @@ -865,7 +1038,7 @@ public final class EdgeGrpcSession implements Closeable { log.debug("Executing requestDeviceCredentialsFromEdge device [{}]", device); DownlinkMsg downlinkMsg = constructDeviceCredentialsRequestMsg(device.getId()); - outputStream.onNext(ResponseMsg.newBuilder() + sendResponseMsg(ResponseMsg.newBuilder() .setDownlinkMsg(downlinkMsg) .build()); } @@ -968,49 +1141,106 @@ public final class EdgeGrpcSession implements Closeable { } } - private void onAlarmUpdate(AlarmUpdateMsg alarmUpdateMsg) { + private ListenableFuture onAlarmUpdate(AlarmUpdateMsg alarmUpdateMsg) { EntityId originatorId = getAlarmOriginator(alarmUpdateMsg.getOriginatorName(), org.thingsboard.server.common.data.EntityType.valueOf(alarmUpdateMsg.getOriginatorType())); - if (originatorId != null) { - try { - Alarm existentAlarm = ctx.getAlarmService().findLatestByOriginatorAndType(edge.getTenantId(), originatorId, alarmUpdateMsg.getType()).get(); - switch (alarmUpdateMsg.getMsgType()) { - case ENTITY_CREATED_RPC_MESSAGE: - case ENTITY_UPDATED_RPC_MESSAGE: - if (existentAlarm == null || existentAlarm.getStatus().isCleared()) { - existentAlarm = new Alarm(); - existentAlarm.setTenantId(edge.getTenantId()); - existentAlarm.setType(alarmUpdateMsg.getName()); - existentAlarm.setOriginator(originatorId); - existentAlarm.setSeverity(AlarmSeverity.valueOf(alarmUpdateMsg.getSeverity())); - existentAlarm.setStartTs(alarmUpdateMsg.getStartTs()); - existentAlarm.setClearTs(alarmUpdateMsg.getClearTs()); - existentAlarm.setPropagate(alarmUpdateMsg.getPropagate()); - } - existentAlarm.setStatus(AlarmStatus.valueOf(alarmUpdateMsg.getStatus())); - existentAlarm.setAckTs(alarmUpdateMsg.getAckTs()); - existentAlarm.setEndTs(alarmUpdateMsg.getEndTs()); - existentAlarm.setDetails(mapper.readTree(alarmUpdateMsg.getDetails())); - ctx.getAlarmService().createOrUpdateAlarm(existentAlarm); - break; - case ALARM_ACK_RPC_MESSAGE: - if (existentAlarm != null) { - ctx.getAlarmService().ackAlarm(edge.getTenantId(), existentAlarm.getId(), alarmUpdateMsg.getAckTs()); - } - break; - case ALARM_CLEAR_RPC_MESSAGE: - if (existentAlarm != null) { - ctx.getAlarmService().clearAlarm(edge.getTenantId(), existentAlarm.getId(), mapper.readTree(alarmUpdateMsg.getDetails()), alarmUpdateMsg.getAckTs()); - } - break; - case ENTITY_DELETED_RPC_MESSAGE: - if (existentAlarm != null) { - ctx.getAlarmService().deleteAlarm(edge.getTenantId(), existentAlarm.getId()); - } - break; - } - } catch (Exception e) { - log.error("Error during finding existent alarm", e); + if (originatorId == null) { + return Futures.immediateFuture(null); + } + try { + Alarm existentAlarm = ctx.getAlarmService().findLatestByOriginatorAndType(edge.getTenantId(), originatorId, alarmUpdateMsg.getType()).get(); + switch (alarmUpdateMsg.getMsgType()) { + case ENTITY_CREATED_RPC_MESSAGE: + case ENTITY_UPDATED_RPC_MESSAGE: + if (existentAlarm == null || existentAlarm.getStatus().isCleared()) { + existentAlarm = new Alarm(); + existentAlarm.setTenantId(edge.getTenantId()); + existentAlarm.setType(alarmUpdateMsg.getName()); + existentAlarm.setOriginator(originatorId); + existentAlarm.setSeverity(AlarmSeverity.valueOf(alarmUpdateMsg.getSeverity())); + existentAlarm.setStartTs(alarmUpdateMsg.getStartTs()); + existentAlarm.setClearTs(alarmUpdateMsg.getClearTs()); + existentAlarm.setPropagate(alarmUpdateMsg.getPropagate()); + } + existentAlarm.setStatus(AlarmStatus.valueOf(alarmUpdateMsg.getStatus())); + existentAlarm.setAckTs(alarmUpdateMsg.getAckTs()); + existentAlarm.setEndTs(alarmUpdateMsg.getEndTs()); + existentAlarm.setDetails(mapper.readTree(alarmUpdateMsg.getDetails())); + ctx.getAlarmService().createOrUpdateAlarm(existentAlarm); + break; + case ALARM_ACK_RPC_MESSAGE: + if (existentAlarm != null) { + ctx.getAlarmService().ackAlarm(edge.getTenantId(), existentAlarm.getId(), alarmUpdateMsg.getAckTs()); + } + break; + case ALARM_CLEAR_RPC_MESSAGE: + if (existentAlarm != null) { + ctx.getAlarmService().clearAlarm(edge.getTenantId(), existentAlarm.getId(), mapper.readTree(alarmUpdateMsg.getDetails()), alarmUpdateMsg.getAckTs()); + } + break; + case ENTITY_DELETED_RPC_MESSAGE: + if (existentAlarm != null) { + ctx.getAlarmService().deleteAlarm(edge.getTenantId(), existentAlarm.getId()); + } + break; } + return Futures.immediateFuture(null); + } catch (Exception e) { + log.error("Error during finding existent alarm", e); + return Futures.immediateFailedFuture(new RuntimeException("Error during finding existent alarm", e)); + } + } + + private void onRelationUpdate(RelationUpdateMsg relationUpdateMsg) { + log.info("onRelationUpdate {}", relationUpdateMsg); + try { + EntityRelation entityRelation = new EntityRelation(); + + UUID fromUUID = new UUID(relationUpdateMsg.getFromIdMSB(), relationUpdateMsg.getFromIdLSB()); + EntityId fromId = EntityIdFactory.getByTypeAndUuid(EntityType.valueOf(relationUpdateMsg.getFromEntityType()), fromUUID); + entityRelation.setFrom(fromId); + + UUID toUUID = new UUID(relationUpdateMsg.getToIdMSB(), relationUpdateMsg.getToIdLSB()); + EntityId toId = EntityIdFactory.getByTypeAndUuid(EntityType.valueOf(relationUpdateMsg.getToEntityType()), toUUID); + entityRelation.setTo(toId); + + entityRelation.setType(relationUpdateMsg.getType()); + entityRelation.setTypeGroup(RelationTypeGroup.valueOf(relationUpdateMsg.getTypeGroup())); + entityRelation.setAdditionalInfo(mapper.readTree(relationUpdateMsg.getAdditionalInfo())); + switch (relationUpdateMsg.getMsgType()) { + case ENTITY_CREATED_RPC_MESSAGE: + case ENTITY_UPDATED_RPC_MESSAGE: + if (isEntityExists(edge.getTenantId(), entityRelation.getTo()) + && isEntityExists(edge.getTenantId(), entityRelation.getFrom())) { + ctx.getRelationService().saveRelationAsync(edge.getTenantId(), entityRelation); + } + break; + case ENTITY_DELETED_RPC_MESSAGE: + ctx.getRelationService().deleteRelation(edge.getTenantId(), entityRelation); + break; + case UNRECOGNIZED: + log.error("Unsupported msg type"); + } + } catch (Exception e) { + log.error("Error during relation update msg", e); + } + } + + private boolean isEntityExists(TenantId tenantId, EntityId entityId) throws ThingsboardException { + switch (entityId.getEntityType()) { + case DEVICE: + return ctx.getDeviceService().findDeviceById(tenantId, new DeviceId(entityId.getId())) != null; + case ASSET: + return ctx.getAssetService().findAssetById(tenantId, new AssetId(entityId.getId())) != null; + case ENTITY_VIEW: + return ctx.getEntityViewService().findEntityViewById(tenantId, new EntityViewId(entityId.getId())) != null; + case CUSTOMER: + return ctx.getCustomerService().findCustomerById(tenantId, new CustomerId(entityId.getId())) != null; + case USER: + return ctx.getUserService().findUserById(tenantId, new UserId(entityId.getId())) != null; + case DASHBOARD: + return ctx.getDashboardService().findDashboardById(tenantId, new DashboardId(entityId.getId())) != null; + default: + throw new ThingsboardException("Unsupported entity type " + entityId.getEntityType(), ThingsboardErrorCode.INVALID_ARGUMENTS); } } diff --git a/application/src/main/java/org/thingsboard/server/service/edge/rpc/constructor/AdminSettingsUpdateMsgConstructor.java b/application/src/main/java/org/thingsboard/server/service/edge/rpc/constructor/AdminSettingsUpdateMsgConstructor.java new file mode 100644 index 0000000000..1b0d188198 --- /dev/null +++ b/application/src/main/java/org/thingsboard/server/service/edge/rpc/constructor/AdminSettingsUpdateMsgConstructor.java @@ -0,0 +1,41 @@ +/** + * Copyright © 2016-2020 The Thingsboard Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.thingsboard.server.service.edge.rpc.constructor; + +import lombok.extern.slf4j.Slf4j; +import org.springframework.stereotype.Component; +import org.thingsboard.server.common.data.AdminSettings; +import org.thingsboard.server.common.data.id.AdminSettingsId; +import org.thingsboard.server.dao.util.mapping.JacksonUtil; +import org.thingsboard.server.gen.edge.AdminSettingsUpdateMsg; + +@Slf4j +@Component +public class AdminSettingsUpdateMsgConstructor { + + public AdminSettingsUpdateMsg constructAdminSettingsUpdateMsg(AdminSettings adminSettings) { + AdminSettingsUpdateMsg.Builder builder = AdminSettingsUpdateMsg.newBuilder() + .setKey(adminSettings.getKey()) + .setJsonValue(JacksonUtil.toString(adminSettings.getJsonValue())); + AdminSettingsId adminSettingsId = adminSettings.getId(); + if (adminSettingsId != null) { + builder.setIdMSB(adminSettingsId.getId().getMostSignificantBits()); + builder.setIdLSB(adminSettingsId.getId().getLeastSignificantBits()); + } + return builder.build(); + } + +} diff --git a/application/src/main/java/org/thingsboard/server/service/edge/rpc/constructor/CustomerUpdateMsgConstructor.java b/application/src/main/java/org/thingsboard/server/service/edge/rpc/constructor/CustomerUpdateMsgConstructor.java new file mode 100644 index 0000000000..a5f54fb443 --- /dev/null +++ b/application/src/main/java/org/thingsboard/server/service/edge/rpc/constructor/CustomerUpdateMsgConstructor.java @@ -0,0 +1,72 @@ +/** + * Copyright © 2016-2020 The Thingsboard Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.thingsboard.server.service.edge.rpc.constructor; + +import lombok.extern.slf4j.Slf4j; +import org.springframework.stereotype.Component; +import org.thingsboard.server.common.data.Customer; +import org.thingsboard.server.common.data.id.CustomerId; +import org.thingsboard.server.dao.util.mapping.JacksonUtil; +import org.thingsboard.server.gen.edge.CustomerUpdateMsg; +import org.thingsboard.server.gen.edge.UpdateMsgType; + +@Component +@Slf4j +public class CustomerUpdateMsgConstructor { + + public CustomerUpdateMsg constructCustomerUpdatedMsg(UpdateMsgType msgType, Customer customer) { + CustomerUpdateMsg.Builder builder = CustomerUpdateMsg.newBuilder() + .setMsgType(msgType) + .setIdMSB(customer.getId().getId().getMostSignificantBits()) + .setIdLSB(customer.getId().getId().getLeastSignificantBits()) + .setTitle(customer.getTitle()); + if (customer.getCountry() != null) { + builder.setCountry(customer.getCountry()); + } + if (customer.getState() != null) { + builder.setState(customer.getState()); + } + if (customer.getCity() != null) { + builder.setCity(customer.getCity()); + } + if (customer.getAddress() != null) { + builder.setAddress(customer.getAddress()); + } + if (customer.getAddress2() != null) { + builder.setAddress2(customer.getAddress2()); + } + if (customer.getZip() != null) { + builder.setZip(customer.getZip()); + } + if (customer.getPhone() != null) { + builder.setPhone(customer.getPhone()); + } + if (customer.getEmail() != null) { + builder.setEmail(customer.getEmail()); + } + if (customer.getAdditionalInfo() != null) { + builder.setAdditionalInfo(JacksonUtil.toString(customer.getAdditionalInfo())); + } + return builder.build(); + } + + public CustomerUpdateMsg constructCustomerDeleteMsg(CustomerId customerId) { + return CustomerUpdateMsg.newBuilder() + .setMsgType(UpdateMsgType.ENTITY_DELETED_RPC_MESSAGE) + .setIdMSB(customerId.getId().getMostSignificantBits()) + .setIdLSB(customerId.getId().getLeastSignificantBits()).build(); + } +} diff --git a/application/src/main/java/org/thingsboard/server/service/edge/rpc/constructor/EntityDataMsgConstructor.java b/application/src/main/java/org/thingsboard/server/service/edge/rpc/constructor/EntityDataMsgConstructor.java index 07f7ffb02a..c1d211af58 100644 --- a/application/src/main/java/org/thingsboard/server/service/edge/rpc/constructor/EntityDataMsgConstructor.java +++ b/application/src/main/java/org/thingsboard/server/service/edge/rpc/constructor/EntityDataMsgConstructor.java @@ -18,6 +18,7 @@ package org.thingsboard.server.service.edge.rpc.constructor; import com.google.gson.Gson; import com.google.gson.JsonArray; import com.google.gson.JsonElement; +import com.google.gson.JsonNull; import com.google.gson.JsonObject; import lombok.extern.slf4j.Slf4j; import org.springframework.stereotype.Component; @@ -41,7 +42,12 @@ public class EntityDataMsgConstructor { switch (actionType) { case TIMESERIES_UPDATED: try { - builder.setPostTelemetryMsg(JsonConverter.convertToTelemetryProto(entityData)); + JsonObject data = entityData.getAsJsonObject(); + long ts = System.currentTimeMillis(); + if (data.get("ts") != null && !data.get("ts").isJsonNull()) { + ts = data.getAsJsonObject("ts").getAsLong(); + } + builder.setPostTelemetryMsg(JsonConverter.convertToTelemetryProto(data.getAsJsonObject("data"), ts)); } catch (Exception e) { log.warn("Can't convert to telemetry proto, entityData [{}]", entityData, e); } diff --git a/application/src/main/java/org/thingsboard/server/service/edge/rpc/constructor/UserUpdateMsgConstructor.java b/application/src/main/java/org/thingsboard/server/service/edge/rpc/constructor/UserUpdateMsgConstructor.java index af42448e19..b50e0c2b5e 100644 --- a/application/src/main/java/org/thingsboard/server/service/edge/rpc/constructor/UserUpdateMsgConstructor.java +++ b/application/src/main/java/org/thingsboard/server/service/edge/rpc/constructor/UserUpdateMsgConstructor.java @@ -36,6 +36,10 @@ public class UserUpdateMsgConstructor { .setIdLSB(user.getId().getId().getLeastSignificantBits()) .setEmail(user.getEmail()) .setAuthority(user.getAuthority().name()); + if (user.getCustomerId() != null) { + builder.setCustomerIdMSB(user.getCustomerId().getId().getMostSignificantBits()); + builder.setCustomerIdLSB(user.getCustomerId().getId().getLeastSignificantBits()); + } if (user.getFirstName() != null) { builder.setFirstName(user.getFirstName()); } diff --git a/application/src/main/java/org/thingsboard/server/service/edge/rpc/constructor/WidgetTypeUpdateMsgConstructor.java b/application/src/main/java/org/thingsboard/server/service/edge/rpc/constructor/WidgetTypeUpdateMsgConstructor.java new file mode 100644 index 0000000000..784bfd6fdc --- /dev/null +++ b/application/src/main/java/org/thingsboard/server/service/edge/rpc/constructor/WidgetTypeUpdateMsgConstructor.java @@ -0,0 +1,61 @@ +/** + * Copyright © 2016-2020 The Thingsboard Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.thingsboard.server.service.edge.rpc.constructor; + +import lombok.extern.slf4j.Slf4j; +import org.springframework.stereotype.Component; +import org.thingsboard.server.common.data.id.TenantId; +import org.thingsboard.server.common.data.id.WidgetTypeId; +import org.thingsboard.server.common.data.widget.WidgetType; +import org.thingsboard.server.dao.util.mapping.JacksonUtil; +import org.thingsboard.server.gen.edge.UpdateMsgType; +import org.thingsboard.server.gen.edge.WidgetTypeUpdateMsg; + +@Component +@Slf4j +public class WidgetTypeUpdateMsgConstructor { + + public WidgetTypeUpdateMsg constructWidgetTypeUpdateMsg(UpdateMsgType msgType, WidgetType widgetType) { + WidgetTypeUpdateMsg.Builder builder = WidgetTypeUpdateMsg.newBuilder() + .setMsgType(msgType) + .setIdMSB(widgetType.getId().getId().getMostSignificantBits()) + .setIdLSB(widgetType.getId().getId().getLeastSignificantBits()); + if (widgetType.getBundleAlias() != null) { + builder.setBundleAlias(widgetType.getBundleAlias()); + } + if (widgetType.getAlias() != null) { + builder.setAlias(widgetType.getAlias()); + } + if (widgetType.getName() != null) { + builder.setName(widgetType.getName()); + } + if (widgetType.getDescriptor() != null) { + builder.setDescriptorJson(JacksonUtil.toString(widgetType.getDescriptor())); + } + if (widgetType.getTenantId().equals(TenantId.SYS_TENANT_ID)) { + builder.setIsSystem(true); + } + return builder.build(); + } + + public WidgetTypeUpdateMsg constructWidgetTypeDeleteMsg(WidgetTypeId widgetTypeId) { + return WidgetTypeUpdateMsg.newBuilder() + .setMsgType(UpdateMsgType.ENTITY_DELETED_RPC_MESSAGE) + .setIdMSB(widgetTypeId.getId().getMostSignificantBits()) + .setIdLSB(widgetTypeId.getId().getLeastSignificantBits()) + .build(); + } +} diff --git a/application/src/main/java/org/thingsboard/server/service/edge/rpc/constructor/WidgetsBundleUpdateMsgConstructor.java b/application/src/main/java/org/thingsboard/server/service/edge/rpc/constructor/WidgetsBundleUpdateMsgConstructor.java new file mode 100644 index 0000000000..4d26e0be59 --- /dev/null +++ b/application/src/main/java/org/thingsboard/server/service/edge/rpc/constructor/WidgetsBundleUpdateMsgConstructor.java @@ -0,0 +1,54 @@ +/** + * Copyright © 2016-2020 The Thingsboard Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.thingsboard.server.service.edge.rpc.constructor; + +import com.google.protobuf.ByteString; +import lombok.extern.slf4j.Slf4j; +import org.springframework.stereotype.Component; +import org.thingsboard.server.common.data.id.TenantId; +import org.thingsboard.server.common.data.id.WidgetsBundleId; +import org.thingsboard.server.common.data.widget.WidgetsBundle; +import org.thingsboard.server.gen.edge.UpdateMsgType; +import org.thingsboard.server.gen.edge.WidgetsBundleUpdateMsg; + +@Component +@Slf4j +public class WidgetsBundleUpdateMsgConstructor { + + public WidgetsBundleUpdateMsg constructWidgetsBundleUpdateMsg(UpdateMsgType msgType, WidgetsBundle widgetsBundle) { + WidgetsBundleUpdateMsg.Builder builder = WidgetsBundleUpdateMsg.newBuilder() + .setMsgType(msgType) + .setIdMSB(widgetsBundle.getId().getId().getMostSignificantBits()) + .setIdLSB(widgetsBundle.getId().getId().getLeastSignificantBits()) + .setTitle(widgetsBundle.getTitle()) + .setAlias(widgetsBundle.getAlias()); + if (widgetsBundle.getImage() != null) { + builder.setImage(ByteString.copyFrom(widgetsBundle.getImage())); + } + if (widgetsBundle.getTenantId().equals(TenantId.SYS_TENANT_ID)) { + builder.setIsSystem(true); + } + return builder.build(); + } + + public WidgetsBundleUpdateMsg constructWidgetsBundleDeleteMsg(WidgetsBundleId widgetsBundleId) { + return WidgetsBundleUpdateMsg.newBuilder() + .setMsgType(UpdateMsgType.ENTITY_DELETED_RPC_MESSAGE) + .setIdMSB(widgetsBundleId.getId().getMostSignificantBits()) + .setIdLSB(widgetsBundleId.getId().getLeastSignificantBits()) + .build(); + } +} diff --git a/application/src/main/java/org/thingsboard/server/service/edge/rpc/init/DefaultSyncEdgeService.java b/application/src/main/java/org/thingsboard/server/service/edge/rpc/init/DefaultSyncEdgeService.java index 1a2a3d3177..4d688f6999 100644 --- a/application/src/main/java/org/thingsboard/server/service/edge/rpc/init/DefaultSyncEdgeService.java +++ b/application/src/main/java/org/thingsboard/server/service/edge/rpc/init/DefaultSyncEdgeService.java @@ -25,6 +25,7 @@ import lombok.extern.slf4j.Slf4j; import org.checkerframework.checker.nullness.qual.Nullable; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.stereotype.Service; +import org.thingsboard.server.common.data.AdminSettings; import org.thingsboard.server.common.data.DashboardInfo; import org.thingsboard.server.common.data.DataConstants; import org.thingsboard.server.common.data.Device; @@ -54,6 +55,8 @@ import org.thingsboard.server.common.data.relation.EntityRelationsQuery; import org.thingsboard.server.common.data.relation.EntitySearchDirection; import org.thingsboard.server.common.data.relation.RelationsSearchParameters; import org.thingsboard.server.common.data.rule.RuleChain; +import org.thingsboard.server.common.data.widget.WidgetType; +import org.thingsboard.server.common.data.widget.WidgetsBundle; import org.thingsboard.server.dao.asset.AssetService; import org.thingsboard.server.dao.attributes.AttributesService; import org.thingsboard.server.dao.dashboard.DashboardService; @@ -62,7 +65,10 @@ import org.thingsboard.server.dao.edge.EdgeEventService; import org.thingsboard.server.dao.entityview.EntityViewService; import org.thingsboard.server.dao.relation.RelationService; import org.thingsboard.server.dao.rule.RuleChainService; +import org.thingsboard.server.dao.settings.AdminSettingsService; import org.thingsboard.server.dao.user.UserService; +import org.thingsboard.server.dao.widget.WidgetTypeService; +import org.thingsboard.server.dao.widget.WidgetsBundleService; import org.thingsboard.server.gen.edge.AttributesRequestMsg; import org.thingsboard.server.gen.edge.DeviceCredentialsRequestMsg; import org.thingsboard.server.gen.edge.RelationRequestMsg; @@ -109,6 +115,15 @@ public class DefaultSyncEdgeService implements SyncEdgeService { @Autowired private UserService userService; + @Autowired + private WidgetsBundleService widgetsBundleService; + + @Autowired + private WidgetTypeService widgetTypeService; + + @Autowired + private AdminSettingsService adminSettingsService; + @Autowired private DbCallbackExecutorService dbCallbackExecutorService; @@ -121,6 +136,8 @@ public class DefaultSyncEdgeService implements SyncEdgeService { syncAssets(edge); syncEntityViews(edge); syncDashboards(edge); + syncWidgetsBundleAndWidgetTypes(edge); + syncAdminSettings(edge); } catch (Exception e) { log.error("Exception during sync process", e); } @@ -253,6 +270,7 @@ public class DefaultSyncEdgeService implements SyncEdgeService { TextPageData pageData = userService.findTenantAdmins(edge.getTenantId(), new TextPageLink(Integer.MAX_VALUE)); pushUsersToEdge(pageData, edge); if (edge.getCustomerId() != null && !EntityId.NULL_UUID.equals(edge.getCustomerId().getId())) { + saveEdgeEvent(edge.getTenantId(), edge.getId(), EdgeEventType.CUSTOMER, ActionType.ADDED, edge.getCustomerId(), null); pageData = userService.findCustomerUsers(edge.getTenantId(), edge.getCustomerId(), new TextPageLink(Integer.MAX_VALUE)); pushUsersToEdge(pageData, edge); } @@ -261,6 +279,33 @@ public class DefaultSyncEdgeService implements SyncEdgeService { } } + private void syncWidgetsBundleAndWidgetTypes(Edge edge) { + List widgetsBundlesToPush = new ArrayList<>(); + List widgetTypesToPush = new ArrayList<>(); + widgetsBundlesToPush.addAll(widgetsBundleService.findAllTenantWidgetsBundlesByTenantId(edge.getTenantId())); + widgetsBundlesToPush.addAll(widgetsBundleService.findSystemWidgetsBundles(edge.getTenantId())); + try { + for (WidgetsBundle widgetsBundle: widgetsBundlesToPush) { + saveEdgeEvent(edge.getTenantId(), edge.getId(), EdgeEventType.WIDGETS_BUNDLE, ActionType.ADDED, widgetsBundle.getId(), null); + widgetTypesToPush.addAll(widgetTypeService.findWidgetTypesByTenantIdAndBundleAlias(widgetsBundle.getTenantId(), widgetsBundle.getAlias())); + } + for (WidgetType widgetType: widgetTypesToPush) { + saveEdgeEvent(edge.getTenantId(), edge.getId(), EdgeEventType.WIDGET_TYPE, ActionType.ADDED, widgetType.getId(), null); + } + } catch (Exception e) { + log.error("Exception during loading widgets bundle(s) and widget type(s) on sync!", e); + } + } + + private void syncAdminSettings(Edge edge) { + try { + AdminSettings mailSettings = adminSettingsService.findAdminSettingsByKey(TenantId.SYS_TENANT_ID, "mail"); + saveEdgeEvent(edge.getTenantId(), edge.getId(), EdgeEventType.ADMIN_SETTINGS, ActionType.UPDATED, null, mapper.valueToTree(mailSettings)); + } catch (Exception e) { + log.error("Can't load admin settings", e); + } + } + private void pushUsersToEdge(TextPageData pageData, Edge edge) { if (pageData != null && pageData.getData() != null && !pageData.getData().isEmpty()) { log.trace("[{}] [{}] user(s) are going to be pushed to edge.", edge.getId(), pageData.getData().size()); @@ -271,64 +316,62 @@ public class DefaultSyncEdgeService implements SyncEdgeService { } @Override - public void processRuleChainMetadataRequestMsg(Edge edge, RuleChainMetadataRequestMsg ruleChainMetadataRequestMsg) { + public ListenableFuture processRuleChainMetadataRequestMsg(Edge edge, RuleChainMetadataRequestMsg ruleChainMetadataRequestMsg) { if (ruleChainMetadataRequestMsg.getRuleChainIdMSB() != 0 && ruleChainMetadataRequestMsg.getRuleChainIdLSB() != 0) { RuleChainId ruleChainId = new RuleChainId(new UUID(ruleChainMetadataRequestMsg.getRuleChainIdMSB(), ruleChainMetadataRequestMsg.getRuleChainIdLSB())); - saveEdgeEvent(edge.getTenantId(), edge.getId(), EdgeEventType.RULE_CHAIN_METADATA, ActionType.ADDED, ruleChainId, null); + ListenableFuture future = saveEdgeEvent(edge.getTenantId(), edge.getId(), EdgeEventType.RULE_CHAIN_METADATA, ActionType.ADDED, ruleChainId, null); + return Futures.transform(future, edgeEvent -> null, dbCallbackExecutorService); } + return Futures.immediateFuture(null); } @Override - public void processAttributesRequestMsg(Edge edge, AttributesRequestMsg attributesRequestMsg) { + public ListenableFuture processAttributesRequestMsg(Edge edge, AttributesRequestMsg attributesRequestMsg) { EntityId entityId = EntityIdFactory.getByTypeAndUuid( EntityType.valueOf(attributesRequestMsg.getEntityType()), new UUID(attributesRequestMsg.getEntityIdMSB(), attributesRequestMsg.getEntityIdLSB())); final EdgeEventType edgeEventType = getEdgeQueueTypeByEntityType(entityId.getEntityType()); if (edgeEventType != null) { ListenableFuture> ssAttrFuture = attributesService.findAll(edge.getTenantId(), entityId, DataConstants.SERVER_SCOPE); - Futures.addCallback(ssAttrFuture, 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()); - } + return Futures.transform(ssAttrFuture, 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", DataConstants.SERVER_SCOPE); - JsonNode entityBody = mapper.valueToTree(entityData); - log.debug("Sending attributes data msg, entityId [{}], attributes [{}]", entityId, entityBody); - saveEdgeEvent(edge.getTenantId(), - edge.getId(), - edgeEventType, - ActionType.ATTRIBUTES_UPDATED, - entityId, - entityBody); - } catch (Exception e) { - log.error("[{}] Failed to send attribute updates to the edge", edge.getName(), e); } + entityData.put("kv", attributes); + entityData.put("scope", DataConstants.SERVER_SCOPE); + JsonNode entityBody = mapper.valueToTree(entityData); + log.debug("Sending attributes data msg, entityId [{}], attributes [{}]", entityId, entityBody); + saveEdgeEvent(edge.getTenantId(), + edge.getId(), + edgeEventType, + ActionType.ATTRIBUTES_UPDATED, + entityId, + entityBody); + } catch (Exception e) { + log.error("[{}] Failed to send attribute updates to the edge", edge.getName(), e); + throw new RuntimeException("[" + edge.getName() + "] Failed to send attribute updates to the edge", e); } } - - @Override - public void onFailure(Throwable t) { - - } + return null; }, dbCallbackExecutorService); // TODO: voba - push shared attributes to edge? - ListenableFuture> shAttrFuture = attributesService.findAll(edge.getTenantId(), entityId, DataConstants.SHARED_SCOPE); - ListenableFuture> clAttrFuture = attributesService.findAll(edge.getTenantId(), entityId, DataConstants.CLIENT_SCOPE); + // ListenableFuture> shAttrFuture = attributesService.findAll(edge.getTenantId(), entityId, DataConstants.SHARED_SCOPE); + // ListenableFuture> clAttrFuture = attributesService.findAll(edge.getTenantId(), entityId, DataConstants.CLIENT_SCOPE); + } else { + return Futures.immediateFuture(null); } } @@ -346,7 +389,7 @@ public class DefaultSyncEdgeService implements SyncEdgeService { } @Override - public void processRelationRequestMsg(Edge edge, RelationRequestMsg relationRequestMsg) { + public ListenableFuture processRelationRequestMsg(Edge edge, RelationRequestMsg relationRequestMsg) { EntityId entityId = EntityIdFactory.getByTypeAndUuid( EntityType.valueOf(relationRequestMsg.getEntityType()), new UUID(relationRequestMsg.getEntityIdMSB(), relationRequestMsg.getEntityIdLSB())); @@ -355,39 +398,33 @@ public class DefaultSyncEdgeService implements SyncEdgeService { futures.add(findRelationByQuery(edge, entityId, EntitySearchDirection.FROM)); futures.add(findRelationByQuery(edge, entityId, EntitySearchDirection.TO)); ListenableFuture>> relationsListFuture = Futures.allAsList(futures); - Futures.addCallback(relationsListFuture, new FutureCallback>>() { - @Override - public void onSuccess(@Nullable List> relationsList) { - try { - if (!relationsList.isEmpty()) { - 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(edge.getTenantId(), - edge.getId(), - EdgeEventType.RELATION, - ActionType.ADDED, - null, - mapper.valueToTree(relation)); - } - } catch (Exception e) { - log.error("Exception during loading relation [{}] to edge on sync!", relation, e); + return Futures.transform(relationsListFuture, relationsList -> { + try { + if (relationsList != null && !relationsList.isEmpty()) { + 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(edge.getTenantId(), + edge.getId(), + EdgeEventType.RELATION, + ActionType.ADDED, + null, + mapper.valueToTree(relation)); } + } catch (Exception e) { + log.error("Exception during loading relation [{}] to edge on sync!", relation, e); } } } - } catch (Exception e) { - log.error("Exception during loading relation(s) to edge on sync!", e); } + } catch (Exception e) { + log.error("Exception during loading relation(s) to edge on sync!", e); + throw new RuntimeException("Exception during loading relation(s) to edge on sync!", e); } - - @Override - public void onFailure(Throwable t) { - log.error("Exception during loading relation(s) to edge on sync!", t); - } + return null; }, dbCallbackExecutorService); } @@ -398,22 +435,26 @@ public class DefaultSyncEdgeService implements SyncEdgeService { } @Override - public void processDeviceCredentialsRequestMsg(Edge edge, DeviceCredentialsRequestMsg deviceCredentialsRequestMsg) { + public ListenableFuture processDeviceCredentialsRequestMsg(Edge edge, DeviceCredentialsRequestMsg deviceCredentialsRequestMsg) { if (deviceCredentialsRequestMsg.getDeviceIdMSB() != 0 && deviceCredentialsRequestMsg.getDeviceIdLSB() != 0) { DeviceId deviceId = new DeviceId(new UUID(deviceCredentialsRequestMsg.getDeviceIdMSB(), deviceCredentialsRequestMsg.getDeviceIdLSB())); - saveEdgeEvent(edge.getTenantId(), edge.getId(), EdgeEventType.DEVICE, ActionType.CREDENTIALS_UPDATED, deviceId, null); + ListenableFuture future = saveEdgeEvent(edge.getTenantId(), edge.getId(), EdgeEventType.DEVICE, ActionType.CREDENTIALS_UPDATED, deviceId, null); + return Futures.transform(future, edgeEvent -> null, dbCallbackExecutorService); } + return Futures.immediateFuture(null); } @Override - public void processUserCredentialsRequestMsg(Edge edge, UserCredentialsRequestMsg userCredentialsRequestMsg) { + public ListenableFuture processUserCredentialsRequestMsg(Edge edge, UserCredentialsRequestMsg userCredentialsRequestMsg) { if (userCredentialsRequestMsg.getUserIdMSB() != 0 && userCredentialsRequestMsg.getUserIdLSB() != 0) { UserId userId = new UserId(new UUID(userCredentialsRequestMsg.getUserIdMSB(), userCredentialsRequestMsg.getUserIdLSB())); - saveEdgeEvent(edge.getTenantId(), edge.getId(), EdgeEventType.USER, ActionType.CREDENTIALS_UPDATED, userId, null); + ListenableFuture future = saveEdgeEvent(edge.getTenantId(), edge.getId(), EdgeEventType.USER, ActionType.CREDENTIALS_UPDATED, userId, null); + return Futures.transform(future, edgeEvent -> null, dbCallbackExecutorService); } + return Futures.immediateFuture(null); } - private void saveEdgeEvent(TenantId tenantId, + private ListenableFuture saveEdgeEvent(TenantId tenantId, EdgeId edgeId, EdgeEventType edgeEventType, ActionType edgeEventAction, @@ -431,6 +472,6 @@ public class DefaultSyncEdgeService implements SyncEdgeService { edgeEvent.setEntityId(entityId.getId()); } edgeEvent.setEntityBody(entityBody); - edgeEventService.saveAsync(edgeEvent); + return edgeEventService.saveAsync(edgeEvent); } } diff --git a/application/src/main/java/org/thingsboard/server/service/edge/rpc/init/SyncEdgeService.java b/application/src/main/java/org/thingsboard/server/service/edge/rpc/init/SyncEdgeService.java index 051af4f6d2..a4df4e227c 100644 --- a/application/src/main/java/org/thingsboard/server/service/edge/rpc/init/SyncEdgeService.java +++ b/application/src/main/java/org/thingsboard/server/service/edge/rpc/init/SyncEdgeService.java @@ -15,6 +15,7 @@ */ package org.thingsboard.server.service.edge.rpc.init; +import com.google.common.util.concurrent.ListenableFuture; import org.thingsboard.server.common.data.edge.Edge; import org.thingsboard.server.gen.edge.AttributesRequestMsg; import org.thingsboard.server.gen.edge.DeviceCredentialsRequestMsg; @@ -26,13 +27,13 @@ public interface SyncEdgeService { void sync(Edge edge); - void processRuleChainMetadataRequestMsg(Edge edge, RuleChainMetadataRequestMsg ruleChainMetadataRequestMsg); + ListenableFuture processRuleChainMetadataRequestMsg(Edge edge, RuleChainMetadataRequestMsg ruleChainMetadataRequestMsg); - void processAttributesRequestMsg(Edge edge, AttributesRequestMsg attributesRequestMsg); + ListenableFuture processAttributesRequestMsg(Edge edge, AttributesRequestMsg attributesRequestMsg); - void processRelationRequestMsg(Edge edge, RelationRequestMsg relationRequestMsg); + ListenableFuture processRelationRequestMsg(Edge edge, RelationRequestMsg relationRequestMsg); - void processDeviceCredentialsRequestMsg(Edge edge, DeviceCredentialsRequestMsg deviceCredentialsRequestMsg); + ListenableFuture processDeviceCredentialsRequestMsg(Edge edge, DeviceCredentialsRequestMsg deviceCredentialsRequestMsg); - void processUserCredentialsRequestMsg(Edge edge, UserCredentialsRequestMsg userCredentialsRequestMsg); + ListenableFuture processUserCredentialsRequestMsg(Edge edge, UserCredentialsRequestMsg userCredentialsRequestMsg); } diff --git a/application/src/main/java/org/thingsboard/server/service/ttl/edge/EdgeEventsCleanUpService.java b/application/src/main/java/org/thingsboard/server/service/ttl/edge/EdgeEventsCleanUpService.java new file mode 100644 index 0000000000..308feb7304 --- /dev/null +++ b/application/src/main/java/org/thingsboard/server/service/ttl/edge/EdgeEventsCleanUpService.java @@ -0,0 +1,56 @@ +/** + * Copyright © 2016-2020 The Thingsboard Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.thingsboard.server.service.ttl.edge; + +import lombok.extern.slf4j.Slf4j; +import org.springframework.beans.factory.annotation.Value; +import org.springframework.scheduling.annotation.Scheduled; +import org.springframework.stereotype.Service; +import org.thingsboard.server.dao.util.PsqlDao; +import org.thingsboard.server.service.ttl.AbstractCleanUpService; + +import java.sql.Connection; +import java.sql.DriverManager; +import java.sql.SQLException; + +@PsqlDao +@Slf4j +@Service +public class EdgeEventsCleanUpService extends AbstractCleanUpService { + + @Value("${sql.ttl.edge_events.edge_events_ttl}") + private long ttl; + + @Value("${sql.ttl.edge_events.enabled}") + private boolean ttlTaskExecutionEnabled; + + @Scheduled(initialDelayString = "${sql.ttl.edge_events.execution_interval_ms}", fixedDelayString = "${sql.ttl.edge_events.execution_interval_ms}") + public void cleanUp() { + if (ttlTaskExecutionEnabled) { + try (Connection conn = DriverManager.getConnection(dbUrl, dbUserName, dbPassword)) { + doCleanUp(conn); + } catch (SQLException e) { + log.error("SQLException occurred during TTL task execution ", e); + } + } + } + + @Override + protected void doCleanUp(Connection connection) { + long totalEdgeEventsRemoved = executeQuery(connection, "call cleanup_edge_events_by_ttl(" + ttl + ", 0);"); + log.info("Total edge events removed by TTL: [{}]", totalEdgeEventsRemoved); + } +} \ No newline at end of file diff --git a/application/src/main/resources/thingsboard.yml b/application/src/main/resources/thingsboard.yml index 3152d9939e..08817dd9e3 100644 --- a/application/src/main/resources/thingsboard.yml +++ b/application/src/main/resources/thingsboard.yml @@ -282,6 +282,10 @@ sql: execution_interval_ms: "${SQL_TTL_EVENTS_EXECUTION_INTERVAL:86400000}" # Number of milliseconds. The current value corresponds to one day events_ttl: "${SQL_TTL_EVENTS_EVENTS_TTL:0}" # Number of seconds debug_events_ttl: "${SQL_TTL_EVENTS_DEBUG_EVENTS_TTL:604800}" # Number of seconds. The current value corresponds to one week + edge_events: + enabled: "${SQL_TTL_EDGE_EVENTS_ENABLED:true}" + execution_interval_ms: "${SQL_TTL_EDGE_EVENTS_EXECUTION_INTERVAL:86400000}" # Number of milliseconds. The current value corresponds to one day + edge_events_ttl: "${SQL_TTL_EDGE_EVENTS_TTL:2628000}" # Number of seconds. The current value corresponds to one month # Actor system parameters actors: diff --git a/common/data/src/main/java/org/thingsboard/server/common/data/EdgeUtils.java b/common/data/src/main/java/org/thingsboard/server/common/data/EdgeUtils.java index 784763f99e..2394ea0047 100644 --- a/common/data/src/main/java/org/thingsboard/server/common/data/EdgeUtils.java +++ b/common/data/src/main/java/org/thingsboard/server/common/data/EdgeUtils.java @@ -16,9 +16,6 @@ package org.thingsboard.server.common.data; import org.thingsboard.server.common.data.edge.EdgeEventType; -import org.thingsboard.server.common.data.id.EdgeId; - -import java.util.Set; public final class EdgeUtils { @@ -37,8 +34,18 @@ public final class EdgeUtils { return EdgeEventType.DASHBOARD; case USER: return EdgeEventType.USER; + case RULE_CHAIN: + return EdgeEventType.RULE_CHAIN; case ALARM: return EdgeEventType.ALARM; + case TENANT: + return EdgeEventType.TENANT; + case CUSTOMER: + return EdgeEventType.CUSTOMER; + case WIDGETS_BUNDLE: + return EdgeEventType.WIDGETS_BUNDLE; + case WIDGET_TYPE: + return EdgeEventType.WIDGET_TYPE; default: return null; } diff --git a/common/data/src/main/java/org/thingsboard/server/common/data/edge/EdgeEventType.java b/common/data/src/main/java/org/thingsboard/server/common/data/edge/EdgeEventType.java index 52259d4507..cc295546a8 100644 --- a/common/data/src/main/java/org/thingsboard/server/common/data/edge/EdgeEventType.java +++ b/common/data/src/main/java/org/thingsboard/server/common/data/edge/EdgeEventType.java @@ -26,5 +26,9 @@ public enum EdgeEventType { EDGE, USER, CUSTOMER, - RELATION + RELATION, + TENANT, + WIDGETS_BUNDLE, + WIDGET_TYPE, + ADMIN_SETTINGS } diff --git a/common/data/src/main/java/org/thingsboard/server/common/data/id/EntityIdFactory.java b/common/data/src/main/java/org/thingsboard/server/common/data/id/EntityIdFactory.java index 17d86e26ec..9297ed5e0b 100644 --- a/common/data/src/main/java/org/thingsboard/server/common/data/id/EntityIdFactory.java +++ b/common/data/src/main/java/org/thingsboard/server/common/data/id/EntityIdFactory.java @@ -87,6 +87,10 @@ public class EntityIdFactory { return new RuleChainId(uuid); case ENTITY_VIEW: return new EntityViewId(uuid); + case WIDGETS_BUNDLE: + return new WidgetsBundleId(uuid); + case WIDGET_TYPE: + return new WidgetTypeId(uuid); case EDGE: return new EdgeId(uuid); } diff --git a/common/edge-api/src/main/java/org/thingsboard/edge/rpc/EdgeGrpcClient.java b/common/edge-api/src/main/java/org/thingsboard/edge/rpc/EdgeGrpcClient.java index 917cc3fa02..45201d3229 100644 --- a/common/edge-api/src/main/java/org/thingsboard/edge/rpc/EdgeGrpcClient.java +++ b/common/edge-api/src/main/java/org/thingsboard/edge/rpc/EdgeGrpcClient.java @@ -28,9 +28,9 @@ import org.thingsboard.server.gen.edge.ConnectRequestMsg; import org.thingsboard.server.gen.edge.ConnectResponseCode; import org.thingsboard.server.gen.edge.ConnectResponseMsg; import org.thingsboard.server.gen.edge.DownlinkMsg; +import org.thingsboard.server.gen.edge.DownlinkResponseMsg; import org.thingsboard.server.gen.edge.EdgeConfiguration; import org.thingsboard.server.gen.edge.EdgeRpcServiceGrpc; -import org.thingsboard.server.gen.edge.EntityUpdateMsg; import org.thingsboard.server.gen.edge.RequestMsg; import org.thingsboard.server.gen.edge.RequestMsgType; import org.thingsboard.server.gen.edge.ResponseMsg; @@ -41,6 +41,7 @@ import javax.net.ssl.SSLException; import java.io.File; import java.net.URISyntaxException; import java.util.concurrent.TimeUnit; +import java.util.concurrent.locks.ReentrantLock; import java.util.function.Consumer; @Service @@ -62,12 +63,13 @@ public class EdgeGrpcClient implements EdgeRpcClient { private StreamObserver inputStream; + private static final ReentrantLock uplinkMsgLock = new ReentrantLock(); + @Override public void connect(String edgeKey, String edgeSecret, Consumer onUplinkResponse, Consumer onEdgeUpdate, - Consumer onEntityUpdate, Consumer onDownlink, Consumer onError) { NettyChannelBuilder builder = NettyChannelBuilder.forAddress(rpcHost, rpcPort).usePlaintext(); @@ -83,7 +85,7 @@ public class EdgeGrpcClient implements EdgeRpcClient { channel = builder.build(); EdgeRpcServiceGrpc.EdgeRpcServiceStub stub = EdgeRpcServiceGrpc.newStub(channel); log.info("[{}] Sending a connect request to the TB!", edgeKey); - this.inputStream = stub.handleMsgs(initOutputStream(edgeKey, onUplinkResponse, onEdgeUpdate, onEntityUpdate, onDownlink, onError)); + this.inputStream = stub.handleMsgs(initOutputStream(edgeKey, onUplinkResponse, onEdgeUpdate, onDownlink, onError)); this.inputStream.onNext(RequestMsg.newBuilder() .setMsgType(RequestMsgType.CONNECT_RPC_MESSAGE) .setConnectRequestMsg(ConnectRequestMsg.newBuilder().setEdgeRoutingKey(edgeKey).setEdgeSecret(edgeSecret).build()) @@ -110,16 +112,33 @@ public class EdgeGrpcClient implements EdgeRpcClient { @Override public void sendUplinkMsg(UplinkMsg msg) { - this.inputStream.onNext(RequestMsg.newBuilder() - .setMsgType(RequestMsgType.UPLINK_RPC_MESSAGE) - .setUplinkMsg(msg) - .build()); + try { + uplinkMsgLock.lock(); + this.inputStream.onNext(RequestMsg.newBuilder() + .setMsgType(RequestMsgType.UPLINK_RPC_MESSAGE) + .setUplinkMsg(msg) + .build()); + } finally { + uplinkMsgLock.unlock(); + } + } + + @Override + public void sendDownlinkResponseMsg(DownlinkResponseMsg downlinkResponseMsg) { + try { + uplinkMsgLock.lock(); + this.inputStream.onNext(RequestMsg.newBuilder() + .setMsgType(RequestMsgType.UPLINK_RPC_MESSAGE) + .setDownlinkResponseMsg(downlinkResponseMsg) + .build()); + } finally { + uplinkMsgLock.unlock(); + } } private StreamObserver initOutputStream(String edgeKey, Consumer onUplinkResponse, Consumer onEdgeUpdate, - Consumer onEntityUpdate, Consumer onDownlink, Consumer onError) { return new StreamObserver() { @@ -137,9 +156,6 @@ public class EdgeGrpcClient implements EdgeRpcClient { } else if (responseMsg.hasUplinkResponseMsg()) { log.debug("[{}] Uplink response message received {}", edgeKey, responseMsg.getUplinkResponseMsg()); onUplinkResponse.accept(responseMsg.getUplinkResponseMsg()); - } else if (responseMsg.hasEntityUpdateMsg()) { - log.debug("[{}] Entity update message received {}", edgeKey, responseMsg.getEntityUpdateMsg()); - onEntityUpdate.accept(responseMsg.getEntityUpdateMsg()); } else if (responseMsg.hasDownlinkMsg()) { log.debug("[{}] Downlink message received {}", edgeKey, responseMsg.getDownlinkMsg()); onDownlink.accept(responseMsg.getDownlinkMsg()); diff --git a/common/edge-api/src/main/java/org/thingsboard/edge/rpc/EdgeRpcClient.java b/common/edge-api/src/main/java/org/thingsboard/edge/rpc/EdgeRpcClient.java index d2ab91dfaf..c268b2a9e9 100644 --- a/common/edge-api/src/main/java/org/thingsboard/edge/rpc/EdgeRpcClient.java +++ b/common/edge-api/src/main/java/org/thingsboard/edge/rpc/EdgeRpcClient.java @@ -16,8 +16,8 @@ package org.thingsboard.edge.rpc; import org.thingsboard.server.gen.edge.DownlinkMsg; +import org.thingsboard.server.gen.edge.DownlinkResponseMsg; import org.thingsboard.server.gen.edge.EdgeConfiguration; -import org.thingsboard.server.gen.edge.EntityUpdateMsg; import org.thingsboard.server.gen.edge.UplinkMsg; import org.thingsboard.server.gen.edge.UplinkResponseMsg; @@ -29,11 +29,12 @@ public interface EdgeRpcClient { String integrationSecret, Consumer onUplinkResponse, Consumer onEdgeUpdate, - Consumer onEntityUpdate, Consumer onDownlink, Consumer onError); void disconnect() throws InterruptedException; - void sendUplinkMsg(UplinkMsg uplinkMsg) throws InterruptedException; + void sendUplinkMsg(UplinkMsg uplinkMsg); + + void sendDownlinkResponseMsg(DownlinkResponseMsg downlinkResponseMsg); } diff --git a/common/edge-api/src/main/proto/edge.proto b/common/edge-api/src/main/proto/edge.proto index 181ebff531..7827d26538 100644 --- a/common/edge-api/src/main/proto/edge.proto +++ b/common/edge-api/src/main/proto/edge.proto @@ -37,28 +37,13 @@ message RequestMsg { RequestMsgType msgType = 1; ConnectRequestMsg connectRequestMsg = 2; UplinkMsg uplinkMsg = 3; + DownlinkResponseMsg downlinkResponseMsg = 4; } message ResponseMsg { ConnectResponseMsg connectResponseMsg = 1; UplinkResponseMsg uplinkResponseMsg = 2; - EntityUpdateMsg entityUpdateMsg = 3; - DownlinkMsg downlinkMsg = 4; -} - -message EntityUpdateMsg { - DeviceUpdateMsg deviceUpdateMsg = 1; - DeviceCredentialsUpdateMsg deviceCredentialsUpdateMsg = 2; - RuleChainUpdateMsg ruleChainUpdateMsg = 3; - RuleChainMetadataUpdateMsg ruleChainMetadataUpdateMsg = 4; - DashboardUpdateMsg dashboardUpdateMsg = 5; - AssetUpdateMsg assetUpdateMsg = 6; - EntityViewUpdateMsg entityViewUpdateMsg = 7; - AlarmUpdateMsg alarmUpdateMsg = 8; - UserUpdateMsg userUpdateMsg = 9; - UserCredentialsUpdateMsg userCredentialsUpdateMsg = 10; - CustomerUpdateMsg customerUpdateMsg = 11; - RelationUpdateMsg relationUpdateMsg = 12; + DownlinkMsg downlinkMsg = 3; } enum RequestMsgType { @@ -259,11 +244,41 @@ message UserUpdateMsg { UpdateMsgType msgType = 1; int64 idMSB = 2; int64 idLSB = 3; - string email = 4; - string authority = 5; - string firstName = 6; - string lastName = 7; - string additionalInfo = 8; + int64 customerIdMSB = 4; + int64 customerIdLSB = 5; + string email = 6; + string authority = 7; + string firstName = 8; + string lastName = 9; + string additionalInfo = 10; +} + +message WidgetsBundleUpdateMsg { + UpdateMsgType msgType = 1; + int64 idMSB = 2; + int64 idLSB = 3; + string title = 4; + string alias = 5; + bytes image = 6; + bool isSystem = 7; +} + +message WidgetTypeUpdateMsg { + UpdateMsgType msgType = 1; + int64 idMSB = 2; + int64 idLSB = 3; + string bundleAlias = 4; + string alias = 5; + string name = 6; + string descriptorJson = 7; + bool isSystem = 8; +} + +message AdminSettingsUpdateMsg { + int64 idMSB = 1; + int64 idLSB = 2; + string key = 3; + string jsonValue = 4; } message UserCredentialsUpdateMsg { @@ -315,11 +330,12 @@ message UplinkMsg { repeated DeviceUpdateMsg deviceUpdateMsg = 3; repeated DeviceCredentialsUpdateMsg deviceCredentialsUpdateMsg = 4; repeated AlarmUpdateMsg alarmUpdateMsg = 5; - repeated RuleChainMetadataRequestMsg ruleChainMetadataRequestMsg = 6; - repeated AttributesRequestMsg attributesRequestMsg = 7; - repeated RelationRequestMsg relationRequestMsg = 8; - repeated UserCredentialsRequestMsg userCredentialsRequestMsg = 9; - repeated DeviceCredentialsRequestMsg deviceCredentialsRequestMsg = 10; + repeated RelationUpdateMsg relationUpdateMsg = 6; + repeated RuleChainMetadataRequestMsg ruleChainMetadataRequestMsg = 7; + repeated AttributesRequestMsg attributesRequestMsg = 8; + repeated RelationRequestMsg relationRequestMsg = 9; + repeated UserCredentialsRequestMsg userCredentialsRequestMsg = 10; + repeated DeviceCredentialsRequestMsg deviceCredentialsRequestMsg = 11; } message UplinkResponseMsg { @@ -327,9 +343,30 @@ message UplinkResponseMsg { string errorMsg = 2; } +message DownlinkResponseMsg { + bool success = 1; + string errorMsg = 2; +} + message DownlinkMsg { int32 downlinkMsgId = 1; repeated EntityDataProto entityData = 2; repeated DeviceCredentialsRequestMsg deviceCredentialsRequestMsg = 3; + repeated DeviceUpdateMsg deviceUpdateMsg = 4; + repeated DeviceCredentialsUpdateMsg deviceCredentialsUpdateMsg = 5; + repeated RuleChainUpdateMsg ruleChainUpdateMsg = 6; + repeated RuleChainMetadataUpdateMsg ruleChainMetadataUpdateMsg = 7; + repeated DashboardUpdateMsg dashboardUpdateMsg = 8; + repeated AssetUpdateMsg assetUpdateMsg = 9; + repeated EntityViewUpdateMsg entityViewUpdateMsg = 10; + repeated AlarmUpdateMsg alarmUpdateMsg = 11; + repeated UserUpdateMsg userUpdateMsg = 12; + repeated UserCredentialsUpdateMsg userCredentialsUpdateMsg = 13; + repeated CustomerUpdateMsg customerUpdateMsg = 14; + repeated RelationUpdateMsg relationUpdateMsg = 15; + repeated WidgetsBundleUpdateMsg widgetsBundleUpdateMsg = 16; + repeated WidgetTypeUpdateMsg widgetTypeUpdateMsg = 17; + repeated AdminSettingsUpdateMsg adminSettingsUpdateMsg = 18; + } diff --git a/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/adaptor/JsonConverter.java b/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/adaptor/JsonConverter.java index 8375b84ffa..8ddfe0f580 100644 --- a/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/adaptor/JsonConverter.java +++ b/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/adaptor/JsonConverter.java @@ -68,12 +68,16 @@ public class JsonConverter { private static int maxStringValueLength = 0; - public static PostTelemetryMsg convertToTelemetryProto(JsonElement jsonElement) throws JsonSyntaxException { + public static PostTelemetryMsg convertToTelemetryProto(JsonElement jsonElement, long ts) throws JsonSyntaxException { PostTelemetryMsg.Builder builder = PostTelemetryMsg.newBuilder(); - convertToTelemetry(jsonElement, System.currentTimeMillis(), null, builder); + convertToTelemetry(jsonElement, ts, null, builder); return builder.build(); } + public static PostTelemetryMsg convertToTelemetryProto(JsonElement jsonElement) throws JsonSyntaxException { + return convertToTelemetryProto(jsonElement, System.currentTimeMillis()); + } + private static void convertToTelemetry(JsonElement jsonElement, long systemTs, Map> result, PostTelemetryMsg.Builder builder) { if (jsonElement.isJsonObject()) { parseObject(systemTs, result, builder, jsonElement.getAsJsonObject()); diff --git a/dao/src/main/resources/sql/schema-entities.sql b/dao/src/main/resources/sql/schema-entities.sql index 606dd0c8a7..04346b4ed3 100644 --- a/dao/src/main/resources/sql/schema-entities.sql +++ b/dao/src/main/resources/sql/schema-entities.sql @@ -307,3 +307,20 @@ BEGIN deleted := ttl_deleted_count + debug_ttl_deleted_count; END $$; + +CREATE OR REPLACE PROCEDURE cleanup_edge_events_by_ttl(IN ttl bigint, INOUT deleted bigint) + LANGUAGE plpgsql AS +$$ +DECLARE + ttl_ts bigint; + ttl_deleted_count bigint DEFAULT 0; +BEGIN + IF ttl > 0 THEN + ttl_ts := (EXTRACT(EPOCH FROM current_timestamp) * 1000 - ttl::bigint * 1000)::bigint; + EXECUTE format( + 'WITH deleted AS (DELETE FROM edge_event WHERE ts < %L::bigint RETURNING *) SELECT count(*) FROM deleted', ttl_ts) into ttl_deleted_count; + END IF; + RAISE NOTICE 'Edge events removed by ttl: %', ttl_deleted_count; + deleted := ttl_deleted_count; +END +$$; diff --git a/pom.xml b/pom.xml index b86c3096b0..7b9e8fbf10 100755 --- a/pom.xml +++ b/pom.xml @@ -725,6 +725,7 @@ **/*.proto.js docker/haproxy/** docker/tb-node/** + .run/** JAVADOC_STYLE diff --git a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/edge/TbMsgPushToEdgeNode.java b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/edge/TbMsgPushToEdgeNode.java index f1134078c1..be265ebc19 100644 --- a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/edge/TbMsgPushToEdgeNode.java +++ b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/edge/TbMsgPushToEdgeNode.java @@ -33,11 +33,15 @@ import org.thingsboard.server.common.data.DataConstants; import org.thingsboard.server.common.data.EdgeUtils; import org.thingsboard.server.common.data.EntityType; import org.thingsboard.server.common.data.audit.ActionType; +import org.thingsboard.server.common.data.edge.Edge; import org.thingsboard.server.common.data.edge.EdgeEvent; import org.thingsboard.server.common.data.edge.EdgeEventType; import org.thingsboard.server.common.data.id.EdgeId; import org.thingsboard.server.common.data.id.EntityId; +import org.thingsboard.server.common.data.id.IdBased; import org.thingsboard.server.common.data.id.TenantId; +import org.thingsboard.server.common.data.page.TextPageData; +import org.thingsboard.server.common.data.page.TextPageLink; import org.thingsboard.server.common.data.plugin.ComponentType; import org.thingsboard.server.common.data.relation.EntityRelation; import org.thingsboard.server.common.data.relation.RelationTypeGroup; @@ -46,10 +50,12 @@ import org.thingsboard.server.common.msg.TbMsg; import org.thingsboard.server.common.msg.session.SessionMsgType; import javax.annotation.Nullable; +import java.util.ArrayList; import java.util.HashMap; import java.util.List; import java.util.Map; import java.util.UUID; +import java.util.stream.Collectors; import static org.thingsboard.rule.engine.api.TbRelationTypes.SUCCESS; @@ -84,34 +90,38 @@ public class TbMsgPushToEdgeNode implements TbNode { } if (isSupportedOriginator(msg.getOriginator().getEntityType())) { if (isSupportedMsgType(msg.getType())) { - ListenableFuture getEdgeIdFuture = getEdgeIdByOriginatorId(ctx, ctx.getTenantId(), msg.getOriginator()); - Futures.addCallback(getEdgeIdFuture, new FutureCallback() { + ListenableFuture> getEdgeIdsFuture = getEdgeIdsByOriginatorId(ctx, ctx.getTenantId(), msg.getOriginator()); + Futures.addCallback(getEdgeIdsFuture, new FutureCallback>() { @Override - public void onSuccess(@Nullable EdgeId edgeId) { - try { - EdgeEvent edgeEvent = buildEdgeEvent(msg, ctx); - if (edgeEvent == null) { - log.debug("Edge event type is null. Entity Type {}", msg.getOriginator().getEntityType()); - ctx.tellFailure(msg, new RuntimeException("Edge event type is null. Entity Type '" + msg.getOriginator().getEntityType() + "'")); - } else { - edgeEvent.setEdgeId(edgeId); - ListenableFuture saveFuture = ctx.getEdgeEventService().saveAsync(edgeEvent); - Futures.addCallback(saveFuture, new FutureCallback() { - @Override - public void onSuccess(@Nullable EdgeEvent event) { - ctx.tellNext(msg, SUCCESS); + public void onSuccess(@Nullable List edgeIds) { + if (edgeIds != null && !edgeIds.isEmpty()) { + for (EdgeId edgeId : edgeIds) { + try { + EdgeEvent edgeEvent = buildEdgeEvent(msg, ctx); + if (edgeEvent == null) { + log.debug("Edge event type is null. Entity Type {}", msg.getOriginator().getEntityType()); + ctx.tellFailure(msg, new RuntimeException("Edge event type is null. Entity Type '" + msg.getOriginator().getEntityType() + "'")); + } else { + edgeEvent.setEdgeId(edgeId); + ListenableFuture saveFuture = ctx.getEdgeEventService().saveAsync(edgeEvent); + Futures.addCallback(saveFuture, new FutureCallback() { + @Override + public void onSuccess(@Nullable EdgeEvent event) { + ctx.tellNext(msg, SUCCESS); + } + + @Override + public void onFailure(Throwable th) { + log.error("Could not save edge event", th); + ctx.tellFailure(msg, th); + } + }, ctx.getDbCallbackExecutor()); } - - @Override - public void onFailure(Throwable th) { - log.error("Could not save edge event", th); - ctx.tellFailure(msg, th); - } - }, ctx.getDbCallbackExecutor()); + } catch (JsonProcessingException e) { + log.error("Failed to build edge event", e); + ctx.tellFailure(msg, e); + } } - } catch (JsonProcessingException e) { - log.error("Failed to build edge event", e); - ctx.tellFailure(msg, e); } } @@ -140,13 +150,7 @@ public class TbMsgPushToEdgeNode implements TbNode { return null; } ActionType actionType = getActionTypeByMsgType(msg.getType()); - JsonNode entityBody = null; - JsonNode data = json.readTree(msg.getData()); - if (actionType.equals(ActionType.ATTRIBUTES_UPDATED) || actionType.equals(ActionType.ATTRIBUTES_DELETED)) { - entityBody = getAttributeEntityBody(actionType, data, msg.getMetaData().getData()); - } else { - entityBody = data; - } + JsonNode entityBody = getEntityBody(actionType, msg.getData(), msg.getMetaData().getData()); return buildEdgeEvent(ctx.getTenantId(), actionType, msg.getOriginator().getId(), edgeEventTypeByEntityType, entityBody); } } @@ -161,19 +165,25 @@ public class TbMsgPushToEdgeNode implements TbNode { return edgeEvent; } - private JsonNode getAttributeEntityBody(ActionType actionType, JsonNode data, Map metadata) throws JsonProcessingException { - Map entityData = new HashMap<>(); + private JsonNode getEntityBody(ActionType actionType, String data, Map metadata) throws JsonProcessingException { + Map entityBody = new HashMap<>(); + JsonNode dataJson = json.readTree(data); switch (actionType) { case ATTRIBUTES_UPDATED: - entityData.put("kv", data); + entityBody.put("kv", dataJson); + entityBody.put("scope", metadata.get("scope")); break; case ATTRIBUTES_DELETED: - List keys = json.treeToValue(data.get("attributes"), List.class); - entityData.put("keys", keys); + List keys = json.treeToValue(dataJson.get("attributes"), List.class); + entityBody.put("keys", keys); + entityBody.put("scope", metadata.get("scope")); + break; + case TIMESERIES_UPDATED: + entityBody.put("data", dataJson); + entityBody.put("ts", metadata.get("ts")); break; } - entityData.put("scope", metadata.get("scope")); - return json.valueToTree(entityData); + return json.valueToTree(entityBody); } private UUID getUUIDFromMsgData(TbMsg msg) throws JsonProcessingException { @@ -201,6 +211,8 @@ public class TbMsgPushToEdgeNode implements TbNode { case ASSET: case ENTITY_VIEW: case DASHBOARD: + case TENANT: + case CUSTOMER: return true; default: return false; @@ -215,15 +227,20 @@ public class TbMsgPushToEdgeNode implements TbNode { || DataConstants.ALARM.equals(msgType); } - private ListenableFuture getEdgeIdByOriginatorId(TbContext ctx, TenantId tenantId, EntityId originatorId) { - ListenableFuture> future = ctx.getRelationService().findByToAndTypeAsync(tenantId, originatorId, EntityRelation.CONTAINS_TYPE, RelationTypeGroup.EDGE); - return Futures.transform(future, relations -> { - if (relations != null && relations.size() > 0) { - return new EdgeId(relations.get(0).getFrom().getId()); - } else { - return null; - } - }, ctx.getDbCallbackExecutor()); + private ListenableFuture> getEdgeIdsByOriginatorId(TbContext ctx, TenantId tenantId, EntityId originatorId) { + if (EntityType.TENANT.equals(originatorId.getEntityType())) { + TextPageData edgesByTenantId = ctx.getEdgeService().findEdgesByTenantId(tenantId, new TextPageLink(Integer.MAX_VALUE)); + return Futures.immediateFuture(edgesByTenantId.getData().stream().map(IdBased::getId).collect(Collectors.toList())); + } else { + ListenableFuture> future = ctx.getRelationService().findByToAndTypeAsync(tenantId, originatorId, EntityRelation.CONTAINS_TYPE, RelationTypeGroup.EDGE); + return Futures.transform(future, relations -> { + List result = new ArrayList<>(); + if (relations != null && relations.size() > 0) { + result.add(new EdgeId(relations.get(0).getFrom().getId())); + } + return result; + }, ctx.getDbCallbackExecutor()); + } } @Override