Browse Source

Added support for multiple customers on edge

pull/7395/head
Volodymyr Babak 4 years ago
parent
commit
8e27ffab4b
  1. 4
      application/src/main/java/org/thingsboard/server/service/edge/EdgeContextComponent.java
  2. 8
      application/src/main/java/org/thingsboard/server/service/edge/rpc/EdgeGrpcSession.java
  3. 14
      application/src/main/java/org/thingsboard/server/service/edge/rpc/EdgeSyncCursor.java
  4. 9
      application/src/main/java/org/thingsboard/server/service/edge/rpc/constructor/AssetMsgConstructor.java
  5. 8
      application/src/main/java/org/thingsboard/server/service/edge/rpc/constructor/DashboardMsgConstructor.java
  6. 8
      application/src/main/java/org/thingsboard/server/service/edge/rpc/constructor/DeviceMsgConstructor.java
  7. 9
      application/src/main/java/org/thingsboard/server/service/edge/rpc/constructor/EntityViewMsgConstructor.java
  8. 9
      application/src/main/java/org/thingsboard/server/service/edge/rpc/constructor/UserMsgConstructor.java
  9. 24
      application/src/main/java/org/thingsboard/server/service/edge/rpc/fetch/CustomerEdgeEventFetcher.java
  10. 41
      application/src/main/java/org/thingsboard/server/service/edge/rpc/fetch/CustomerUsersEdgeEventFetcher.java
  11. 34
      application/src/main/java/org/thingsboard/server/service/edge/rpc/fetch/TenantAdminUsersEdgeEventFetcher.java
  12. 6
      application/src/main/java/org/thingsboard/server/service/edge/rpc/fetch/UsersEdgeEventFetcher.java
  13. 3
      application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/AdminSettingsEdgeProcessor.java
  14. 12
      application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/AlarmEdgeProcessor.java
  15. 7
      application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/AssetEdgeProcessor.java
  16. 15
      application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/BaseEdgeProcessor.java
  17. 8
      application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/DashboardEdgeProcessor.java
  18. 9
      application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/DeviceEdgeProcessor.java
  19. 38
      application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/EdgeProcessor.java
  20. 43
      application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/EntityEdgeProcessor.java
  21. 7
      application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/EntityViewEdgeProcessor.java
  22. 9
      application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/RelationEdgeProcessor.java
  23. 5
      application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/TelemetryEdgeProcessor.java
  24. 3
      application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/UserEdgeProcessor.java
  25. 10
      application/src/main/java/org/thingsboard/server/service/edge/rpc/sync/DefaultEdgeRequestsService.java
  26. 27
      application/src/main/java/org/thingsboard/server/service/entitiy/DefaultTbNotificationEntityService.java
  27. 4
      application/src/main/java/org/thingsboard/server/service/entitiy/TbNotificationEntityService.java
  28. 12
      application/src/main/java/org/thingsboard/server/service/entitiy/edge/DefaultTbEdgeService.java
  29. 1
      common/edge-api/src/main/proto/edge.proto

