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 29ff49d626..cd2b8fa626 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 @@ -23,6 +23,7 @@ import org.springframework.stereotype.Component; import org.thingsboard.server.cluster.TbClusterService; import org.thingsboard.server.dao.asset.AssetService; import org.thingsboard.server.dao.attributes.AttributesService; +import org.thingsboard.server.dao.customer.CustomerService; import org.thingsboard.server.dao.dashboard.DashboardService; import org.thingsboard.server.dao.device.DeviceProfileService; import org.thingsboard.server.dao.edge.EdgeEventService; @@ -95,6 +96,9 @@ public class EdgeContextComponent { @Autowired private UserService userService; + @Autowired + private CustomerService customerService; + @Autowired private WidgetsBundleService widgetsBundleService; 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 3f8950f281..08770c7b43 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 @@ -184,7 +184,7 @@ public final class EdgeGrpcSession implements Closeable { public void startSyncProcess(TenantId tenantId, EdgeId edgeId) { log.trace("[{}][{}] Staring edge sync process", tenantId, edgeId); syncCompleted = false; - doSync(new EdgeSyncCursor(ctx, edge)); + doSync(new EdgeSyncCursor(ctx)); } private void doSync(EdgeSyncCursor cursor) { @@ -514,11 +514,11 @@ public final class EdgeGrpcSession implements Closeable { case DEVICE_PROFILE: return ctx.getDeviceProfileProcessor().processDeviceProfileToEdge(edgeEvent, msgType, action); case ASSET: - return ctx.getAssetProcessor().processAssetToEdge(edge, edgeEvent, msgType, action); + return ctx.getAssetProcessor().processAssetToEdge(edgeEvent, msgType, action); case ENTITY_VIEW: - return ctx.getEntityViewProcessor().processEntityViewToEdge(edge, edgeEvent, msgType, action); + return ctx.getEntityViewProcessor().processEntityViewToEdge(edgeEvent, msgType, action); case DASHBOARD: - return ctx.getDashboardProcessor().processDashboardToEdge(edge, edgeEvent, msgType, action); + return ctx.getDashboardProcessor().processDashboardToEdge(edgeEvent, msgType, action); case CUSTOMER: return ctx.getCustomerProcessor().processCustomerToEdge(edgeEvent, msgType, action); case RULE_CHAIN: diff --git a/application/src/main/java/org/thingsboard/server/service/edge/rpc/EdgeSyncCursor.java b/application/src/main/java/org/thingsboard/server/service/edge/rpc/EdgeSyncCursor.java index 1df44dd589..f679a88a77 100644 --- a/application/src/main/java/org/thingsboard/server/service/edge/rpc/EdgeSyncCursor.java +++ b/application/src/main/java/org/thingsboard/server/service/edge/rpc/EdgeSyncCursor.java @@ -15,13 +15,10 @@ */ package org.thingsboard.server.service.edge.rpc; -import org.thingsboard.server.common.data.edge.Edge; -import org.thingsboard.server.common.data.id.EntityId; import org.thingsboard.server.service.edge.EdgeContextComponent; import org.thingsboard.server.service.edge.rpc.fetch.AdminSettingsEdgeEventFetcher; import org.thingsboard.server.service.edge.rpc.fetch.AssetsEdgeEventFetcher; import org.thingsboard.server.service.edge.rpc.fetch.CustomerEdgeEventFetcher; -import org.thingsboard.server.service.edge.rpc.fetch.CustomerUsersEdgeEventFetcher; import org.thingsboard.server.service.edge.rpc.fetch.DashboardsEdgeEventFetcher; import org.thingsboard.server.service.edge.rpc.fetch.DeviceProfilesEdgeEventFetcher; import org.thingsboard.server.service.edge.rpc.fetch.EdgeEventFetcher; @@ -29,8 +26,8 @@ import org.thingsboard.server.service.edge.rpc.fetch.OtaPackagesEdgeEventFetcher import org.thingsboard.server.service.edge.rpc.fetch.QueuesEdgeEventFetcher; import org.thingsboard.server.service.edge.rpc.fetch.RuleChainsEdgeEventFetcher; import org.thingsboard.server.service.edge.rpc.fetch.SystemWidgetsBundlesEdgeEventFetcher; -import org.thingsboard.server.service.edge.rpc.fetch.TenantAdminUsersEdgeEventFetcher; import org.thingsboard.server.service.edge.rpc.fetch.TenantWidgetsBundlesEdgeEventFetcher; +import org.thingsboard.server.service.edge.rpc.fetch.UsersEdgeEventFetcher; import java.util.LinkedList; import java.util.List; @@ -42,16 +39,13 @@ public class EdgeSyncCursor { int currentIdx = 0; - public EdgeSyncCursor(EdgeContextComponent ctx, Edge edge) { + public EdgeSyncCursor(EdgeContextComponent ctx) { fetchers.add(new QueuesEdgeEventFetcher(ctx.getQueueService())); fetchers.add(new RuleChainsEdgeEventFetcher(ctx.getRuleChainService())); fetchers.add(new AdminSettingsEdgeEventFetcher(ctx.getAdminSettingsService(), ctx.getFreemarkerConfig())); fetchers.add(new DeviceProfilesEdgeEventFetcher(ctx.getDeviceProfileService())); - fetchers.add(new TenantAdminUsersEdgeEventFetcher(ctx.getUserService())); - if (edge.getCustomerId() != null && !EntityId.NULL_UUID.equals(edge.getCustomerId().getId())) { - fetchers.add(new CustomerEdgeEventFetcher()); - fetchers.add(new CustomerUsersEdgeEventFetcher(ctx.getUserService(), edge.getCustomerId())); - } + fetchers.add(new CustomerEdgeEventFetcher(ctx.getCustomerService())); + fetchers.add(new UsersEdgeEventFetcher(ctx.getUserService())); fetchers.add(new AssetsEdgeEventFetcher(ctx.getAssetService())); fetchers.add(new SystemWidgetsBundlesEdgeEventFetcher(ctx.getWidgetsBundleService())); fetchers.add(new TenantWidgetsBundlesEdgeEventFetcher(ctx.getWidgetsBundleService())); diff --git a/application/src/main/java/org/thingsboard/server/service/edge/rpc/constructor/AssetMsgConstructor.java b/application/src/main/java/org/thingsboard/server/service/edge/rpc/constructor/AssetMsgConstructor.java index 0a78afa6b7..47e5bbf1b3 100644 --- a/application/src/main/java/org/thingsboard/server/service/edge/rpc/constructor/AssetMsgConstructor.java +++ b/application/src/main/java/org/thingsboard/server/service/edge/rpc/constructor/AssetMsgConstructor.java @@ -19,7 +19,6 @@ import org.springframework.stereotype.Component; import org.thingsboard.common.util.JacksonUtil; import org.thingsboard.server.common.data.asset.Asset; import org.thingsboard.server.common.data.id.AssetId; -import org.thingsboard.server.common.data.id.CustomerId; import org.thingsboard.server.gen.edge.v1.AssetUpdateMsg; import org.thingsboard.server.gen.edge.v1.UpdateMsgType; import org.thingsboard.server.queue.util.TbCoreComponent; @@ -28,7 +27,7 @@ import org.thingsboard.server.queue.util.TbCoreComponent; @TbCoreComponent public class AssetMsgConstructor { - public AssetUpdateMsg constructAssetUpdatedMsg(UpdateMsgType msgType, Asset asset, CustomerId customerId) { + public AssetUpdateMsg constructAssetUpdatedMsg(UpdateMsgType msgType, Asset asset) { AssetUpdateMsg.Builder builder = AssetUpdateMsg.newBuilder() .setMsgType(msgType) .setIdMSB(asset.getId().getId().getMostSignificantBits()) @@ -38,9 +37,9 @@ public class AssetMsgConstructor { if (asset.getLabel() != null) { builder.setLabel(asset.getLabel()); } - if (customerId != null) { - builder.setCustomerIdMSB(customerId.getId().getMostSignificantBits()); - builder.setCustomerIdLSB(customerId.getId().getLeastSignificantBits()); + if (asset.getCustomerId() != null) { + builder.setCustomerIdMSB(asset.getCustomerId().getId().getMostSignificantBits()); + builder.setCustomerIdLSB(asset.getCustomerId().getId().getLeastSignificantBits()); } if (asset.getAdditionalInfo() != null) { builder.setAdditionalInfo(JacksonUtil.toString(asset.getAdditionalInfo())); diff --git a/application/src/main/java/org/thingsboard/server/service/edge/rpc/constructor/DashboardMsgConstructor.java b/application/src/main/java/org/thingsboard/server/service/edge/rpc/constructor/DashboardMsgConstructor.java index e240fa142d..c61c5eb260 100644 --- a/application/src/main/java/org/thingsboard/server/service/edge/rpc/constructor/DashboardMsgConstructor.java +++ b/application/src/main/java/org/thingsboard/server/service/edge/rpc/constructor/DashboardMsgConstructor.java @@ -18,7 +18,6 @@ package org.thingsboard.server.service.edge.rpc.constructor; import org.springframework.stereotype.Component; import org.thingsboard.common.util.JacksonUtil; import org.thingsboard.server.common.data.Dashboard; -import org.thingsboard.server.common.data.id.CustomerId; import org.thingsboard.server.common.data.id.DashboardId; import org.thingsboard.server.gen.edge.v1.DashboardUpdateMsg; import org.thingsboard.server.gen.edge.v1.UpdateMsgType; @@ -28,16 +27,15 @@ import org.thingsboard.server.queue.util.TbCoreComponent; @TbCoreComponent public class DashboardMsgConstructor { - public DashboardUpdateMsg constructDashboardUpdatedMsg(UpdateMsgType msgType, Dashboard dashboard, CustomerId customerId) { + public DashboardUpdateMsg constructDashboardUpdatedMsg(UpdateMsgType msgType, Dashboard dashboard) { DashboardUpdateMsg.Builder builder = DashboardUpdateMsg.newBuilder() .setMsgType(msgType) .setIdMSB(dashboard.getId().getId().getMostSignificantBits()) .setIdLSB(dashboard.getId().getId().getLeastSignificantBits()) .setTitle(dashboard.getTitle()) .setConfiguration(JacksonUtil.toString(dashboard.getConfiguration())); - if (customerId != null) { - builder.setCustomerIdMSB(customerId.getId().getMostSignificantBits()); - builder.setCustomerIdLSB(customerId.getId().getLeastSignificantBits()); + if (dashboard.getAssignedCustomers() != null) { + builder.setAssignedCustomers(JacksonUtil.toString(dashboard.getAssignedCustomers())); } return builder.build(); } diff --git a/application/src/main/java/org/thingsboard/server/service/edge/rpc/constructor/DeviceMsgConstructor.java b/application/src/main/java/org/thingsboard/server/service/edge/rpc/constructor/DeviceMsgConstructor.java index 3c2db96b31..511910dd8a 100644 --- a/application/src/main/java/org/thingsboard/server/service/edge/rpc/constructor/DeviceMsgConstructor.java +++ b/application/src/main/java/org/thingsboard/server/service/edge/rpc/constructor/DeviceMsgConstructor.java @@ -41,7 +41,7 @@ public class DeviceMsgConstructor { @Autowired private DataDecodingEncodingService dataDecodingEncodingService; - public DeviceUpdateMsg constructDeviceUpdatedMsg(UpdateMsgType msgType, Device device, CustomerId customerId, String conflictName) { + public DeviceUpdateMsg constructDeviceUpdatedMsg(UpdateMsgType msgType, Device device, String conflictName) { DeviceUpdateMsg.Builder builder = DeviceUpdateMsg.newBuilder() .setMsgType(msgType) .setIdMSB(device.getId().getId().getMostSignificantBits()) @@ -51,9 +51,9 @@ public class DeviceMsgConstructor { if (device.getLabel() != null) { builder.setLabel(device.getLabel()); } - if (customerId != null) { - builder.setCustomerIdMSB(customerId.getId().getMostSignificantBits()); - builder.setCustomerIdLSB(customerId.getId().getLeastSignificantBits()); + if (device.getCustomerId() != null) { + builder.setCustomerIdMSB(device.getCustomerId().getId().getMostSignificantBits()); + builder.setCustomerIdLSB(device.getCustomerId().getId().getLeastSignificantBits()); } if (device.getDeviceProfileId() != null) { builder.setDeviceProfileIdMSB(device.getDeviceProfileId().getId().getMostSignificantBits()); diff --git a/application/src/main/java/org/thingsboard/server/service/edge/rpc/constructor/EntityViewMsgConstructor.java b/application/src/main/java/org/thingsboard/server/service/edge/rpc/constructor/EntityViewMsgConstructor.java index 759ce9e500..d5aa79700d 100644 --- a/application/src/main/java/org/thingsboard/server/service/edge/rpc/constructor/EntityViewMsgConstructor.java +++ b/application/src/main/java/org/thingsboard/server/service/edge/rpc/constructor/EntityViewMsgConstructor.java @@ -18,7 +18,6 @@ package org.thingsboard.server.service.edge.rpc.constructor; import org.springframework.stereotype.Component; import org.thingsboard.common.util.JacksonUtil; import org.thingsboard.server.common.data.EntityView; -import org.thingsboard.server.common.data.id.CustomerId; import org.thingsboard.server.common.data.id.EntityViewId; import org.thingsboard.server.gen.edge.v1.EdgeEntityType; import org.thingsboard.server.gen.edge.v1.EntityViewUpdateMsg; @@ -29,7 +28,7 @@ import org.thingsboard.server.queue.util.TbCoreComponent; @TbCoreComponent public class EntityViewMsgConstructor { - public EntityViewUpdateMsg constructEntityViewUpdatedMsg(UpdateMsgType msgType, EntityView entityView, CustomerId customerId) { + public EntityViewUpdateMsg constructEntityViewUpdatedMsg(UpdateMsgType msgType, EntityView entityView) { EdgeEntityType entityType; switch (entityView.getEntityId().getEntityType()) { case DEVICE: @@ -50,9 +49,9 @@ public class EntityViewMsgConstructor { .setEntityIdMSB(entityView.getEntityId().getId().getMostSignificantBits()) .setEntityIdLSB(entityView.getEntityId().getId().getLeastSignificantBits()) .setEntityType(entityType); - if (customerId != null) { - builder.setCustomerIdMSB(customerId.getId().getMostSignificantBits()); - builder.setCustomerIdLSB(customerId.getId().getLeastSignificantBits()); + if (entityView.getCustomerId() != null) { + builder.setCustomerIdMSB(entityView.getCustomerId().getId().getMostSignificantBits()); + builder.setCustomerIdLSB(entityView.getCustomerId().getId().getLeastSignificantBits()); } if (entityView.getAdditionalInfo() != null) { builder.setAdditionalInfo(JacksonUtil.toString(entityView.getAdditionalInfo())); diff --git a/application/src/main/java/org/thingsboard/server/service/edge/rpc/constructor/UserMsgConstructor.java b/application/src/main/java/org/thingsboard/server/service/edge/rpc/constructor/UserMsgConstructor.java index ac6b822bef..625b555e24 100644 --- a/application/src/main/java/org/thingsboard/server/service/edge/rpc/constructor/UserMsgConstructor.java +++ b/application/src/main/java/org/thingsboard/server/service/edge/rpc/constructor/UserMsgConstructor.java @@ -18,7 +18,6 @@ package org.thingsboard.server.service.edge.rpc.constructor; import org.springframework.stereotype.Component; import org.thingsboard.common.util.JacksonUtil; import org.thingsboard.server.common.data.User; -import org.thingsboard.server.common.data.id.CustomerId; import org.thingsboard.server.common.data.id.UserId; import org.thingsboard.server.common.data.security.UserCredentials; import org.thingsboard.server.gen.edge.v1.UpdateMsgType; @@ -30,16 +29,16 @@ import org.thingsboard.server.queue.util.TbCoreComponent; @TbCoreComponent public class UserMsgConstructor { - public UserUpdateMsg constructUserUpdatedMsg(UpdateMsgType msgType, User user, CustomerId customerId) { + public UserUpdateMsg constructUserUpdatedMsg(UpdateMsgType msgType, User user) { UserUpdateMsg.Builder builder = UserUpdateMsg.newBuilder() .setMsgType(msgType) .setIdMSB(user.getId().getId().getMostSignificantBits()) .setIdLSB(user.getId().getId().getLeastSignificantBits()) .setEmail(user.getEmail()) .setAuthority(user.getAuthority().name()); - if (customerId != null) { - builder.setCustomerIdMSB(customerId.getId().getMostSignificantBits()); - builder.setCustomerIdLSB(customerId.getId().getLeastSignificantBits()); + 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/fetch/CustomerEdgeEventFetcher.java b/application/src/main/java/org/thingsboard/server/service/edge/rpc/fetch/CustomerEdgeEventFetcher.java index 1d5c618a0a..8674a4e23b 100644 --- a/application/src/main/java/org/thingsboard/server/service/edge/rpc/fetch/CustomerEdgeEventFetcher.java +++ b/application/src/main/java/org/thingsboard/server/service/edge/rpc/fetch/CustomerEdgeEventFetcher.java @@ -17,6 +17,7 @@ package org.thingsboard.server.service.edge.rpc.fetch; import lombok.AllArgsConstructor; import lombok.extern.slf4j.Slf4j; +import org.thingsboard.server.common.data.Customer; import org.thingsboard.server.common.data.EdgeUtils; import org.thingsboard.server.common.data.edge.Edge; import org.thingsboard.server.common.data.edge.EdgeEvent; @@ -25,25 +26,22 @@ import org.thingsboard.server.common.data.edge.EdgeEventType; import org.thingsboard.server.common.data.id.TenantId; import org.thingsboard.server.common.data.page.PageData; import org.thingsboard.server.common.data.page.PageLink; +import org.thingsboard.server.dao.customer.CustomerService; -import java.util.ArrayList; -import java.util.List; - -@AllArgsConstructor @Slf4j -public class CustomerEdgeEventFetcher implements EdgeEventFetcher { +@AllArgsConstructor +public class CustomerEdgeEventFetcher extends BasePageableEdgeEventFetcher { + + private final CustomerService customerService; @Override - public PageLink getPageLink(int pageSize) { - return null; + PageData fetchPageData(TenantId tenantId, Edge edge, PageLink pageLink) { + return customerService.findCustomersByTenantId(tenantId, pageLink); } @Override - public PageData fetchEdgeEvents(TenantId tenantId, Edge edge, PageLink pageLink) { - List result = new ArrayList<>(); - result.add(EdgeUtils.constructEdgeEvent(edge.getTenantId(), edge.getId(), - EdgeEventType.CUSTOMER, EdgeEventActionType.ADDED, edge.getCustomerId(), null)); - // @voba - returns PageData object to be in sync with other fetchers - return new PageData<>(result, 1, result.size(), false); + EdgeEvent constructEdgeEvent(TenantId tenantId, Edge edge, Customer customer) { + return EdgeUtils.constructEdgeEvent(edge.getTenantId(), edge.getId(), EdgeEventType.CUSTOMER, + EdgeEventActionType.ADDED, customer.getId(), null); } } diff --git a/application/src/main/java/org/thingsboard/server/service/edge/rpc/fetch/CustomerUsersEdgeEventFetcher.java b/application/src/main/java/org/thingsboard/server/service/edge/rpc/fetch/CustomerUsersEdgeEventFetcher.java deleted file mode 100644 index cb72ca8bcb..0000000000 --- a/application/src/main/java/org/thingsboard/server/service/edge/rpc/fetch/CustomerUsersEdgeEventFetcher.java +++ /dev/null @@ -1,41 +0,0 @@ -/** - * Copyright © 2016-2022 The Thingsboard Authors - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ -package org.thingsboard.server.service.edge.rpc.fetch; - -import org.thingsboard.server.common.data.User; -import org.thingsboard.server.common.data.edge.Edge; -import org.thingsboard.server.common.data.edge.EdgeEvent; -import org.thingsboard.server.common.data.id.CustomerId; -import org.thingsboard.server.common.data.id.TenantId; -import org.thingsboard.server.common.data.page.PageData; -import org.thingsboard.server.common.data.page.PageLink; -import org.thingsboard.server.dao.user.UserService; - -public class CustomerUsersEdgeEventFetcher extends BaseUsersEdgeEventFetcher { - - private final CustomerId customerId; - - public CustomerUsersEdgeEventFetcher(UserService userService, CustomerId customerId) { - super(userService); - this.customerId = customerId; - } - - @Override - protected PageData findUsers(TenantId tenantId, PageLink pageLink) { - return userService.findCustomerUsers(tenantId, customerId, pageLink); - } - -} diff --git a/application/src/main/java/org/thingsboard/server/service/edge/rpc/fetch/TenantAdminUsersEdgeEventFetcher.java b/application/src/main/java/org/thingsboard/server/service/edge/rpc/fetch/TenantAdminUsersEdgeEventFetcher.java deleted file mode 100644 index 2186bdc608..0000000000 --- a/application/src/main/java/org/thingsboard/server/service/edge/rpc/fetch/TenantAdminUsersEdgeEventFetcher.java +++ /dev/null @@ -1,34 +0,0 @@ -/** - * Copyright © 2016-2022 The Thingsboard Authors - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ -package org.thingsboard.server.service.edge.rpc.fetch; - -import org.thingsboard.server.common.data.User; -import org.thingsboard.server.common.data.id.TenantId; -import org.thingsboard.server.common.data.page.PageData; -import org.thingsboard.server.common.data.page.PageLink; -import org.thingsboard.server.dao.user.UserService; - -public class TenantAdminUsersEdgeEventFetcher extends BaseUsersEdgeEventFetcher { - - public TenantAdminUsersEdgeEventFetcher(UserService userService) { - super(userService); - } - - @Override - protected PageData findUsers(TenantId tenantId, PageLink pageLink) { - return userService.findTenantAdmins(tenantId, pageLink); - } -} diff --git a/application/src/main/java/org/thingsboard/server/service/edge/rpc/fetch/BaseUsersEdgeEventFetcher.java b/application/src/main/java/org/thingsboard/server/service/edge/rpc/fetch/UsersEdgeEventFetcher.java similarity index 88% rename from application/src/main/java/org/thingsboard/server/service/edge/rpc/fetch/BaseUsersEdgeEventFetcher.java rename to application/src/main/java/org/thingsboard/server/service/edge/rpc/fetch/UsersEdgeEventFetcher.java index 6791ba69a6..e2883bd951 100644 --- a/application/src/main/java/org/thingsboard/server/service/edge/rpc/fetch/BaseUsersEdgeEventFetcher.java +++ b/application/src/main/java/org/thingsboard/server/service/edge/rpc/fetch/UsersEdgeEventFetcher.java @@ -30,13 +30,13 @@ import org.thingsboard.server.dao.user.UserService; @Slf4j @AllArgsConstructor -public abstract class BaseUsersEdgeEventFetcher extends BasePageableEdgeEventFetcher { +public class UsersEdgeEventFetcher extends BasePageableEdgeEventFetcher { protected final UserService userService; @Override PageData fetchPageData(TenantId tenantId, Edge edge, PageLink pageLink) { - return findUsers(tenantId, pageLink); + return userService.findUsersByTenantId(tenantId, pageLink); } @Override @@ -44,6 +44,4 @@ public abstract class BaseUsersEdgeEventFetcher extends BasePageableEdgeEventFet return EdgeUtils.constructEdgeEvent(tenantId, edge.getId(), EdgeEventType.USER, EdgeEventActionType.ADDED, user.getId(), null); } - - protected abstract PageData findUsers(TenantId tenantId, PageLink pageLink); } diff --git a/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/AdminSettingsEdgeProcessor.java b/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/AdminSettingsEdgeProcessor.java index 21063c1ed7..db30ee2036 100644 --- a/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/AdminSettingsEdgeProcessor.java +++ b/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/AdminSettingsEdgeProcessor.java @@ -17,6 +17,7 @@ package org.thingsboard.server.service.edge.rpc.processor; import lombok.extern.slf4j.Slf4j; import org.springframework.stereotype.Component; +import org.thingsboard.common.util.JacksonUtil; import org.thingsboard.server.common.data.AdminSettings; import org.thingsboard.server.common.data.EdgeUtils; import org.thingsboard.server.common.data.edge.EdgeEvent; @@ -30,7 +31,7 @@ import org.thingsboard.server.queue.util.TbCoreComponent; public class AdminSettingsEdgeProcessor extends BaseEdgeProcessor { public DownlinkMsg processAdminSettingsToEdge(EdgeEvent edgeEvent) { - AdminSettings adminSettings = mapper.convertValue(edgeEvent.getBody(), AdminSettings.class); + AdminSettings adminSettings = JacksonUtil.OBJECT_MAPPER.convertValue(edgeEvent.getBody(), AdminSettings.class); AdminSettingsUpdateMsg adminSettingsUpdateMsg = adminSettingsMsgConstructor.constructAdminSettingsUpdateMsg(adminSettings); return DownlinkMsg.newBuilder() .setDownlinkMsgId(EdgeUtils.nextPositiveInt()) diff --git a/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/AlarmEdgeProcessor.java b/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/AlarmEdgeProcessor.java index a63e377a5b..5fb327df05 100644 --- a/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/AlarmEdgeProcessor.java +++ b/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/AlarmEdgeProcessor.java @@ -20,6 +20,7 @@ import com.google.common.util.concurrent.Futures; import com.google.common.util.concurrent.ListenableFuture; import lombok.extern.slf4j.Slf4j; import org.springframework.stereotype.Component; +import org.thingsboard.common.util.JacksonUtil; import org.thingsboard.server.common.data.EdgeUtils; import org.thingsboard.server.common.data.EntityType; import org.thingsboard.server.common.data.alarm.Alarm; @@ -76,7 +77,7 @@ public class AlarmEdgeProcessor extends BaseEdgeProcessor { existentAlarm.setStatus(AlarmStatus.valueOf(alarmUpdateMsg.getStatus())); existentAlarm.setAckTs(alarmUpdateMsg.getAckTs()); existentAlarm.setEndTs(alarmUpdateMsg.getEndTs()); - existentAlarm.setDetails(mapper.readTree(alarmUpdateMsg.getDetails())); + existentAlarm.setDetails(JacksonUtil.OBJECT_MAPPER.readTree(alarmUpdateMsg.getDetails())); alarmService.createOrUpdateAlarm(existentAlarm); break; case ALARM_ACK_RPC_MESSAGE: @@ -86,7 +87,8 @@ public class AlarmEdgeProcessor extends BaseEdgeProcessor { break; case ALARM_CLEAR_RPC_MESSAGE: if (existentAlarm != null) { - alarmService.clearAlarm(tenantId, existentAlarm.getId(), mapper.readTree(alarmUpdateMsg.getDetails()), alarmUpdateMsg.getAckTs()); + alarmService.clearAlarm(tenantId, existentAlarm.getId(), + JacksonUtil.OBJECT_MAPPER.readTree(alarmUpdateMsg.getDetails()), alarmUpdateMsg.getAckTs()); } break; case ENTITY_DELETED_RPC_MESSAGE: @@ -136,7 +138,7 @@ public class AlarmEdgeProcessor extends BaseEdgeProcessor { } break; case DELETED: - Alarm alarm = mapper.convertValue(edgeEvent.getBody(), Alarm.class); + Alarm alarm = JacksonUtil.OBJECT_MAPPER.convertValue(edgeEvent.getBody(), Alarm.class); AlarmUpdateMsg alarmUpdateMsg = alarmMsgConstructor.constructAlarmUpdatedMsg(edge.getTenantId(), msgType, alarm); downlinkMsg = DownlinkMsg.newBuilder() @@ -154,8 +156,8 @@ public class AlarmEdgeProcessor extends BaseEdgeProcessor { switch (actionType) { case DELETED: EdgeId edgeId = new EdgeId(new UUID(edgeNotificationMsg.getEdgeIdMSB(), edgeNotificationMsg.getEdgeIdLSB())); - Alarm deletedAlarm = mapper.readValue(edgeNotificationMsg.getBody(), Alarm.class); - return saveEdgeEvent(tenantId, edgeId, EdgeEventType.ALARM, actionType, alarmId, mapper.valueToTree(deletedAlarm)); + Alarm deletedAlarm = JacksonUtil.OBJECT_MAPPER.readValue(edgeNotificationMsg.getBody(), Alarm.class); + return saveEdgeEvent(tenantId, edgeId, EdgeEventType.ALARM, actionType, alarmId, JacksonUtil.OBJECT_MAPPER.valueToTree(deletedAlarm)); default: ListenableFuture alarmFuture = alarmService.findAlarmByIdAsync(tenantId, alarmId); return Futures.transformAsync(alarmFuture, alarm -> { diff --git a/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/AssetEdgeProcessor.java b/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/AssetEdgeProcessor.java index 328f67dcaf..8a8fec7539 100644 --- a/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/AssetEdgeProcessor.java +++ b/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/AssetEdgeProcessor.java @@ -19,11 +19,9 @@ import lombok.extern.slf4j.Slf4j; import org.springframework.stereotype.Component; import org.thingsboard.server.common.data.EdgeUtils; import org.thingsboard.server.common.data.asset.Asset; -import org.thingsboard.server.common.data.edge.Edge; import org.thingsboard.server.common.data.edge.EdgeEvent; import org.thingsboard.server.common.data.edge.EdgeEventActionType; import org.thingsboard.server.common.data.id.AssetId; -import org.thingsboard.server.common.data.id.CustomerId; import org.thingsboard.server.gen.edge.v1.AssetUpdateMsg; import org.thingsboard.server.gen.edge.v1.DownlinkMsg; import org.thingsboard.server.gen.edge.v1.UpdateMsgType; @@ -34,7 +32,7 @@ import org.thingsboard.server.queue.util.TbCoreComponent; @TbCoreComponent public class AssetEdgeProcessor extends BaseEdgeProcessor { - public DownlinkMsg processAssetToEdge(Edge edge, EdgeEvent edgeEvent, UpdateMsgType msgType, EdgeEventActionType action) { + public DownlinkMsg processAssetToEdge(EdgeEvent edgeEvent, UpdateMsgType msgType, EdgeEventActionType action) { AssetId assetId = new AssetId(edgeEvent.getEntityId()); DownlinkMsg downlinkMsg = null; switch (action) { @@ -45,9 +43,8 @@ public class AssetEdgeProcessor extends BaseEdgeProcessor { case UNASSIGNED_FROM_CUSTOMER: Asset asset = assetService.findAssetById(edgeEvent.getTenantId(), assetId); if (asset != null) { - CustomerId customerId = getCustomerIdIfEdgeAssignedToCustomer(asset, edge); AssetUpdateMsg assetUpdateMsg = - assetMsgConstructor.constructAssetUpdatedMsg(msgType, asset, customerId); + assetMsgConstructor.constructAssetUpdatedMsg(msgType, asset); downlinkMsg = DownlinkMsg.newBuilder() .setDownlinkMsgId(EdgeUtils.nextPositiveInt()) .addAssetUpdateMsg(assetUpdateMsg) diff --git a/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/BaseEdgeProcessor.java b/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/BaseEdgeProcessor.java index 9d6199bdcd..1867b11bc4 100644 --- a/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/BaseEdgeProcessor.java +++ b/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/BaseEdgeProcessor.java @@ -16,7 +16,6 @@ package org.thingsboard.server.service.edge.rpc.processor; import com.fasterxml.jackson.databind.JsonNode; -import com.fasterxml.jackson.databind.ObjectMapper; import com.google.common.util.concurrent.Futures; import com.google.common.util.concurrent.ListenableFuture; import lombok.extern.slf4j.Slf4j; @@ -25,12 +24,10 @@ import org.springframework.context.annotation.Lazy; import org.thingsboard.server.cluster.TbClusterService; import org.thingsboard.server.common.data.Device; import org.thingsboard.server.common.data.EdgeUtils; -import org.thingsboard.server.common.data.HasCustomerId; import org.thingsboard.server.common.data.edge.Edge; import org.thingsboard.server.common.data.edge.EdgeEvent; import org.thingsboard.server.common.data.edge.EdgeEventActionType; import org.thingsboard.server.common.data.edge.EdgeEventType; -import org.thingsboard.server.common.data.id.CustomerId; import org.thingsboard.server.common.data.id.EdgeId; import org.thingsboard.server.common.data.id.EntityId; import org.thingsboard.server.common.data.id.TenantId; @@ -84,9 +81,7 @@ import java.util.List; @Slf4j public abstract class BaseEdgeProcessor { - protected static final ObjectMapper mapper = new ObjectMapper(); - - protected static final int DEFAULT_PAGE_SIZE = 1000; + protected static final int DEFAULT_PAGE_SIZE = 100; @Autowired protected RuleChainService ruleChainService; @@ -233,14 +228,6 @@ public abstract class BaseEdgeProcessor { }, dbCallbackExecutorService); } - protected CustomerId getCustomerIdIfEdgeAssignedToCustomer(HasCustomerId hasCustomerIdEntity, Edge edge) { - if (!edge.getCustomerId().isNullUid() && edge.getCustomerId().equals(hasCustomerIdEntity.getCustomerId())) { - return edge.getCustomerId(); - } else { - return null; - } - } - protected ListenableFuture processActionForAllEdges(TenantId tenantId, EdgeEventType type, EdgeEventActionType actionType, EntityId entityId) { List> futures = new ArrayList<>(); if (TenantId.SYS_TENANT_ID.equals(tenantId)) { diff --git a/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/DashboardEdgeProcessor.java b/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/DashboardEdgeProcessor.java index f07ed80817..0770d9352b 100644 --- a/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/DashboardEdgeProcessor.java +++ b/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/DashboardEdgeProcessor.java @@ -36,7 +36,7 @@ import java.util.Collections; @TbCoreComponent public class DashboardEdgeProcessor extends BaseEdgeProcessor { - public DownlinkMsg processDashboardToEdge(Edge edge, EdgeEvent edgeEvent, UpdateMsgType msgType, EdgeEventActionType action) { + public DownlinkMsg processDashboardToEdge(EdgeEvent edgeEvent, UpdateMsgType msgType, EdgeEventActionType action) { DashboardId dashboardId = new DashboardId(edgeEvent.getEntityId()); DownlinkMsg downlinkMsg = null; switch (action) { @@ -47,12 +47,8 @@ public class DashboardEdgeProcessor extends BaseEdgeProcessor { case UNASSIGNED_FROM_CUSTOMER: Dashboard dashboard = dashboardService.findDashboardById(edgeEvent.getTenantId(), dashboardId); if (dashboard != null) { - CustomerId customerId = null; - if (!edge.getCustomerId().isNullUid() && dashboard.isAssignedToCustomer(edge.getCustomerId())) { - customerId = edge.getCustomerId(); - } DashboardUpdateMsg dashboardUpdateMsg = - dashboardMsgConstructor.constructDashboardUpdatedMsg(msgType, dashboard, customerId); + dashboardMsgConstructor.constructDashboardUpdatedMsg(msgType, dashboard); downlinkMsg = DownlinkMsg.newBuilder() .setDownlinkMsgId(EdgeUtils.nextPositiveInt()) .addDashboardUpdateMsg(dashboardUpdateMsg) diff --git a/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/DeviceEdgeProcessor.java b/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/DeviceEdgeProcessor.java index 6d0f66a884..50e25166a6 100644 --- a/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/DeviceEdgeProcessor.java +++ b/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/DeviceEdgeProcessor.java @@ -103,7 +103,7 @@ public class DeviceEdgeProcessor extends BaseEdgeProcessor { log.error(errMsg, e); return Futures.immediateFuture(null); } - ObjectNode body = mapper.createObjectNode(); + ObjectNode body = JacksonUtil.OBJECT_MAPPER.createObjectNode(); body.put("conflictName", deviceName); ListenableFuture input = saveEdgeEvent(tenantId, edge.getId(), EdgeEventType.DEVICE, EdgeEventActionType.ENTITY_MERGE_REQUEST, newDevice.getId(), body); return Futures.transformAsync(input, unused -> @@ -296,9 +296,9 @@ public class DeviceEdgeProcessor extends BaseEdgeProcessor { private void pushDeviceCreatedEventToRuleEngine(TenantId tenantId, Edge edge, Device device) { try { DeviceId deviceId = device.getId(); - ObjectNode entityNode = mapper.valueToTree(device); + ObjectNode entityNode = JacksonUtil.OBJECT_MAPPER.valueToTree(device); TbMsg tbMsg = TbMsg.newMsg(DataConstants.ENTITY_CREATED, deviceId, device.getCustomerId(), - getActionTbMsgMetaData(edge, device.getCustomerId()), TbMsgDataType.JSON, mapper.writeValueAsString(entityNode)); + getActionTbMsgMetaData(edge, device.getCustomerId()), TbMsgDataType.JSON, JacksonUtil.OBJECT_MAPPER.writeValueAsString(entityNode)); tbClusterService.pushMsgToRuleEngine(tenantId, deviceId, tbMsg, new TbQueueCallback() { @Override public void onSuccess(TbQueueMsgMetadata metadata) { @@ -374,9 +374,8 @@ public class DeviceEdgeProcessor extends BaseEdgeProcessor { case UNASSIGNED_FROM_CUSTOMER: Device device = deviceService.findDeviceById(edgeEvent.getTenantId(), deviceId); if (device != null) { - CustomerId customerId = getCustomerIdIfEdgeAssignedToCustomer(device, edge); DeviceUpdateMsg deviceUpdateMsg = - deviceMsgConstructor.constructDeviceUpdatedMsg(msgType, device, customerId, null); + deviceMsgConstructor.constructDeviceUpdatedMsg(msgType, device, null); downlinkMsg = DownlinkMsg.newBuilder() .setDownlinkMsgId(EdgeUtils.nextPositiveInt()) .addDeviceUpdateMsg(deviceUpdateMsg) diff --git a/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/EdgeProcessor.java b/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/EdgeProcessor.java index fdbebafc98..73fa828e43 100644 --- a/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/EdgeProcessor.java +++ b/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/EdgeProcessor.java @@ -19,6 +19,7 @@ import com.google.common.util.concurrent.Futures; import com.google.common.util.concurrent.ListenableFuture; import lombok.extern.slf4j.Slf4j; import org.springframework.stereotype.Component; +import org.thingsboard.common.util.JacksonUtil; import org.thingsboard.server.common.data.User; import org.thingsboard.server.common.data.edge.Edge; import org.thingsboard.server.common.data.edge.EdgeEventActionType; @@ -43,43 +44,8 @@ public class EdgeProcessor extends BaseEdgeProcessor { public ListenableFuture processEdgeNotification(TenantId tenantId, TransportProtos.EdgeNotificationMsgProto edgeNotificationMsg) { try { EdgeEventActionType actionType = EdgeEventActionType.valueOf(edgeNotificationMsg.getAction()); - EdgeId edgeId = new EdgeId(new UUID(edgeNotificationMsg.getEntityIdMSB(), edgeNotificationMsg.getEntityIdLSB())); - ListenableFuture edgeFuture; switch (actionType) { - case ASSIGNED_TO_CUSTOMER: - CustomerId customerId = mapper.readValue(edgeNotificationMsg.getBody(), CustomerId.class); - edgeFuture = edgeService.findEdgeByIdAsync(tenantId, edgeId); - return Futures.transformAsync(edgeFuture, edge -> { - if (edge == null || customerId.isNullUid()) { - return Futures.immediateFuture(null); - } - List> futures = new ArrayList<>(); - futures.add(saveEdgeEvent(edge.getTenantId(), edge.getId(), EdgeEventType.CUSTOMER, EdgeEventActionType.ADDED, customerId, null)); - PageLink pageLink = new PageLink(DEFAULT_PAGE_SIZE); - PageData pageData; - do { - pageData = userService.findCustomerUsers(tenantId, customerId, pageLink); - if (pageData != null && pageData.getData() != null && !pageData.getData().isEmpty()) { - log.trace("[{}] [{}] user(s) are going to be added to edge.", edge.getId(), pageData.getData().size()); - for (User user : pageData.getData()) { - futures.add(saveEdgeEvent(edge.getTenantId(), edge.getId(), EdgeEventType.USER, EdgeEventActionType.ADDED, user.getId(), null)); - } - if (pageData.hasNext()) { - pageLink = pageLink.nextPageLink(); - } - } - } while (pageData != null && pageData.hasNext()); - return Futures.transform(Futures.allAsList(futures), voids -> null, dbCallbackExecutorService); - }, dbCallbackExecutorService); - case UNASSIGNED_FROM_CUSTOMER: - CustomerId customerIdToDelete = mapper.readValue(edgeNotificationMsg.getBody(), CustomerId.class); - edgeFuture = edgeService.findEdgeByIdAsync(tenantId, edgeId); - return Futures.transformAsync(edgeFuture, edge -> { - if (edge == null || customerIdToDelete.isNullUid()) { - return Futures.immediateFuture(null); - } - return saveEdgeEvent(edge.getTenantId(), edge.getId(), EdgeEventType.CUSTOMER, EdgeEventActionType.DELETED, customerIdToDelete, null); - }, dbCallbackExecutorService); + // TODO: add support for edge update default: return Futures.immediateFuture(null); } diff --git a/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/EntityEdgeProcessor.java b/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/EntityEdgeProcessor.java index 195f932b93..c2b7a5cab5 100644 --- a/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/EntityEdgeProcessor.java +++ b/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/EntityEdgeProcessor.java @@ -19,6 +19,7 @@ import com.google.common.util.concurrent.Futures; import com.google.common.util.concurrent.ListenableFuture; import lombok.extern.slf4j.Slf4j; import org.springframework.stereotype.Component; +import org.thingsboard.common.util.JacksonUtil; import org.thingsboard.server.common.data.Device; import org.thingsboard.server.common.data.EdgeUtils; import org.thingsboard.server.common.data.edge.Edge; @@ -57,13 +58,12 @@ public class EntityEdgeProcessor extends BaseEdgeProcessor { if (EdgeEventType.DEVICE.equals(edgeEvent.getType())) { DeviceId deviceId = new DeviceId(edgeEvent.getEntityId()); Device device = deviceService.findDeviceById(edge.getTenantId(), deviceId); - CustomerId customerId = getCustomerIdIfEdgeAssignedToCustomer(device, edge); String conflictName = null; if(edgeEvent.getBody() != null) { conflictName = edgeEvent.getBody().get("conflictName").asText(); } DeviceUpdateMsg deviceUpdateMsg = deviceMsgConstructor - .constructDeviceUpdatedMsg(UpdateMsgType.ENTITY_MERGE_RPC_MESSAGE, device, customerId, conflictName); + .constructDeviceUpdatedMsg(UpdateMsgType.ENTITY_MERGE_RPC_MESSAGE, device, conflictName); downlinkMsg = DownlinkMsg.newBuilder() .setDownlinkMsgId(EdgeUtils.nextPositiveInt()) .addDeviceUpdateMsg(deviceUpdateMsg) @@ -98,10 +98,9 @@ public class EntityEdgeProcessor extends BaseEdgeProcessor { case ADDED: // used only for USER entity case UPDATED: case CREDENTIALS_UPDATED: - return pushNotificationToAllRelatedEdges(tenantId, entityId, type, actionType); case ASSIGNED_TO_CUSTOMER: case UNASSIGNED_FROM_CUSTOMER: - return pushNotificationToAllRelatedCustomerEdges(tenantId, edgeNotificationMsg, entityId, actionType, type); + return pushNotificationToAllRelatedEdges(tenantId, entityId, type, actionType); case DELETED: if (edgeId != null) { return saveEdgeEvent(tenantId, edgeId, type, actionType, entityId, null); @@ -123,42 +122,6 @@ public class EntityEdgeProcessor extends BaseEdgeProcessor { } } - private ListenableFuture pushNotificationToAllRelatedCustomerEdges(TenantId tenantId, - TransportProtos.EdgeNotificationMsgProto edgeNotificationMsg, - EntityId entityId, - EdgeEventActionType actionType, - EdgeEventType type) { - PageLink pageLink = new PageLink(DEFAULT_PAGE_SIZE); - PageData pageData; - List> futures = new ArrayList<>(); - do { - pageData = edgeService.findRelatedEdgeIdsByEntityId(tenantId, entityId, pageLink); - if (pageData != null && pageData.getData() != null && !pageData.getData().isEmpty()) { - for (EdgeId relatedEdgeId : pageData.getData()) { - try { - CustomerId customerId = mapper.readValue(edgeNotificationMsg.getBody(), CustomerId.class); - ListenableFuture future = edgeService.findEdgeByIdAsync(tenantId, relatedEdgeId); - futures.add(Futures.transformAsync(future, edge -> { - if (edge != null && edge.getCustomerId() != null && - !edge.getCustomerId().isNullUid() && edge.getCustomerId().equals(customerId)) { - return saveEdgeEvent(tenantId, relatedEdgeId, type, actionType, entityId, null); - } else { - return Futures.immediateFuture(null); - } - }, dbCallbackExecutorService)); - } catch (Exception e) { - log.error("Can't parse customer id from entity body [{}]", edgeNotificationMsg, e); - return Futures.immediateFailedFuture(e); - } - } - if (pageData.hasNext()) { - pageLink = pageLink.nextPageLink(); - } - } - } while (pageData != null && pageData.hasNext()); - return Futures.transform(Futures.allAsList(futures), voids -> null, dbCallbackExecutorService); - } - private EdgeId safeGetEdgeId(TransportProtos.EdgeNotificationMsgProto edgeNotificationMsg) { if (edgeNotificationMsg.getEdgeIdMSB() != 0 && edgeNotificationMsg.getEdgeIdLSB() != 0) { return new EdgeId(new UUID(edgeNotificationMsg.getEdgeIdMSB(), edgeNotificationMsg.getEdgeIdLSB())); diff --git a/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/EntityViewEdgeProcessor.java b/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/EntityViewEdgeProcessor.java index e5d69e36cf..f018272aa0 100644 --- a/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/EntityViewEdgeProcessor.java +++ b/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/EntityViewEdgeProcessor.java @@ -19,10 +19,8 @@ import lombok.extern.slf4j.Slf4j; import org.springframework.stereotype.Component; import org.thingsboard.server.common.data.EdgeUtils; import org.thingsboard.server.common.data.EntityView; -import org.thingsboard.server.common.data.edge.Edge; import org.thingsboard.server.common.data.edge.EdgeEvent; import org.thingsboard.server.common.data.edge.EdgeEventActionType; -import org.thingsboard.server.common.data.id.CustomerId; import org.thingsboard.server.common.data.id.EntityViewId; import org.thingsboard.server.gen.edge.v1.DownlinkMsg; import org.thingsboard.server.gen.edge.v1.EntityViewUpdateMsg; @@ -34,7 +32,7 @@ import org.thingsboard.server.queue.util.TbCoreComponent; @TbCoreComponent public class EntityViewEdgeProcessor extends BaseEdgeProcessor { - public DownlinkMsg processEntityViewToEdge(Edge edge, EdgeEvent edgeEvent, UpdateMsgType msgType, EdgeEventActionType action) { + public DownlinkMsg processEntityViewToEdge(EdgeEvent edgeEvent, UpdateMsgType msgType, EdgeEventActionType action) { EntityViewId entityViewId = new EntityViewId(edgeEvent.getEntityId()); DownlinkMsg downlinkMsg = null; switch (action) { @@ -45,9 +43,8 @@ public class EntityViewEdgeProcessor extends BaseEdgeProcessor { case UNASSIGNED_FROM_CUSTOMER: EntityView entityView = entityViewService.findEntityViewById(edgeEvent.getTenantId(), entityViewId); if (entityView != null) { - CustomerId customerId = getCustomerIdIfEdgeAssignedToCustomer(entityView, edge); EntityViewUpdateMsg entityViewUpdateMsg = - entityViewMsgConstructor.constructEntityViewUpdatedMsg(msgType, entityView, customerId); + entityViewMsgConstructor.constructEntityViewUpdatedMsg(msgType, entityView); downlinkMsg = DownlinkMsg.newBuilder() .setDownlinkMsgId(EdgeUtils.nextPositiveInt()) .addEntityViewUpdateMsg(entityViewUpdateMsg) diff --git a/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/RelationEdgeProcessor.java b/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/RelationEdgeProcessor.java index f86ea8aad2..c73a60e285 100644 --- a/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/RelationEdgeProcessor.java +++ b/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/RelationEdgeProcessor.java @@ -20,6 +20,7 @@ import com.google.common.util.concurrent.Futures; import com.google.common.util.concurrent.ListenableFuture; import lombok.extern.slf4j.Slf4j; import org.springframework.stereotype.Component; +import org.thingsboard.common.util.JacksonUtil; import org.thingsboard.server.common.data.EdgeUtils; import org.thingsboard.server.common.data.EntityType; import org.thingsboard.server.common.data.edge.EdgeEvent; @@ -75,7 +76,7 @@ public class RelationEdgeProcessor extends BaseEdgeProcessor { if (relationUpdateMsg.hasTypeGroup()) { entityRelation.setTypeGroup(RelationTypeGroup.valueOf(relationUpdateMsg.getTypeGroup())); } - entityRelation.setAdditionalInfo(mapper.readTree(relationUpdateMsg.getAdditionalInfo())); + entityRelation.setAdditionalInfo(JacksonUtil.OBJECT_MAPPER.readTree(relationUpdateMsg.getAdditionalInfo())); switch (relationUpdateMsg.getMsgType()) { case ENTITY_CREATED_RPC_MESSAGE: case ENTITY_UPDATED_RPC_MESSAGE: @@ -118,7 +119,7 @@ public class RelationEdgeProcessor extends BaseEdgeProcessor { } public DownlinkMsg processRelationToEdge(EdgeEvent edgeEvent, UpdateMsgType msgType) { - EntityRelation entityRelation = mapper.convertValue(edgeEvent.getBody(), EntityRelation.class); + EntityRelation entityRelation = JacksonUtil.OBJECT_MAPPER.convertValue(edgeEvent.getBody(), EntityRelation.class); RelationUpdateMsg relationUpdateMsg = relationMsgConstructor.constructRelationUpdatedMsg(msgType, entityRelation); return DownlinkMsg.newBuilder() .setDownlinkMsgId(EdgeUtils.nextPositiveInt()) @@ -127,7 +128,7 @@ public class RelationEdgeProcessor extends BaseEdgeProcessor { } public ListenableFuture processRelationNotification(TenantId tenantId, TransportProtos.EdgeNotificationMsgProto edgeNotificationMsg) throws JsonProcessingException { - EntityRelation relation = mapper.readValue(edgeNotificationMsg.getBody(), EntityRelation.class); + EntityRelation relation = JacksonUtil.OBJECT_MAPPER.readValue(edgeNotificationMsg.getBody(), EntityRelation.class); if (relation.getFrom().getEntityType().equals(EntityType.EDGE) || relation.getTo().getEntityType().equals(EntityType.EDGE)) { return Futures.immediateFuture(null); @@ -146,7 +147,7 @@ public class RelationEdgeProcessor extends BaseEdgeProcessor { EdgeEventType.RELATION, EdgeEventActionType.valueOf(edgeNotificationMsg.getAction()), null, - mapper.valueToTree(relation))); + JacksonUtil.OBJECT_MAPPER.valueToTree(relation))); } return Futures.transform(Futures.allAsList(futures), voids -> null, dbCallbackExecutorService); } diff --git a/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/TelemetryEdgeProcessor.java b/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/TelemetryEdgeProcessor.java index ad6446cb95..9c75373cad 100644 --- a/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/TelemetryEdgeProcessor.java +++ b/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/TelemetryEdgeProcessor.java @@ -27,6 +27,7 @@ import lombok.extern.slf4j.Slf4j; import org.apache.commons.lang3.tuple.ImmutablePair; import org.apache.commons.lang3.tuple.Pair; import org.springframework.stereotype.Component; +import org.thingsboard.common.util.JacksonUtil; import org.thingsboard.rule.engine.api.msg.DeviceAttributesEventNotificationMsg; import org.thingsboard.server.common.data.DataConstants; import org.thingsboard.server.common.data.Device; @@ -44,7 +45,6 @@ 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.EntityViewId; -import org.thingsboard.server.common.data.id.QueueId; import org.thingsboard.server.common.data.id.RuleChainId; import org.thingsboard.server.common.data.id.TenantId; import org.thingsboard.server.common.data.id.UserId; @@ -332,7 +332,8 @@ public class TelemetryEdgeProcessor extends BaseEdgeProcessor { log.warn("Unsupported edge event type [{}]", edgeEvent); return null; } - return constructEntityDataProtoMsg(entityId, edgeEvent.getAction(), JsonUtils.parse(mapper.writeValueAsString(edgeEvent.getBody()))); + return constructEntityDataProtoMsg(entityId, edgeEvent.getAction(), + JsonUtils.parse(JacksonUtil.OBJECT_MAPPER.writeValueAsString(edgeEvent.getBody()))); } private DownlinkMsg constructEntityDataProtoMsg(EntityId entityId, EdgeEventActionType actionType, JsonElement entityData) { diff --git a/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/UserEdgeProcessor.java b/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/UserEdgeProcessor.java index a64fb133f1..6c47dc936a 100644 --- a/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/UserEdgeProcessor.java +++ b/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/UserEdgeProcessor.java @@ -43,10 +43,9 @@ public class UserEdgeProcessor extends BaseEdgeProcessor { case UPDATED: User user = userService.findUserById(edgeEvent.getTenantId(), userId); if (user != null) { - CustomerId customerId = getCustomerIdIfEdgeAssignedToCustomer(user, edge); downlinkMsg = DownlinkMsg.newBuilder() .setDownlinkMsgId(EdgeUtils.nextPositiveInt()) - .addUserUpdateMsg(userMsgConstructor.constructUserUpdatedMsg(msgType, user, customerId)) + .addUserUpdateMsg(userMsgConstructor.constructUserUpdatedMsg(msgType, user)) .build(); } break; diff --git a/application/src/main/java/org/thingsboard/server/service/edge/rpc/sync/DefaultEdgeRequestsService.java b/application/src/main/java/org/thingsboard/server/service/edge/rpc/sync/DefaultEdgeRequestsService.java index e66d7d5595..d3732cd8a6 100644 --- a/application/src/main/java/org/thingsboard/server/service/edge/rpc/sync/DefaultEdgeRequestsService.java +++ b/application/src/main/java/org/thingsboard/server/service/edge/rpc/sync/DefaultEdgeRequestsService.java @@ -16,7 +16,6 @@ package org.thingsboard.server.service.edge.rpc.sync; import com.fasterxml.jackson.databind.JsonNode; -import com.fasterxml.jackson.databind.ObjectMapper; import com.fasterxml.jackson.databind.node.ObjectNode; import com.google.common.util.concurrent.FutureCallback; import com.google.common.util.concurrent.Futures; @@ -27,6 +26,7 @@ import org.checkerframework.checker.nullness.qual.Nullable; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.context.annotation.Lazy; import org.springframework.stereotype.Service; +import org.thingsboard.common.util.JacksonUtil; import org.thingsboard.server.cluster.TbClusterService; import org.thingsboard.server.common.data.Device; import org.thingsboard.server.common.data.DeviceProfile; @@ -88,8 +88,6 @@ import java.util.UUID; @Slf4j public class DefaultEdgeRequestsService implements EdgeRequestsService { - private static final ObjectMapper mapper = new ObjectMapper(); - private static final int DEFAULT_PAGE_SIZE = 1000; @Autowired @@ -163,7 +161,7 @@ public class DefaultEdgeRequestsService implements EdgeRequestsService { try { Map entityData = new HashMap<>(); - ObjectNode attributes = mapper.createObjectNode(); + ObjectNode attributes = JacksonUtil.OBJECT_MAPPER.createObjectNode(); for (AttributeKvEntry attr : ssAttributes) { if (DefaultDeviceStateService.PERSISTENT_ATTRIBUTES.contains(attr.getKey())) { continue; @@ -180,7 +178,7 @@ public class DefaultEdgeRequestsService implements EdgeRequestsService { } entityData.put("kv", attributes); entityData.put("scope", scope); - JsonNode body = mapper.valueToTree(entityData); + JsonNode body = JacksonUtil.OBJECT_MAPPER.valueToTree(entityData); log.debug("Sending attributes data msg, entityId [{}], attributes [{}]", entityId, body); ListenableFuture future = saveEdgeEvent(tenantId, edge.getId(), type, EdgeEventActionType.ATTRIBUTES_UPDATED, entityId, body); Futures.addCallback(future, new FutureCallback<>() { @@ -242,7 +240,7 @@ public class DefaultEdgeRequestsService implements EdgeRequestsService { EdgeEventType.RELATION, EdgeEventActionType.ADDED, null, - mapper.valueToTree(relation))); + JacksonUtil.OBJECT_MAPPER.valueToTree(relation))); } } catch (Exception e) { String errMsg = String.format("[%s] Exception during loading relation [%s] to edge on sync!", edge.getId(), relation); diff --git a/application/src/main/java/org/thingsboard/server/service/entitiy/DefaultTbNotificationEntityService.java b/application/src/main/java/org/thingsboard/server/service/entitiy/DefaultTbNotificationEntityService.java index 187d8deabd..8c4b62a137 100644 --- a/application/src/main/java/org/thingsboard/server/service/entitiy/DefaultTbNotificationEntityService.java +++ b/application/src/main/java/org/thingsboard/server/service/entitiy/DefaultTbNotificationEntityService.java @@ -135,7 +135,7 @@ public class DefaultTbNotificationEntityService implements TbNotificationEntityS logEntityAction(tenantId, entityId, entity, customerId, actionType, user, additionalInfo); if (sendToEdge) { - sendEntityAssignToCustomerNotificationMsg(tenantId, entityId, customerId, edgeTypeByActionType(actionType)); + sendEntityNotificationMsg(tenantId, entityId, edgeTypeByActionType(actionType)); } } @@ -203,24 +203,17 @@ public class DefaultTbNotificationEntityService implements TbNotificationEntityS } @Override - public void notifyEdge(TenantId tenantId, EdgeId edgeId, CustomerId customerId, Edge edge, - ActionType actionType, User user, Object... additionalInfo) { + public void notifyCreateOrUpdateOrDeleteEdge(TenantId tenantId, EdgeId edgeId, CustomerId customerId, Edge edge, + ActionType actionType, User user, Object... additionalInfo) { ComponentLifecycleEvent lifecycleEvent; - EdgeEventActionType edgeEventActionType = null; switch (actionType) { case ADDED: lifecycleEvent = ComponentLifecycleEvent.CREATED; break; case UPDATED: - lifecycleEvent = ComponentLifecycleEvent.UPDATED; - break; case ASSIGNED_TO_CUSTOMER: - lifecycleEvent = ComponentLifecycleEvent.UPDATED; - edgeEventActionType = EdgeEventActionType.ASSIGNED_TO_CUSTOMER; - break; case UNASSIGNED_FROM_CUSTOMER: lifecycleEvent = ComponentLifecycleEvent.UPDATED; - edgeEventActionType = EdgeEventActionType.UNASSIGNED_FROM_CUSTOMER; break; case DELETED: lifecycleEvent = ComponentLifecycleEvent.DELETED; @@ -228,14 +221,8 @@ public class DefaultTbNotificationEntityService implements TbNotificationEntityS default: throw new IllegalArgumentException("Unknown actionType: " + actionType); } - tbClusterService.broadcastEntityStateChangeEvent(tenantId, edgeId, lifecycleEvent); logEntityAction(tenantId, edgeId, edge, customerId, actionType, user, additionalInfo); - - //Send notification to edge - if (edgeEventActionType != null) { - sendEntityAssignToCustomerNotificationMsg(tenantId, edgeId, customerId, edgeEventActionType); - } } @Override @@ -274,14 +261,6 @@ public class DefaultTbNotificationEntityService implements TbNotificationEntityS sendNotificationMsgToEdge(tenantId, null, entityId, null, null, action); } - private void sendEntityAssignToCustomerNotificationMsg(TenantId tenantId, EntityId entityId, CustomerId customerId, EdgeEventActionType action) { - try { - sendNotificationMsgToEdge(tenantId, null, entityId, JacksonUtil.toString(customerId), null, action); - } catch (Exception e) { - log.warn("Failed to push assign/unassign to/from customer to core: {}", customerId, e); - } - } - private void sendAlarmDeleteNotificationMsg(TenantId tenantId, Alarm alarm, List edgeIds, String body) { try { sendDeleteNotificationMsg(tenantId, alarm.getId(), edgeIds, body); diff --git a/application/src/main/java/org/thingsboard/server/service/entitiy/TbNotificationEntityService.java b/application/src/main/java/org/thingsboard/server/service/entitiy/TbNotificationEntityService.java index 7e929254a6..dc6e148cab 100644 --- a/application/src/main/java/org/thingsboard/server/service/entitiy/TbNotificationEntityService.java +++ b/application/src/main/java/org/thingsboard/server/service/entitiy/TbNotificationEntityService.java @@ -97,8 +97,8 @@ public interface TbNotificationEntityService { void notifyAssignDeviceToTenant(TenantId tenantId, TenantId newTenantId, DeviceId deviceId, CustomerId customerId, Device device, Tenant tenant, User user, Object... additionalInfo); - void notifyEdge(TenantId tenantId, EdgeId edgeId, CustomerId customerId, Edge edge, ActionType actionType, - User user, Object... additionalInfo); + void notifyCreateOrUpdateOrDeleteEdge(TenantId tenantId, EdgeId edgeId, CustomerId customerId, Edge edge, ActionType actionType, + User user, Object... additionalInfo); void notifyCreateOrUpdateAlarm(Alarm alarm, ActionType actionType, User user, Object... additionalInfo); diff --git a/application/src/main/java/org/thingsboard/server/service/entitiy/edge/DefaultTbEdgeService.java b/application/src/main/java/org/thingsboard/server/service/entitiy/edge/DefaultTbEdgeService.java index b06466b01f..13591baa32 100644 --- a/application/src/main/java/org/thingsboard/server/service/entitiy/edge/DefaultTbEdgeService.java +++ b/application/src/main/java/org/thingsboard/server/service/entitiy/edge/DefaultTbEdgeService.java @@ -57,7 +57,7 @@ public class DefaultTbEdgeService extends AbstractTbEntityService implements TbE edgeService.assignDefaultRuleChainsToEdge(tenantId, edgeId); } - notificationEntityService.notifyEdge(tenantId, edgeId, savedEdge.getCustomerId(), savedEdge, actionType, user); + notificationEntityService.notifyCreateOrUpdateOrDeleteEdge(tenantId, edgeId, savedEdge.getCustomerId(), savedEdge, actionType, user); return savedEdge; } catch (Exception e) { @@ -72,7 +72,7 @@ public class DefaultTbEdgeService extends AbstractTbEntityService implements TbE TenantId tenantId = edge.getTenantId(); try { edgeService.deleteEdge(tenantId, edgeId); - notificationEntityService.notifyEdge(tenantId, edgeId, edge.getCustomerId(), edge, ActionType.DELETED, user, edgeId.toString()); + notificationEntityService.notifyCreateOrUpdateOrDeleteEdge(tenantId, edgeId, edge.getCustomerId(), edge, ActionType.DELETED, user, edgeId.toString()); } catch (Exception e) { notificationEntityService.logEntityAction(tenantId, emptyId(EntityType.EDGE), ActionType.DELETED, user, e, edgeId.toString()); @@ -85,7 +85,7 @@ public class DefaultTbEdgeService extends AbstractTbEntityService implements TbE CustomerId customerId = customer.getId(); try { Edge savedEdge = checkNotNull(edgeService.assignEdgeToCustomer(tenantId, edgeId, customerId)); - notificationEntityService.notifyEdge(tenantId, edgeId, customerId, savedEdge, ActionType.ASSIGNED_TO_CUSTOMER, user, + notificationEntityService.notifyCreateOrUpdateOrDeleteEdge(tenantId, edgeId, customerId, savedEdge, ActionType.ASSIGNED_TO_CUSTOMER, user, edgeId.toString(), customerId.toString(), customer.getName()); return savedEdge; @@ -104,7 +104,7 @@ public class DefaultTbEdgeService extends AbstractTbEntityService implements TbE try { Edge savedEdge = checkNotNull(edgeService.unassignEdgeFromCustomer(tenantId, edgeId)); - notificationEntityService.notifyEdge(tenantId, edgeId, customerId, savedEdge, ActionType.UNASSIGNED_FROM_CUSTOMER, user, + notificationEntityService.notifyCreateOrUpdateOrDeleteEdge(tenantId, edgeId, customerId, savedEdge, ActionType.UNASSIGNED_FROM_CUSTOMER, user, edgeId.toString(), customerId.toString(), customer.getName()); return savedEdge; } catch (Exception e) { @@ -121,7 +121,7 @@ public class DefaultTbEdgeService extends AbstractTbEntityService implements TbE try { Edge savedEdge = checkNotNull(edgeService.assignEdgeToCustomer(tenantId, edgeId, customerId)); - notificationEntityService.notifyEdge(tenantId, edgeId, customerId, savedEdge, ActionType.ASSIGNED_TO_CUSTOMER, user, + notificationEntityService.notifyCreateOrUpdateOrDeleteEdge(tenantId, edgeId, customerId, savedEdge, ActionType.ASSIGNED_TO_CUSTOMER, user, edgeId.toString(), customerId.toString(), publicCustomer.getName()); return savedEdge; @@ -138,7 +138,7 @@ public class DefaultTbEdgeService extends AbstractTbEntityService implements TbE EdgeId edgeId = edge.getId(); try { Edge updatedEdge = edgeNotificationService.setEdgeRootRuleChain(tenantId, edge, ruleChainId); - notificationEntityService.notifyEdge(tenantId, edgeId, null, updatedEdge, ActionType.UPDATED, user); + notificationEntityService.logEntityAction(tenantId, edgeId, edge, null, ActionType.UPDATED, user); return updatedEdge; } catch (Exception e) { notificationEntityService.logEntityAction(tenantId, emptyId(EntityType.EDGE), diff --git a/common/edge-api/src/main/proto/edge.proto b/common/edge-api/src/main/proto/edge.proto index 7272e6f96a..06ca2e33a5 100644 --- a/common/edge-api/src/main/proto/edge.proto +++ b/common/edge-api/src/main/proto/edge.proto @@ -183,6 +183,7 @@ message DashboardUpdateMsg { optional int64 customerIdLSB = 5; string title = 6; string configuration = 7; + optional string assignedCustomers = 8; } message DeviceUpdateMsg {