4
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.cluster.TbClusterService;
import org.thingsboard.server.dao.asset.AssetService; import org.thingsboard.server.dao.asset.AssetService;
import org.thingsboard.server.dao.attributes.AttributesService; 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.dashboard.DashboardService;
import org.thingsboard.server.dao.device.DeviceProfileService; import org.thingsboard.server.dao.device.DeviceProfileService;
import org.thingsboard.server.dao.edge.EdgeEventService; import org.thingsboard.server.dao.edge.EdgeEventService;
@ -95,6 +96,9 @@ public class EdgeContextComponent {
@Autowired @Autowired
private UserService userService; private UserService userService;
@Autowired
private CustomerService customerService;
@Autowired @Autowired
private WidgetsBundleService widgetsBundleService; private WidgetsBundleService widgetsBundleService;

8
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) { public void startSyncProcess(TenantId tenantId, EdgeId edgeId) {
log.trace("[{}][{}] Staring edge sync process", tenantId, edgeId); log.trace("[{}][{}] Staring edge sync process", tenantId, edgeId);
syncCompleted = false; syncCompleted = false;
doSync(new EdgeSyncCursor(ctx, edge)); doSync(new EdgeSyncCursor(ctx));
} }
private void doSync(EdgeSyncCursor cursor) { private void doSync(EdgeSyncCursor cursor) {
@ -514,11 +514,11 @@ public final class EdgeGrpcSession implements Closeable {
case DEVICE_PROFILE: case DEVICE_PROFILE:
return ctx.getDeviceProfileProcessor().processDeviceProfileToEdge(edgeEvent, msgType, action); return ctx.getDeviceProfileProcessor().processDeviceProfileToEdge(edgeEvent, msgType, action);
case ASSET: case ASSET:
return ctx.getAssetProcessor().processAssetToEdge(edge, edgeEvent, msgType, action); return ctx.getAssetProcessor().processAssetToEdge(edgeEvent, msgType, action);
case ENTITY_VIEW: case ENTITY_VIEW:
return ctx.getEntityViewProcessor().processEntityViewToEdge(edge, edgeEvent, msgType, action); return ctx.getEntityViewProcessor().processEntityViewToEdge(edgeEvent, msgType, action);
case DASHBOARD: case DASHBOARD:
return ctx.getDashboardProcessor().processDashboardToEdge(edge, edgeEvent, msgType, action); return ctx.getDashboardProcessor().processDashboardToEdge(edgeEvent, msgType, action);
case CUSTOMER: case CUSTOMER:
return ctx.getCustomerProcessor().processCustomerToEdge(edgeEvent, msgType, action); return ctx.getCustomerProcessor().processCustomerToEdge(edgeEvent, msgType, action);
case RULE_CHAIN: case RULE_CHAIN:

14
application/src/main/java/org/thingsboard/server/service/edge/rpc/EdgeSyncCursor.java

@ -15,13 +15,10 @@
*/ */
package org.thingsboard.server.service.edge.rpc; 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.EdgeContextComponent;
import org.thingsboard.server.service.edge.rpc.fetch.AdminSettingsEdgeEventFetcher; 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.AssetsEdgeEventFetcher;
import org.thingsboard.server.service.edge.rpc.fetch.CustomerEdgeEventFetcher; 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.DashboardsEdgeEventFetcher;
import org.thingsboard.server.service.edge.rpc.fetch.DeviceProfilesEdgeEventFetcher; import org.thingsboard.server.service.edge.rpc.fetch.DeviceProfilesEdgeEventFetcher;
import org.thingsboard.server.service.edge.rpc.fetch.EdgeEventFetcher; 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.QueuesEdgeEventFetcher;
import org.thingsboard.server.service.edge.rpc.fetch.RuleChainsEdgeEventFetcher; 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.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.TenantWidgetsBundlesEdgeEventFetcher;
import org.thingsboard.server.service.edge.rpc.fetch.UsersEdgeEventFetcher;
import java.util.LinkedList; import java.util.LinkedList;
import java.util.List; import java.util.List;
@ -42,16 +39,13 @@ public class EdgeSyncCursor {
int currentIdx = 0; int currentIdx = 0;
public EdgeSyncCursor(EdgeContextComponent ctx, Edge edge) { public EdgeSyncCursor(EdgeContextComponent ctx) {
fetchers.add(new QueuesEdgeEventFetcher(ctx.getQueueService())); fetchers.add(new QueuesEdgeEventFetcher(ctx.getQueueService()));
fetchers.add(new RuleChainsEdgeEventFetcher(ctx.getRuleChainService())); fetchers.add(new RuleChainsEdgeEventFetcher(ctx.getRuleChainService()));
fetchers.add(new AdminSettingsEdgeEventFetcher(ctx.getAdminSettingsService(), ctx.getFreemarkerConfig())); fetchers.add(new AdminSettingsEdgeEventFetcher(ctx.getAdminSettingsService(), ctx.getFreemarkerConfig()));
fetchers.add(new DeviceProfilesEdgeEventFetcher(ctx.getDeviceProfileService())); fetchers.add(new DeviceProfilesEdgeEventFetcher(ctx.getDeviceProfileService()));
fetchers.add(new TenantAdminUsersEdgeEventFetcher(ctx.getUserService())); fetchers.add(new CustomerEdgeEventFetcher(ctx.getCustomerService()));
if (edge.getCustomerId() != null && !EntityId.NULL_UUID.equals(edge.getCustomerId().getId())) { fetchers.add(new UsersEdgeEventFetcher(ctx.getUserService()));
fetchers.add(new CustomerEdgeEventFetcher());
fetchers.add(new CustomerUsersEdgeEventFetcher(ctx.getUserService(), edge.getCustomerId()));
}
fetchers.add(new AssetsEdgeEventFetcher(ctx.getAssetService())); fetchers.add(new AssetsEdgeEventFetcher(ctx.getAssetService()));
fetchers.add(new SystemWidgetsBundlesEdgeEventFetcher(ctx.getWidgetsBundleService())); fetchers.add(new SystemWidgetsBundlesEdgeEventFetcher(ctx.getWidgetsBundleService()));
fetchers.add(new TenantWidgetsBundlesEdgeEventFetcher(ctx.getWidgetsBundleService())); fetchers.add(new TenantWidgetsBundlesEdgeEventFetcher(ctx.getWidgetsBundleService()));

9
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.common.util.JacksonUtil;
import org.thingsboard.server.common.data.asset.Asset; import org.thingsboard.server.common.data.asset.Asset;
import org.thingsboard.server.common.data.id.AssetId; 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.AssetUpdateMsg;
import org.thingsboard.server.gen.edge.v1.UpdateMsgType; import org.thingsboard.server.gen.edge.v1.UpdateMsgType;
import org.thingsboard.server.queue.util.TbCoreComponent; import org.thingsboard.server.queue.util.TbCoreComponent;
@ -28,7 +27,7 @@ import org.thingsboard.server.queue.util.TbCoreComponent;
@TbCoreComponent @TbCoreComponent
public class AssetMsgConstructor { public class AssetMsgConstructor {
public AssetUpdateMsg constructAssetUpdatedMsg(UpdateMsgType msgType, Asset asset, CustomerId customerId) { public AssetUpdateMsg constructAssetUpdatedMsg(UpdateMsgType msgType, Asset asset) {
AssetUpdateMsg.Builder builder = AssetUpdateMsg.newBuilder() AssetUpdateMsg.Builder builder = AssetUpdateMsg.newBuilder()
.setMsgType(msgType) .setMsgType(msgType)
.setIdMSB(asset.getId().getId().getMostSignificantBits()) .setIdMSB(asset.getId().getId().getMostSignificantBits())
@ -38,9 +37,9 @@ public class AssetMsgConstructor {
if (asset.getLabel() != null) { if (asset.getLabel() != null) {
builder.setLabel(asset.getLabel()); builder.setLabel(asset.getLabel());
} }
if (customerId != null) { if (asset.getCustomerId() != null) {
builder.setCustomerIdMSB(customerId.getId().getMostSignificantBits()); builder.setCustomerIdMSB(asset.getCustomerId().getId().getMostSignificantBits());
builder.setCustomerIdLSB(customerId.getId().getLeastSignificantBits()); builder.setCustomerIdLSB(asset.getCustomerId().getId().getLeastSignificantBits());
} }
if (asset.getAdditionalInfo() != null) { if (asset.getAdditionalInfo() != null) {
builder.setAdditionalInfo(JacksonUtil.toString(asset.getAdditionalInfo())); builder.setAdditionalInfo(JacksonUtil.toString(asset.getAdditionalInfo()));

8
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.springframework.stereotype.Component;
import org.thingsboard.common.util.JacksonUtil; import org.thingsboard.common.util.JacksonUtil;
import org.thingsboard.server.common.data.Dashboard; 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.common.data.id.DashboardId;
import org.thingsboard.server.gen.edge.v1.DashboardUpdateMsg; import org.thingsboard.server.gen.edge.v1.DashboardUpdateMsg;
import org.thingsboard.server.gen.edge.v1.UpdateMsgType; import org.thingsboard.server.gen.edge.v1.UpdateMsgType;
@ -28,16 +27,15 @@ import org.thingsboard.server.queue.util.TbCoreComponent;
@TbCoreComponent @TbCoreComponent
public class DashboardMsgConstructor { public class DashboardMsgConstructor {
public DashboardUpdateMsg constructDashboardUpdatedMsg(UpdateMsgType msgType, Dashboard dashboard, CustomerId customerId) { public DashboardUpdateMsg constructDashboardUpdatedMsg(UpdateMsgType msgType, Dashboard dashboard) {
DashboardUpdateMsg.Builder builder = DashboardUpdateMsg.newBuilder() DashboardUpdateMsg.Builder builder = DashboardUpdateMsg.newBuilder()
.setMsgType(msgType) .setMsgType(msgType)
.setIdMSB(dashboard.getId().getId().getMostSignificantBits()) .setIdMSB(dashboard.getId().getId().getMostSignificantBits())
.setIdLSB(dashboard.getId().getId().getLeastSignificantBits()) .setIdLSB(dashboard.getId().getId().getLeastSignificantBits())
.setTitle(dashboard.getTitle()) .setTitle(dashboard.getTitle())
.setConfiguration(JacksonUtil.toString(dashboard.getConfiguration())); .setConfiguration(JacksonUtil.toString(dashboard.getConfiguration()));
if (customerId != null) { if (dashboard.getAssignedCustomers() != null) {
builder.setCustomerIdMSB(customerId.getId().getMostSignificantBits()); builder.setAssignedCustomers(JacksonUtil.toString(dashboard.getAssignedCustomers()));
builder.setCustomerIdLSB(customerId.getId().getLeastSignificantBits());
} }
return builder.build(); return builder.build();
} }

8
application/src/main/java/org/thingsboard/server/service/edge/rpc/constructor/DeviceMsgConstructor.java

@ -41,7 +41,7 @@ public class DeviceMsgConstructor {
@Autowired @Autowired
private DataDecodingEncodingService dataDecodingEncodingService; 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() DeviceUpdateMsg.Builder builder = DeviceUpdateMsg.newBuilder()
.setMsgType(msgType) .setMsgType(msgType)
.setIdMSB(device.getId().getId().getMostSignificantBits()) .setIdMSB(device.getId().getId().getMostSignificantBits())
@ -51,9 +51,9 @@ public class DeviceMsgConstructor {
if (device.getLabel() != null) { if (device.getLabel() != null) {
builder.setLabel(device.getLabel()); builder.setLabel(device.getLabel());
} }
if (customerId != null) { if (device.getCustomerId() != null) {
builder.setCustomerIdMSB(customerId.getId().getMostSignificantBits()); builder.setCustomerIdMSB(device.getCustomerId().getId().getMostSignificantBits());
builder.setCustomerIdLSB(customerId.getId().getLeastSignificantBits()); builder.setCustomerIdLSB(device.getCustomerId().getId().getLeastSignificantBits());
} }
if (device.getDeviceProfileId() != null) { if (device.getDeviceProfileId() != null) {
builder.setDeviceProfileIdMSB(device.getDeviceProfileId().getId().getMostSignificantBits()); builder.setDeviceProfileIdMSB(device.getDeviceProfileId().getId().getMostSignificantBits());

9
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.springframework.stereotype.Component;
import org.thingsboard.common.util.JacksonUtil; import org.thingsboard.common.util.JacksonUtil;
import org.thingsboard.server.common.data.EntityView; 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.common.data.id.EntityViewId;
import org.thingsboard.server.gen.edge.v1.EdgeEntityType; import org.thingsboard.server.gen.edge.v1.EdgeEntityType;
import org.thingsboard.server.gen.edge.v1.EntityViewUpdateMsg; import org.thingsboard.server.gen.edge.v1.EntityViewUpdateMsg;
@ -29,7 +28,7 @@ import org.thingsboard.server.queue.util.TbCoreComponent;
@TbCoreComponent @TbCoreComponent
public class EntityViewMsgConstructor { public class EntityViewMsgConstructor {
public EntityViewUpdateMsg constructEntityViewUpdatedMsg(UpdateMsgType msgType, EntityView entityView, CustomerId customerId) { public EntityViewUpdateMsg constructEntityViewUpdatedMsg(UpdateMsgType msgType, EntityView entityView) {
EdgeEntityType entityType; EdgeEntityType entityType;
switch (entityView.getEntityId().getEntityType()) { switch (entityView.getEntityId().getEntityType()) {
case DEVICE: case DEVICE:
@ -50,9 +49,9 @@ public class EntityViewMsgConstructor {
.setEntityIdMSB(entityView.getEntityId().getId().getMostSignificantBits()) .setEntityIdMSB(entityView.getEntityId().getId().getMostSignificantBits())
.setEntityIdLSB(entityView.getEntityId().getId().getLeastSignificantBits()) .setEntityIdLSB(entityView.getEntityId().getId().getLeastSignificantBits())
.setEntityType(entityType); .setEntityType(entityType);
if (customerId != null) { if (entityView.getCustomerId() != null) {
builder.setCustomerIdMSB(customerId.getId().getMostSignificantBits()); builder.setCustomerIdMSB(entityView.getCustomerId().getId().getMostSignificantBits());
builder.setCustomerIdLSB(customerId.getId().getLeastSignificantBits()); builder.setCustomerIdLSB(entityView.getCustomerId().getId().getLeastSignificantBits());
} }
if (entityView.getAdditionalInfo() != null) { if (entityView.getAdditionalInfo() != null) {
builder.setAdditionalInfo(JacksonUtil.toString(entityView.getAdditionalInfo())); builder.setAdditionalInfo(JacksonUtil.toString(entityView.getAdditionalInfo()));

9
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.springframework.stereotype.Component;
import org.thingsboard.common.util.JacksonUtil; import org.thingsboard.common.util.JacksonUtil;
import org.thingsboard.server.common.data.User; 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.id.UserId;
import org.thingsboard.server.common.data.security.UserCredentials; import org.thingsboard.server.common.data.security.UserCredentials;
import org.thingsboard.server.gen.edge.v1.UpdateMsgType; import org.thingsboard.server.gen.edge.v1.UpdateMsgType;
@ -30,16 +29,16 @@ import org.thingsboard.server.queue.util.TbCoreComponent;
@TbCoreComponent @TbCoreComponent
public class UserMsgConstructor { public class UserMsgConstructor {
public UserUpdateMsg constructUserUpdatedMsg(UpdateMsgType msgType, User user, CustomerId customerId) { public UserUpdateMsg constructUserUpdatedMsg(UpdateMsgType msgType, User user) {
UserUpdateMsg.Builder builder = UserUpdateMsg.newBuilder() UserUpdateMsg.Builder builder = UserUpdateMsg.newBuilder()
.setMsgType(msgType) .setMsgType(msgType)
.setIdMSB(user.getId().getId().getMostSignificantBits()) .setIdMSB(user.getId().getId().getMostSignificantBits())
.setIdLSB(user.getId().getId().getLeastSignificantBits()) .setIdLSB(user.getId().getId().getLeastSignificantBits())
.setEmail(user.getEmail()) .setEmail(user.getEmail())
.setAuthority(user.getAuthority().name()); .setAuthority(user.getAuthority().name());
if (customerId != null) { if (user.getCustomerId() != null) {
builder.setCustomerIdMSB(customerId.getId().getMostSignificantBits()); builder.setCustomerIdMSB(user.getCustomerId().getId().getMostSignificantBits());
builder.setCustomerIdLSB(customerId.getId().getLeastSignificantBits()); builder.setCustomerIdLSB(user.getCustomerId().getId().getLeastSignificantBits());
} }
if (user.getFirstName() != null) { if (user.getFirstName() != null) {
builder.setFirstName(user.getFirstName()); builder.setFirstName(user.getFirstName());

24
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.AllArgsConstructor;
import lombok.extern.slf4j.Slf4j; 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.EdgeUtils;
import org.thingsboard.server.common.data.edge.Edge; import org.thingsboard.server.common.data.edge.Edge;
import org.thingsboard.server.common.data.edge.EdgeEvent; 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.id.TenantId;
import org.thingsboard.server.common.data.page.PageData; import org.thingsboard.server.common.data.page.PageData;
import org.thingsboard.server.common.data.page.PageLink; 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 @Slf4j
public class CustomerEdgeEventFetcher implements EdgeEventFetcher { @AllArgsConstructor
public class CustomerEdgeEventFetcher extends BasePageableEdgeEventFetcher<Customer> {
private final CustomerService customerService;
@Override @Override
public PageLink getPageLink(int pageSize) { PageData<Customer> fetchPageData(TenantId tenantId, Edge edge, PageLink pageLink) {
return null; return customerService.findCustomersByTenantId(tenantId, pageLink);
} }
@Override @Override
public PageData<EdgeEvent> fetchEdgeEvents(TenantId tenantId, Edge edge, PageLink pageLink) { EdgeEvent constructEdgeEvent(TenantId tenantId, Edge edge, Customer customer) {
List<EdgeEvent> result = new ArrayList<>(); return EdgeUtils.constructEdgeEvent(edge.getTenantId(), edge.getId(), EdgeEventType.CUSTOMER,
result.add(EdgeUtils.constructEdgeEvent(edge.getTenantId(), edge.getId(), EdgeEventActionType.ADDED, customer.getId(), null);
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);
} }
} }

41
application/src/main/java/org/thingsboard/server/service/edge/rpc/fetch/CustomerUsersEdgeEventFetcher.java

@ -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<User> findUsers(TenantId tenantId, PageLink pageLink) {
return userService.findCustomerUsers(tenantId, customerId, pageLink);
}
}

34
application/src/main/java/org/thingsboard/server/service/edge/rpc/fetch/TenantAdminUsersEdgeEventFetcher.java

@ -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<User> findUsers(TenantId tenantId, PageLink pageLink) {
return userService.findTenantAdmins(tenantId, pageLink);
}
}

6
application/src/main/java/org/thingsboard/server/service/edge/rpc/fetch/BaseUsersEdgeEventFetcher.java → 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 @Slf4j
@AllArgsConstructor @AllArgsConstructor
public abstract class BaseUsersEdgeEventFetcher extends BasePageableEdgeEventFetcher<User> { public class UsersEdgeEventFetcher extends BasePageableEdgeEventFetcher<User> {
protected final UserService userService; protected final UserService userService;
@Override @Override
PageData<User> fetchPageData(TenantId tenantId, Edge edge, PageLink pageLink) { PageData<User> fetchPageData(TenantId tenantId, Edge edge, PageLink pageLink) {
return findUsers(tenantId, pageLink); return userService.findUsersByTenantId(tenantId, pageLink);
} }
@Override @Override
@ -44,6 +44,4 @@ public abstract class BaseUsersEdgeEventFetcher extends BasePageableEdgeEventFet
return EdgeUtils.constructEdgeEvent(tenantId, edge.getId(), EdgeEventType.USER, return EdgeUtils.constructEdgeEvent(tenantId, edge.getId(), EdgeEventType.USER,
EdgeEventActionType.ADDED, user.getId(), null); EdgeEventActionType.ADDED, user.getId(), null);
} }
protected abstract PageData<User> findUsers(TenantId tenantId, PageLink pageLink);
} }

3
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 lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component; import org.springframework.stereotype.Component;
import org.thingsboard.common.util.JacksonUtil;
import org.thingsboard.server.common.data.AdminSettings; import org.thingsboard.server.common.data.AdminSettings;
import org.thingsboard.server.common.data.EdgeUtils; import org.thingsboard.server.common.data.EdgeUtils;
import org.thingsboard.server.common.data.edge.EdgeEvent; 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 class AdminSettingsEdgeProcessor extends BaseEdgeProcessor {
public DownlinkMsg processAdminSettingsToEdge(EdgeEvent edgeEvent) { 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); AdminSettingsUpdateMsg adminSettingsUpdateMsg = adminSettingsMsgConstructor.constructAdminSettingsUpdateMsg(adminSettings);
return DownlinkMsg.newBuilder() return DownlinkMsg.newBuilder()
.setDownlinkMsgId(EdgeUtils.nextPositiveInt()) .setDownlinkMsgId(EdgeUtils.nextPositiveInt())

12
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 com.google.common.util.concurrent.ListenableFuture;
import lombok.extern.slf4j.Slf4j; import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component; import org.springframework.stereotype.Component;
import org.thingsboard.common.util.JacksonUtil;
import org.thingsboard.server.common.data.EdgeUtils; import org.thingsboard.server.common.data.EdgeUtils;
import org.thingsboard.server.common.data.EntityType; import org.thingsboard.server.common.data.EntityType;
import org.thingsboard.server.common.data.alarm.Alarm; import org.thingsboard.server.common.data.alarm.Alarm;
@ -76,7 +77,7 @@ public class AlarmEdgeProcessor extends BaseEdgeProcessor {
existentAlarm.setStatus(AlarmStatus.valueOf(alarmUpdateMsg.getStatus())); existentAlarm.setStatus(AlarmStatus.valueOf(alarmUpdateMsg.getStatus()));
existentAlarm.setAckTs(alarmUpdateMsg.getAckTs()); existentAlarm.setAckTs(alarmUpdateMsg.getAckTs());
existentAlarm.setEndTs(alarmUpdateMsg.getEndTs()); existentAlarm.setEndTs(alarmUpdateMsg.getEndTs());
existentAlarm.setDetails(mapper.readTree(alarmUpdateMsg.getDetails())); existentAlarm.setDetails(JacksonUtil.OBJECT_MAPPER.readTree(alarmUpdateMsg.getDetails()));
alarmService.createOrUpdateAlarm(existentAlarm); alarmService.createOrUpdateAlarm(existentAlarm);
break; break;
case ALARM_ACK_RPC_MESSAGE: case ALARM_ACK_RPC_MESSAGE:
@ -86,7 +87,8 @@ public class AlarmEdgeProcessor extends BaseEdgeProcessor {
break; break;
case ALARM_CLEAR_RPC_MESSAGE: case ALARM_CLEAR_RPC_MESSAGE:
if (existentAlarm != null) { 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; break;
case ENTITY_DELETED_RPC_MESSAGE: case ENTITY_DELETED_RPC_MESSAGE:
@ -136,7 +138,7 @@ public class AlarmEdgeProcessor extends BaseEdgeProcessor {
} }
break; break;
case DELETED: case DELETED:
Alarm alarm = mapper.convertValue(edgeEvent.getBody(), Alarm.class); Alarm alarm = JacksonUtil.OBJECT_MAPPER.convertValue(edgeEvent.getBody(), Alarm.class);
AlarmUpdateMsg alarmUpdateMsg = AlarmUpdateMsg alarmUpdateMsg =
alarmMsgConstructor.constructAlarmUpdatedMsg(edge.getTenantId(), msgType, alarm); alarmMsgConstructor.constructAlarmUpdatedMsg(edge.getTenantId(), msgType, alarm);
downlinkMsg = DownlinkMsg.newBuilder() downlinkMsg = DownlinkMsg.newBuilder()
@ -154,8 +156,8 @@ public class AlarmEdgeProcessor extends BaseEdgeProcessor {
switch (actionType) { switch (actionType) {
case DELETED: case DELETED:
EdgeId edgeId = new EdgeId(new UUID(edgeNotificationMsg.getEdgeIdMSB(), edgeNotificationMsg.getEdgeIdLSB())); EdgeId edgeId = new EdgeId(new UUID(edgeNotificationMsg.getEdgeIdMSB(), edgeNotificationMsg.getEdgeIdLSB()));
Alarm deletedAlarm = mapper.readValue(edgeNotificationMsg.getBody(), Alarm.class); Alarm deletedAlarm = JacksonUtil.OBJECT_MAPPER.readValue(edgeNotificationMsg.getBody(), Alarm.class);
return saveEdgeEvent(tenantId, edgeId, EdgeEventType.ALARM, actionType, alarmId, mapper.valueToTree(deletedAlarm)); return saveEdgeEvent(tenantId, edgeId, EdgeEventType.ALARM, actionType, alarmId, JacksonUtil.OBJECT_MAPPER.valueToTree(deletedAlarm));
default: default:
ListenableFuture<Alarm> alarmFuture = alarmService.findAlarmByIdAsync(tenantId, alarmId); ListenableFuture<Alarm> alarmFuture = alarmService.findAlarmByIdAsync(tenantId, alarmId);
return Futures.transformAsync(alarmFuture, alarm -> { return Futures.transformAsync(alarmFuture, alarm -> {

7
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.springframework.stereotype.Component;
import org.thingsboard.server.common.data.EdgeUtils; import org.thingsboard.server.common.data.EdgeUtils;
import org.thingsboard.server.common.data.asset.Asset; 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.EdgeEvent;
import org.thingsboard.server.common.data.edge.EdgeEventActionType; import org.thingsboard.server.common.data.edge.EdgeEventActionType;
import org.thingsboard.server.common.data.id.AssetId; 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.AssetUpdateMsg;
import org.thingsboard.server.gen.edge.v1.DownlinkMsg; import org.thingsboard.server.gen.edge.v1.DownlinkMsg;
import org.thingsboard.server.gen.edge.v1.UpdateMsgType; import org.thingsboard.server.gen.edge.v1.UpdateMsgType;
@ -34,7 +32,7 @@ import org.thingsboard.server.queue.util.TbCoreComponent;
@TbCoreComponent @TbCoreComponent
public class AssetEdgeProcessor extends BaseEdgeProcessor { 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()); AssetId assetId = new AssetId(edgeEvent.getEntityId());
DownlinkMsg downlinkMsg = null; DownlinkMsg downlinkMsg = null;
switch (action) { switch (action) {
@ -45,9 +43,8 @@ public class AssetEdgeProcessor extends BaseEdgeProcessor {
case UNASSIGNED_FROM_CUSTOMER: case UNASSIGNED_FROM_CUSTOMER:
Asset asset = assetService.findAssetById(edgeEvent.getTenantId(), assetId); Asset asset = assetService.findAssetById(edgeEvent.getTenantId(), assetId);
if (asset != null) { if (asset != null) {
CustomerId customerId = getCustomerIdIfEdgeAssignedToCustomer(asset, edge);
AssetUpdateMsg assetUpdateMsg = AssetUpdateMsg assetUpdateMsg =
assetMsgConstructor.constructAssetUpdatedMsg(msgType, asset, customerId); assetMsgConstructor.constructAssetUpdatedMsg(msgType, asset);
downlinkMsg = DownlinkMsg.newBuilder() downlinkMsg = DownlinkMsg.newBuilder()
.setDownlinkMsgId(EdgeUtils.nextPositiveInt()) .setDownlinkMsgId(EdgeUtils.nextPositiveInt())
.addAssetUpdateMsg(assetUpdateMsg) .addAssetUpdateMsg(assetUpdateMsg)

15
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; package org.thingsboard.server.service.edge.rpc.processor;
import com.fasterxml.jackson.databind.JsonNode; 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.Futures;
import com.google.common.util.concurrent.ListenableFuture; import com.google.common.util.concurrent.ListenableFuture;
import lombok.extern.slf4j.Slf4j; 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.cluster.TbClusterService;
import org.thingsboard.server.common.data.Device; import org.thingsboard.server.common.data.Device;
import org.thingsboard.server.common.data.EdgeUtils; 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.Edge;
import org.thingsboard.server.common.data.edge.EdgeEvent; import org.thingsboard.server.common.data.edge.EdgeEvent;
import org.thingsboard.server.common.data.edge.EdgeEventActionType; import org.thingsboard.server.common.data.edge.EdgeEventActionType;
import org.thingsboard.server.common.data.edge.EdgeEventType; 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.EdgeId;
import org.thingsboard.server.common.data.id.EntityId; import org.thingsboard.server.common.data.id.EntityId;
import org.thingsboard.server.common.data.id.TenantId; import org.thingsboard.server.common.data.id.TenantId;
@ -84,9 +81,7 @@ import java.util.List;
@Slf4j @Slf4j
public abstract class BaseEdgeProcessor { public abstract class BaseEdgeProcessor {
protected static final ObjectMapper mapper = new ObjectMapper(); protected static final int DEFAULT_PAGE_SIZE = 100;
protected static final int DEFAULT_PAGE_SIZE = 1000;
@Autowired @Autowired
protected RuleChainService ruleChainService; protected RuleChainService ruleChainService;
@ -233,14 +228,6 @@ public abstract class BaseEdgeProcessor {
}, dbCallbackExecutorService); }, 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<Void> processActionForAllEdges(TenantId tenantId, EdgeEventType type, EdgeEventActionType actionType, EntityId entityId) { protected ListenableFuture<Void> processActionForAllEdges(TenantId tenantId, EdgeEventType type, EdgeEventActionType actionType, EntityId entityId) {
List<ListenableFuture<Void>> futures = new ArrayList<>(); List<ListenableFuture<Void>> futures = new ArrayList<>();
if (TenantId.SYS_TENANT_ID.equals(tenantId)) { if (TenantId.SYS_TENANT_ID.equals(tenantId)) {

8
application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/DashboardEdgeProcessor.java

@ -36,7 +36,7 @@ import java.util.Collections;
@TbCoreComponent @TbCoreComponent
public class DashboardEdgeProcessor extends BaseEdgeProcessor { 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()); DashboardId dashboardId = new DashboardId(edgeEvent.getEntityId());
DownlinkMsg downlinkMsg = null; DownlinkMsg downlinkMsg = null;
switch (action) { switch (action) {
@ -47,12 +47,8 @@ public class DashboardEdgeProcessor extends BaseEdgeProcessor {
case UNASSIGNED_FROM_CUSTOMER: case UNASSIGNED_FROM_CUSTOMER:
Dashboard dashboard = dashboardService.findDashboardById(edgeEvent.getTenantId(), dashboardId); Dashboard dashboard = dashboardService.findDashboardById(edgeEvent.getTenantId(), dashboardId);
if (dashboard != null) { if (dashboard != null) {
CustomerId customerId = null;
if (!edge.getCustomerId().isNullUid() && dashboard.isAssignedToCustomer(edge.getCustomerId())) {
customerId = edge.getCustomerId();
}
DashboardUpdateMsg dashboardUpdateMsg = DashboardUpdateMsg dashboardUpdateMsg =
dashboardMsgConstructor.constructDashboardUpdatedMsg(msgType, dashboard, customerId); dashboardMsgConstructor.constructDashboardUpdatedMsg(msgType, dashboard);
downlinkMsg = DownlinkMsg.newBuilder() downlinkMsg = DownlinkMsg.newBuilder()
.setDownlinkMsgId(EdgeUtils.nextPositiveInt()) .setDownlinkMsgId(EdgeUtils.nextPositiveInt())
.addDashboardUpdateMsg(dashboardUpdateMsg) .addDashboardUpdateMsg(dashboardUpdateMsg)

9
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); log.error(errMsg, e);
return Futures.immediateFuture(null); return Futures.immediateFuture(null);
} }
ObjectNode body = mapper.createObjectNode(); ObjectNode body = JacksonUtil.OBJECT_MAPPER.createObjectNode();
body.put("conflictName", deviceName); body.put("conflictName", deviceName);
ListenableFuture<Void> input = saveEdgeEvent(tenantId, edge.getId(), EdgeEventType.DEVICE, EdgeEventActionType.ENTITY_MERGE_REQUEST, newDevice.getId(), body); ListenableFuture<Void> input = saveEdgeEvent(tenantId, edge.getId(), EdgeEventType.DEVICE, EdgeEventActionType.ENTITY_MERGE_REQUEST, newDevice.getId(), body);
return Futures.transformAsync(input, unused -> return Futures.transformAsync(input, unused ->
@ -296,9 +296,9 @@ public class DeviceEdgeProcessor extends BaseEdgeProcessor {
private void pushDeviceCreatedEventToRuleEngine(TenantId tenantId, Edge edge, Device device) { private void pushDeviceCreatedEventToRuleEngine(TenantId tenantId, Edge edge, Device device) {
try { try {
DeviceId deviceId = device.getId(); 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(), 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() { tbClusterService.pushMsgToRuleEngine(tenantId, deviceId, tbMsg, new TbQueueCallback() {
@Override @Override
public void onSuccess(TbQueueMsgMetadata metadata) { public void onSuccess(TbQueueMsgMetadata metadata) {
@ -374,9 +374,8 @@ public class DeviceEdgeProcessor extends BaseEdgeProcessor {
case UNASSIGNED_FROM_CUSTOMER: case UNASSIGNED_FROM_CUSTOMER:
Device device = deviceService.findDeviceById(edgeEvent.getTenantId(), deviceId); Device device = deviceService.findDeviceById(edgeEvent.getTenantId(), deviceId);
if (device != null) { if (device != null) {
CustomerId customerId = getCustomerIdIfEdgeAssignedToCustomer(device, edge);
DeviceUpdateMsg deviceUpdateMsg = DeviceUpdateMsg deviceUpdateMsg =
deviceMsgConstructor.constructDeviceUpdatedMsg(msgType, device, customerId, null); deviceMsgConstructor.constructDeviceUpdatedMsg(msgType, device, null);
downlinkMsg = DownlinkMsg.newBuilder() downlinkMsg = DownlinkMsg.newBuilder()
.setDownlinkMsgId(EdgeUtils.nextPositiveInt()) .setDownlinkMsgId(EdgeUtils.nextPositiveInt())
.addDeviceUpdateMsg(deviceUpdateMsg) .addDeviceUpdateMsg(deviceUpdateMsg)

38
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 com.google.common.util.concurrent.ListenableFuture;
import lombok.extern.slf4j.Slf4j; import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component; import org.springframework.stereotype.Component;
import org.thingsboard.common.util.JacksonUtil;
import org.thingsboard.server.common.data.User; import org.thingsboard.server.common.data.User;
import org.thingsboard.server.common.data.edge.Edge; import org.thingsboard.server.common.data.edge.Edge;
import org.thingsboard.server.common.data.edge.EdgeEventActionType; import org.thingsboard.server.common.data.edge.EdgeEventActionType;
@ -43,43 +44,8 @@ public class EdgeProcessor extends BaseEdgeProcessor {
public ListenableFuture<Void> processEdgeNotification(TenantId tenantId, TransportProtos.EdgeNotificationMsgProto edgeNotificationMsg) { public ListenableFuture<Void> processEdgeNotification(TenantId tenantId, TransportProtos.EdgeNotificationMsgProto edgeNotificationMsg) {
try { try {
EdgeEventActionType actionType = EdgeEventActionType.valueOf(edgeNotificationMsg.getAction()); EdgeEventActionType actionType = EdgeEventActionType.valueOf(edgeNotificationMsg.getAction());
EdgeId edgeId = new EdgeId(new UUID(edgeNotificationMsg.getEntityIdMSB(), edgeNotificationMsg.getEntityIdLSB()));
ListenableFuture<Edge> edgeFuture;
switch (actionType) { switch (actionType) {
case ASSIGNED_TO_CUSTOMER: // TODO: add support for edge update
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<ListenableFuture<Void>> futures = new ArrayList<>();
futures.add(saveEdgeEvent(edge.getTenantId(), edge.getId(), EdgeEventType.CUSTOMER, EdgeEventActionType.ADDED, customerId, null));
PageLink pageLink = new PageLink(DEFAULT_PAGE_SIZE);
PageData<User> pageData;
do {
pageData = userService.findCustomerUsers(tenantId, customerId, pageLink);
if (pageData != null && pageData.getData() != null && !pageData.getData().isEmpty()) {
log.trace("[{}] [{}] user(s) are going to be added to edge.", edge.getId(), pageData.getData().size());
for (User user : pageData.getData()) {
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);
default: default:
return Futures.immediateFuture(null); return Futures.immediateFuture(null);
} }

43
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 com.google.common.util.concurrent.ListenableFuture;
import lombok.extern.slf4j.Slf4j; import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component; import org.springframework.stereotype.Component;
import org.thingsboard.common.util.JacksonUtil;
import org.thingsboard.server.common.data.Device; import org.thingsboard.server.common.data.Device;
import org.thingsboard.server.common.data.EdgeUtils; import org.thingsboard.server.common.data.EdgeUtils;
import org.thingsboard.server.common.data.edge.Edge; import org.thingsboard.server.common.data.edge.Edge;
@ -57,13 +58,12 @@ public class EntityEdgeProcessor extends BaseEdgeProcessor {
if (EdgeEventType.DEVICE.equals(edgeEvent.getType())) { if (EdgeEventType.DEVICE.equals(edgeEvent.getType())) {
DeviceId deviceId = new DeviceId(edgeEvent.getEntityId()); DeviceId deviceId = new DeviceId(edgeEvent.getEntityId());
Device device = deviceService.findDeviceById(edge.getTenantId(), deviceId); Device device = deviceService.findDeviceById(edge.getTenantId(), deviceId);
CustomerId customerId = getCustomerIdIfEdgeAssignedToCustomer(device, edge);
String conflictName = null; String conflictName = null;
if(edgeEvent.getBody() != null) { if(edgeEvent.getBody() != null) {
conflictName = edgeEvent.getBody().get("conflictName").asText(); conflictName = edgeEvent.getBody().get("conflictName").asText();
} }
DeviceUpdateMsg deviceUpdateMsg = deviceMsgConstructor DeviceUpdateMsg deviceUpdateMsg = deviceMsgConstructor
.constructDeviceUpdatedMsg(UpdateMsgType.ENTITY_MERGE_RPC_MESSAGE, device, customerId, conflictName); .constructDeviceUpdatedMsg(UpdateMsgType.ENTITY_MERGE_RPC_MESSAGE, device, conflictName);
downlinkMsg = DownlinkMsg.newBuilder() downlinkMsg = DownlinkMsg.newBuilder()
.setDownlinkMsgId(EdgeUtils.nextPositiveInt()) .setDownlinkMsgId(EdgeUtils.nextPositiveInt())
.addDeviceUpdateMsg(deviceUpdateMsg) .addDeviceUpdateMsg(deviceUpdateMsg)
@ -98,10 +98,9 @@ public class EntityEdgeProcessor extends BaseEdgeProcessor {
case ADDED: // used only for USER entity case ADDED: // used only for USER entity
case UPDATED: case UPDATED:
case CREDENTIALS_UPDATED: case CREDENTIALS_UPDATED:
return pushNotificationToAllRelatedEdges(tenantId, entityId, type, actionType);
case ASSIGNED_TO_CUSTOMER: case ASSIGNED_TO_CUSTOMER:
case UNASSIGNED_FROM_CUSTOMER: case UNASSIGNED_FROM_CUSTOMER:
return pushNotificationToAllRelatedCustomerEdges(tenantId, edgeNotificationMsg, entityId, actionType, type); return pushNotificationToAllRelatedEdges(tenantId, entityId, type, actionType);
case DELETED: case DELETED:
if (edgeId != null) { if (edgeId != null) {
return saveEdgeEvent(tenantId, edgeId, type, actionType, entityId, null); return saveEdgeEvent(tenantId, edgeId, type, actionType, entityId, null);
@ -123,42 +122,6 @@ public class EntityEdgeProcessor extends BaseEdgeProcessor {
} }
} }
private ListenableFuture<Void> pushNotificationToAllRelatedCustomerEdges(TenantId tenantId,
TransportProtos.EdgeNotificationMsgProto edgeNotificationMsg,
EntityId entityId,
EdgeEventActionType actionType,
EdgeEventType type) {
PageLink pageLink = new PageLink(DEFAULT_PAGE_SIZE);
PageData<EdgeId> pageData;
List<ListenableFuture<Void>> 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<Edge> 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) { private EdgeId safeGetEdgeId(TransportProtos.EdgeNotificationMsgProto edgeNotificationMsg) {
if (edgeNotificationMsg.getEdgeIdMSB() != 0 && edgeNotificationMsg.getEdgeIdLSB() != 0) { if (edgeNotificationMsg.getEdgeIdMSB() != 0 && edgeNotificationMsg.getEdgeIdLSB() != 0) {
return new EdgeId(new UUID(edgeNotificationMsg.getEdgeIdMSB(), edgeNotificationMsg.getEdgeIdLSB())); return new EdgeId(new UUID(edgeNotificationMsg.getEdgeIdMSB(), edgeNotificationMsg.getEdgeIdLSB()));

7
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.springframework.stereotype.Component;
import org.thingsboard.server.common.data.EdgeUtils; import org.thingsboard.server.common.data.EdgeUtils;
import org.thingsboard.server.common.data.EntityView; 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.EdgeEvent;
import org.thingsboard.server.common.data.edge.EdgeEventActionType; 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.common.data.id.EntityViewId;
import org.thingsboard.server.gen.edge.v1.DownlinkMsg; import org.thingsboard.server.gen.edge.v1.DownlinkMsg;
import org.thingsboard.server.gen.edge.v1.EntityViewUpdateMsg; import org.thingsboard.server.gen.edge.v1.EntityViewUpdateMsg;
@ -34,7 +32,7 @@ import org.thingsboard.server.queue.util.TbCoreComponent;
@TbCoreComponent @TbCoreComponent
public class EntityViewEdgeProcessor extends BaseEdgeProcessor { 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()); EntityViewId entityViewId = new EntityViewId(edgeEvent.getEntityId());
DownlinkMsg downlinkMsg = null; DownlinkMsg downlinkMsg = null;
switch (action) { switch (action) {
@ -45,9 +43,8 @@ public class EntityViewEdgeProcessor extends BaseEdgeProcessor {
case UNASSIGNED_FROM_CUSTOMER: case UNASSIGNED_FROM_CUSTOMER:
EntityView entityView = entityViewService.findEntityViewById(edgeEvent.getTenantId(), entityViewId); EntityView entityView = entityViewService.findEntityViewById(edgeEvent.getTenantId(), entityViewId);
if (entityView != null) { if (entityView != null) {
CustomerId customerId = getCustomerIdIfEdgeAssignedToCustomer(entityView, edge);
EntityViewUpdateMsg entityViewUpdateMsg = EntityViewUpdateMsg entityViewUpdateMsg =
entityViewMsgConstructor.constructEntityViewUpdatedMsg(msgType, entityView, customerId); entityViewMsgConstructor.constructEntityViewUpdatedMsg(msgType, entityView);
downlinkMsg = DownlinkMsg.newBuilder() downlinkMsg = DownlinkMsg.newBuilder()
.setDownlinkMsgId(EdgeUtils.nextPositiveInt()) .setDownlinkMsgId(EdgeUtils.nextPositiveInt())
.addEntityViewUpdateMsg(entityViewUpdateMsg) .addEntityViewUpdateMsg(entityViewUpdateMsg)

9
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 com.google.common.util.concurrent.ListenableFuture;
import lombok.extern.slf4j.Slf4j; import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component; import org.springframework.stereotype.Component;
import org.thingsboard.common.util.JacksonUtil;
import org.thingsboard.server.common.data.EdgeUtils; import org.thingsboard.server.common.data.EdgeUtils;
import org.thingsboard.server.common.data.EntityType; import org.thingsboard.server.common.data.EntityType;
import org.thingsboard.server.common.data.edge.EdgeEvent; import org.thingsboard.server.common.data.edge.EdgeEvent;
@ -75,7 +76,7 @@ public class RelationEdgeProcessor extends BaseEdgeProcessor {
if (relationUpdateMsg.hasTypeGroup()) { if (relationUpdateMsg.hasTypeGroup()) {
entityRelation.setTypeGroup(RelationTypeGroup.valueOf(relationUpdateMsg.getTypeGroup())); entityRelation.setTypeGroup(RelationTypeGroup.valueOf(relationUpdateMsg.getTypeGroup()));
} }
entityRelation.setAdditionalInfo(mapper.readTree(relationUpdateMsg.getAdditionalInfo())); entityRelation.setAdditionalInfo(JacksonUtil.OBJECT_MAPPER.readTree(relationUpdateMsg.getAdditionalInfo()));
switch (relationUpdateMsg.getMsgType()) { switch (relationUpdateMsg.getMsgType()) {
case ENTITY_CREATED_RPC_MESSAGE: case ENTITY_CREATED_RPC_MESSAGE:
case ENTITY_UPDATED_RPC_MESSAGE: case ENTITY_UPDATED_RPC_MESSAGE:
@ -118,7 +119,7 @@ public class RelationEdgeProcessor extends BaseEdgeProcessor {
} }
public DownlinkMsg processRelationToEdge(EdgeEvent edgeEvent, UpdateMsgType msgType) { 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); RelationUpdateMsg relationUpdateMsg = relationMsgConstructor.constructRelationUpdatedMsg(msgType, entityRelation);
return DownlinkMsg.newBuilder() return DownlinkMsg.newBuilder()
.setDownlinkMsgId(EdgeUtils.nextPositiveInt()) .setDownlinkMsgId(EdgeUtils.nextPositiveInt())
@ -127,7 +128,7 @@ public class RelationEdgeProcessor extends BaseEdgeProcessor {
} }
public ListenableFuture<Void> processRelationNotification(TenantId tenantId, TransportProtos.EdgeNotificationMsgProto edgeNotificationMsg) throws JsonProcessingException { public ListenableFuture<Void> 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) || if (relation.getFrom().getEntityType().equals(EntityType.EDGE) ||
relation.getTo().getEntityType().equals(EntityType.EDGE)) { relation.getTo().getEntityType().equals(EntityType.EDGE)) {
return Futures.immediateFuture(null); return Futures.immediateFuture(null);
@ -146,7 +147,7 @@ public class RelationEdgeProcessor extends BaseEdgeProcessor {
EdgeEventType.RELATION, EdgeEventType.RELATION,
EdgeEventActionType.valueOf(edgeNotificationMsg.getAction()), EdgeEventActionType.valueOf(edgeNotificationMsg.getAction()),
null, null,
mapper.valueToTree(relation))); JacksonUtil.OBJECT_MAPPER.valueToTree(relation)));
} }
return Futures.transform(Futures.allAsList(futures), voids -> null, dbCallbackExecutorService); return Futures.transform(Futures.allAsList(futures), voids -> null, dbCallbackExecutorService);
} }

5
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.ImmutablePair;
import org.apache.commons.lang3.tuple.Pair; import org.apache.commons.lang3.tuple.Pair;
import org.springframework.stereotype.Component; import org.springframework.stereotype.Component;
import org.thingsboard.common.util.JacksonUtil;
import org.thingsboard.rule.engine.api.msg.DeviceAttributesEventNotificationMsg; import org.thingsboard.rule.engine.api.msg.DeviceAttributesEventNotificationMsg;
import org.thingsboard.server.common.data.DataConstants; import org.thingsboard.server.common.data.DataConstants;
import org.thingsboard.server.common.data.Device; 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.EdgeId;
import org.thingsboard.server.common.data.id.EntityId; import org.thingsboard.server.common.data.id.EntityId;
import org.thingsboard.server.common.data.id.EntityViewId; 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.RuleChainId;
import org.thingsboard.server.common.data.id.TenantId; import org.thingsboard.server.common.data.id.TenantId;
import org.thingsboard.server.common.data.id.UserId; import org.thingsboard.server.common.data.id.UserId;
@ -332,7 +332,8 @@ public class TelemetryEdgeProcessor extends BaseEdgeProcessor {
log.warn("Unsupported edge event type [{}]", edgeEvent); log.warn("Unsupported edge event type [{}]", edgeEvent);
return null; 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) { private DownlinkMsg constructEntityDataProtoMsg(EntityId entityId, EdgeEventActionType actionType, JsonElement entityData) {

3
application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/UserEdgeProcessor.java

@ -43,10 +43,9 @@ public class UserEdgeProcessor extends BaseEdgeProcessor {
case UPDATED: case UPDATED:
User user = userService.findUserById(edgeEvent.getTenantId(), userId); User user = userService.findUserById(edgeEvent.getTenantId(), userId);
if (user != null) { if (user != null) {
CustomerId customerId = getCustomerIdIfEdgeAssignedToCustomer(user, edge);
downlinkMsg = DownlinkMsg.newBuilder() downlinkMsg = DownlinkMsg.newBuilder()
.setDownlinkMsgId(EdgeUtils.nextPositiveInt()) .setDownlinkMsgId(EdgeUtils.nextPositiveInt())
.addUserUpdateMsg(userMsgConstructor.constructUserUpdatedMsg(msgType, user, customerId)) .addUserUpdateMsg(userMsgConstructor.constructUserUpdatedMsg(msgType, user))
.build(); .build();
} }
break; break;

10
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; package org.thingsboard.server.service.edge.rpc.sync;
import com.fasterxml.jackson.databind.JsonNode; import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.node.ObjectNode; import com.fasterxml.jackson.databind.node.ObjectNode;
import com.google.common.util.concurrent.FutureCallback; import com.google.common.util.concurrent.FutureCallback;
import com.google.common.util.concurrent.Futures; 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.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Lazy; import org.springframework.context.annotation.Lazy;
import org.springframework.stereotype.Service; import org.springframework.stereotype.Service;
import org.thingsboard.common.util.JacksonUtil;
import org.thingsboard.server.cluster.TbClusterService; import org.thingsboard.server.cluster.TbClusterService;
import org.thingsboard.server.common.data.Device; import org.thingsboard.server.common.data.Device;
import org.thingsboard.server.common.data.DeviceProfile; import org.thingsboard.server.common.data.DeviceProfile;
@ -88,8 +88,6 @@ import java.util.UUID;
@Slf4j @Slf4j
public class DefaultEdgeRequestsService implements EdgeRequestsService { public class DefaultEdgeRequestsService implements EdgeRequestsService {
private static final ObjectMapper mapper = new ObjectMapper();
private static final int DEFAULT_PAGE_SIZE = 1000; private static final int DEFAULT_PAGE_SIZE = 1000;
@Autowired @Autowired
@ -163,7 +161,7 @@ public class DefaultEdgeRequestsService implements EdgeRequestsService {
try { try {
Map<String, Object> entityData = new HashMap<>(); Map<String, Object> entityData = new HashMap<>();
ObjectNode attributes = mapper.createObjectNode(); ObjectNode attributes = JacksonUtil.OBJECT_MAPPER.createObjectNode();
for (AttributeKvEntry attr : ssAttributes) { for (AttributeKvEntry attr : ssAttributes) {
if (DefaultDeviceStateService.PERSISTENT_ATTRIBUTES.contains(attr.getKey())) { if (DefaultDeviceStateService.PERSISTENT_ATTRIBUTES.contains(attr.getKey())) {
continue; continue;
@ -180,7 +178,7 @@ public class DefaultEdgeRequestsService implements EdgeRequestsService {
} }
entityData.put("kv", attributes); entityData.put("kv", attributes);
entityData.put("scope", scope); 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); log.debug("Sending attributes data msg, entityId [{}], attributes [{}]", entityId, body);
ListenableFuture<Void> future = saveEdgeEvent(tenantId, edge.getId(), type, EdgeEventActionType.ATTRIBUTES_UPDATED, entityId, body); ListenableFuture<Void> future = saveEdgeEvent(tenantId, edge.getId(), type, EdgeEventActionType.ATTRIBUTES_UPDATED, entityId, body);
Futures.addCallback(future, new FutureCallback<>() { Futures.addCallback(future, new FutureCallback<>() {
@ -242,7 +240,7 @@ public class DefaultEdgeRequestsService implements EdgeRequestsService {
EdgeEventType.RELATION, EdgeEventType.RELATION,
EdgeEventActionType.ADDED, EdgeEventActionType.ADDED,
null, null,
mapper.valueToTree(relation))); JacksonUtil.OBJECT_MAPPER.valueToTree(relation)));
} }
} catch (Exception e) { } catch (Exception e) {
String errMsg = String.format("[%s] Exception during loading relation [%s] to edge on sync!", edge.getId(), relation); String errMsg = String.format("[%s] Exception during loading relation [%s] to edge on sync!", edge.getId(), relation);

27
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); logEntityAction(tenantId, entityId, entity, customerId, actionType, user, additionalInfo);
if (sendToEdge) { if (sendToEdge) {
sendEntityAssignToCustomerNotificationMsg(tenantId, entityId, customerId, edgeTypeByActionType(actionType)); sendEntityNotificationMsg(tenantId, entityId, edgeTypeByActionType(actionType));
} }
} }
@ -203,24 +203,17 @@ public class DefaultTbNotificationEntityService implements TbNotificationEntityS
} }
@Override @Override
public void notifyEdge(TenantId tenantId, EdgeId edgeId, CustomerId customerId, Edge edge, public void notifyCreateOrUpdateOrDeleteEdge(TenantId tenantId, EdgeId edgeId, CustomerId customerId, Edge edge,
ActionType actionType, User user, Object... additionalInfo) { ActionType actionType, User user, Object... additionalInfo) {
ComponentLifecycleEvent lifecycleEvent; ComponentLifecycleEvent lifecycleEvent;
EdgeEventActionType edgeEventActionType = null;
switch (actionType) { switch (actionType) {
case ADDED: case ADDED:
lifecycleEvent = ComponentLifecycleEvent.CREATED; lifecycleEvent = ComponentLifecycleEvent.CREATED;
break; break;
case UPDATED: case UPDATED:
lifecycleEvent = ComponentLifecycleEvent.UPDATED;
break;
case ASSIGNED_TO_CUSTOMER: case ASSIGNED_TO_CUSTOMER:
lifecycleEvent = ComponentLifecycleEvent.UPDATED;
edgeEventActionType = EdgeEventActionType.ASSIGNED_TO_CUSTOMER;
break;
case UNASSIGNED_FROM_CUSTOMER: case UNASSIGNED_FROM_CUSTOMER:
lifecycleEvent = ComponentLifecycleEvent.UPDATED; lifecycleEvent = ComponentLifecycleEvent.UPDATED;
edgeEventActionType = EdgeEventActionType.UNASSIGNED_FROM_CUSTOMER;
break; break;
case DELETED: case DELETED:
lifecycleEvent = ComponentLifecycleEvent.DELETED; lifecycleEvent = ComponentLifecycleEvent.DELETED;
@ -228,14 +221,8 @@ public class DefaultTbNotificationEntityService implements TbNotificationEntityS
default: default:
throw new IllegalArgumentException("Unknown actionType: " + actionType); throw new IllegalArgumentException("Unknown actionType: " + actionType);
} }
tbClusterService.broadcastEntityStateChangeEvent(tenantId, edgeId, lifecycleEvent); tbClusterService.broadcastEntityStateChangeEvent(tenantId, edgeId, lifecycleEvent);
logEntityAction(tenantId, edgeId, edge, customerId, actionType, user, additionalInfo); logEntityAction(tenantId, edgeId, edge, customerId, actionType, user, additionalInfo);
//Send notification to edge
if (edgeEventActionType != null) {
sendEntityAssignToCustomerNotificationMsg(tenantId, edgeId, customerId, edgeEventActionType);
}
} }
@Override @Override
@ -274,14 +261,6 @@ public class DefaultTbNotificationEntityService implements TbNotificationEntityS
sendNotificationMsgToEdge(tenantId, null, entityId, null, null, action); 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<EdgeId> edgeIds, String body) { private void sendAlarmDeleteNotificationMsg(TenantId tenantId, Alarm alarm, List<EdgeId> edgeIds, String body) {
try { try {
sendDeleteNotificationMsg(tenantId, alarm.getId(), edgeIds, body); sendDeleteNotificationMsg(tenantId, alarm.getId(), edgeIds, body);

4
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, void notifyAssignDeviceToTenant(TenantId tenantId, TenantId newTenantId, DeviceId deviceId, CustomerId customerId,
Device device, Tenant tenant, User user, Object... additionalInfo); Device device, Tenant tenant, User user, Object... additionalInfo);
void notifyEdge(TenantId tenantId, EdgeId edgeId, CustomerId customerId, Edge edge, ActionType actionType, void notifyCreateOrUpdateOrDeleteEdge(TenantId tenantId, EdgeId edgeId, CustomerId customerId, Edge edge, ActionType actionType,
User user, Object... additionalInfo); User user, Object... additionalInfo);
void notifyCreateOrUpdateAlarm(Alarm alarm, ActionType actionType, User user, Object... additionalInfo); void notifyCreateOrUpdateAlarm(Alarm alarm, ActionType actionType, User user, Object... additionalInfo);

12
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); edgeService.assignDefaultRuleChainsToEdge(tenantId, edgeId);
} }
notificationEntityService.notifyEdge(tenantId, edgeId, savedEdge.getCustomerId(), savedEdge, actionType, user); notificationEntityService.notifyCreateOrUpdateOrDeleteEdge(tenantId, edgeId, savedEdge.getCustomerId(), savedEdge, actionType, user);
return savedEdge; return savedEdge;
} catch (Exception e) { } catch (Exception e) {
@ -72,7 +72,7 @@ public class DefaultTbEdgeService extends AbstractTbEntityService implements TbE
TenantId tenantId = edge.getTenantId(); TenantId tenantId = edge.getTenantId();
try { try {
edgeService.deleteEdge(tenantId, edgeId); 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) { } catch (Exception e) {
notificationEntityService.logEntityAction(tenantId, emptyId(EntityType.EDGE), ActionType.DELETED, notificationEntityService.logEntityAction(tenantId, emptyId(EntityType.EDGE), ActionType.DELETED,
user, e, edgeId.toString()); user, e, edgeId.toString());
@ -85,7 +85,7 @@ public class DefaultTbEdgeService extends AbstractTbEntityService implements TbE
CustomerId customerId = customer.getId(); CustomerId customerId = customer.getId();
try { try {
Edge savedEdge = checkNotNull(edgeService.assignEdgeToCustomer(tenantId, edgeId, customerId)); 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()); edgeId.toString(), customerId.toString(), customer.getName());
return savedEdge; return savedEdge;
@ -104,7 +104,7 @@ public class DefaultTbEdgeService extends AbstractTbEntityService implements TbE
try { try {
Edge savedEdge = checkNotNull(edgeService.unassignEdgeFromCustomer(tenantId, edgeId)); 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()); edgeId.toString(), customerId.toString(), customer.getName());
return savedEdge; return savedEdge;
} catch (Exception e) { } catch (Exception e) {
@ -121,7 +121,7 @@ public class DefaultTbEdgeService extends AbstractTbEntityService implements TbE
try { try {
Edge savedEdge = checkNotNull(edgeService.assignEdgeToCustomer(tenantId, edgeId, customerId)); 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()); edgeId.toString(), customerId.toString(), publicCustomer.getName());
return savedEdge; return savedEdge;
@ -138,7 +138,7 @@ public class DefaultTbEdgeService extends AbstractTbEntityService implements TbE
EdgeId edgeId = edge.getId(); EdgeId edgeId = edge.getId();
try { try {
Edge updatedEdge = edgeNotificationService.setEdgeRootRuleChain(tenantId, edge, ruleChainId); 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; return updatedEdge;
} catch (Exception e) { } catch (Exception e) {
notificationEntityService.logEntityAction(tenantId, emptyId(EntityType.EDGE), notificationEntityService.logEntityAction(tenantId, emptyId(EntityType.EDGE),

1
common/edge-api/src/main/proto/edge.proto

@ -183,6 +183,7 @@ message DashboardUpdateMsg {
optional int64 customerIdLSB = 5; optional int64 customerIdLSB = 5;
string title = 6; string title = 6;
string configuration = 7; string configuration = 7;
optional string assignedCustomers = 8;
} }
message DeviceUpdateMsg { message DeviceUpdateMsg {

Loading…
Cancel
Save