tsai
/
thingsboard
mirror of https://github.com/thingsboard/thingsboard.git
1
0
Fork 0
Code Issues Projects Releases Wiki Activity
Browse Source

Merge pull request #14489 from MazurenkoNick/avoid-duplicate-edge-updates 

Avoid duplicate edge updates
pull/14702/head
Viacheslav Klimov 9 months ago
committed by GitHub
parent
9a846b5dab 1433d54250
commit
65d52bef93
No known key found for this signature in database GPG Key ID: B5690EEEBB952194
6 changed files with 110 additions and 73 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Split View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 28
      application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/BaseEdgeProcessor.java
  2. 33
      application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/asset/BaseAssetProcessor.java
  3. 15
      application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/dashboard/BaseDashboardProcessor.java
  4. 34
      application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/device/BaseDeviceProcessor.java
  5. 33
      application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/entityview/BaseEntityViewProcessor.java
  6. 40
      application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/user/BaseUserProcessor.java

28
application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/BaseEdgeProcessor.java
View File

@ -19,7 +19,7 @@ import com.fasterxml.jackson.databind.JsonNode;
import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.ListenableFuture;
import lombok.extern.slf4j.Slf4j;
import org.jetbrains.annotations.Nullable;
import org.jetbrains.annotations.NotNull;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Lazy;
import org.thingsboard.common.util.JacksonUtil;
@ -27,6 +27,9 @@ import org.thingsboard.server.common.data.AttributeScope;
import org.thingsboard.server.common.data.EdgeUtils;
import org.thingsboard.server.common.data.EntityType;
import org.thingsboard.server.common.data.HasCustomerId;
import org.thingsboard.server.common.data.HasName;
import org.thingsboard.server.common.data.HasVersion;
import org.thingsboard.server.common.data.StringUtils;
import org.thingsboard.server.common.data.edge.Edge;
import org.thingsboard.server.common.data.edge.EdgeEvent;
import org.thingsboard.server.common.data.edge.EdgeEventActionType;
@ -64,6 +67,7 @@ import org.thingsboard.server.service.edge.EdgeContextComponent;
import org.thingsboard.server.service.executors.DbCallbackExecutorService;
import org.thingsboard.server.service.state.DefaultDeviceStateService;
import javax.annotation.Nullable;
import java.util.ArrayList;
import java.util.List;
import java.util.Optional;
@ -407,4 +411,26 @@ public abstract class BaseEdgeProcessor implements EdgeProcessor {
});
}
protected boolean isSaveRequired(HasVersion current, HasVersion updated) {
updated.setVersion(null);
return !updated.equals(current);
}
protected <I extends EntityId, E extends HasName & HasId<I>> Optional<String> generateUniqueNameIfDuplicateExists(
TenantId tenantId, I entityId, E entity, @Nullable E entityWithSameName) {
if (entityWithSameName == null || entityWithSameName.getId().equals(entityId)) {
return Optional.empty();
}
String currentName = entity.getName();
String newEntityName = generateRandomAlphabeticString(currentName);
log.warn("[{}] Entity with name '{}' already exists (id={}). Renaming to '{}'", tenantId, currentName, entityWithSameName.getId(), newEntityName);
return Optional.of(newEntityName);
}
protected static String generateRandomAlphabeticString(String prefix) {
return prefix + "_" + StringUtils.randomAlphabetic(15);
}
}

33
application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/asset/BaseAssetProcessor.java
View File

@ -19,7 +19,6 @@ import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.util.Pair;
import org.thingsboard.common.util.JacksonUtil;
import org.thingsboard.server.common.data.StringUtils;
import org.thingsboard.server.common.data.asset.Asset;
import org.thingsboard.server.common.data.edge.Edge;
import org.thingsboard.server.common.data.id.AssetId;
@ -52,22 +51,15 @@ public abstract class BaseAssetProcessor extends BaseEdgeProcessor {
} else {
asset.setId(assetId);
}
String assetName = asset.getName();
Asset assetByName = edgeCtx.getAssetService().findAssetByTenantIdAndName(tenantId, assetName);
if (assetByName != null && !assetByName.getId().equals(assetId)) {
assetName = assetName + "_" + StringUtils.randomAlphanumeric(15);
log.warn("[{}] Asset with name {} already exists. Renaming asset name to {}",
tenantId, asset.getName(), assetName);
assetNameUpdated = true;
if (isSaveRequired(assetById, asset)) {
assetNameUpdated = updateAssetNameIfDuplicateExists(tenantId, assetId, asset);
setCustomerId(tenantId, created ? null : assetById.getCustomerId(), asset, assetUpdateMsg);
assetValidator.validate(asset, Asset::getTenantId);
if (created) {
asset.setId(assetId);
}
edgeCtx.getAssetService().saveAsset(asset, false);
}
asset.setName(assetName);
setCustomerId(tenantId, created ? null : assetById.getCustomerId(), asset, assetUpdateMsg);
assetValidator.validate(asset, Asset::getTenantId);
if (created) {
asset.setId(assetId);
}
edgeCtx.getAssetService().saveAsset(asset, false);
} catch (Exception e) {
log.error("[{}] Failed to process asset update msg [{}]", tenantId, assetUpdateMsg, e);
throw e;
@ -77,6 +69,15 @@ public abstract class BaseAssetProcessor extends BaseEdgeProcessor {
return Pair.of(created, assetNameUpdated);
}
private boolean updateAssetNameIfDuplicateExists(TenantId tenantId, AssetId assetId, Asset asset) {
Asset assetByName = edgeCtx.getAssetService().findAssetByTenantIdAndName(tenantId, asset.getName());
return generateUniqueNameIfDuplicateExists(tenantId, assetId, asset, assetByName).map(uniqueName -> {
asset.setName(uniqueName);
return true;
}).orElse(false);
}
protected abstract void setCustomerId(TenantId tenantId, CustomerId customerId, Asset asset, AssetUpdateMsg assetUpdateMsg);
protected void deleteAsset(TenantId tenantId, AssetId assetId) {

15
application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/dashboard/BaseDashboardProcessor.java
View File

@ -58,15 +58,14 @@ public abstract class BaseDashboardProcessor extends BaseEdgeProcessor {
dashboard.setAssignedCustomers(dashboardById.getAssignedCustomers());
}
dashboardValidator.validate(dashboard, Dashboard::getTenantId);
if (created) {
dashboard.setId(dashboardId);
if (isSaveRequired(dashboardById, dashboard)) {
dashboardValidator.validate(dashboard, Dashboard::getTenantId);
if (created) {
dashboard.setId(dashboardId);
}
dashboard = edgeCtx.getDashboardService().saveDashboard(dashboard, false);
}
Dashboard savedDashboard = edgeCtx.getDashboardService().saveDashboard(dashboard, false);
updateDashboardAssignments(tenantId, customerId, dashboardById, savedDashboard, newAssignedCustomers);
updateDashboardAssignments(tenantId, customerId, dashboardById, dashboard, newAssignedCustomers);
return created;
}

34
application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/device/BaseDeviceProcessor.java
View File

@ -20,7 +20,6 @@ import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.util.Pair;
import org.thingsboard.common.util.JacksonUtil;
import org.thingsboard.server.common.data.Device;
import org.thingsboard.server.common.data.StringUtils;
import org.thingsboard.server.common.data.edge.Edge;
import org.thingsboard.server.common.data.id.CustomerId;
import org.thingsboard.server.common.data.id.DeviceId;
@ -54,23 +53,17 @@ public abstract class BaseDeviceProcessor extends BaseEdgeProcessor {
} else {
device.setId(deviceId);
}
String deviceName = device.getName();
Device deviceByName = edgeCtx.getDeviceService().findDeviceByTenantIdAndName(tenantId, deviceName);
if (deviceByName != null && !deviceByName.getId().equals(deviceId)) {
deviceName = deviceName + "_" + StringUtils.randomAlphabetic(15);
log.warn("[{}] Device with name {} already exists. Renaming device name to {}",
tenantId, device.getName(), deviceName);
deviceNameUpdated = true;
}
device.setName(deviceName);
setCustomerId(tenantId, created ? null : deviceById.getCustomerId(), device, deviceUpdateMsg);
if (isSaveRequired(deviceById, device)) {
deviceNameUpdated = updateDeviceNameIfDuplicateExists(tenantId, deviceId, device);
setCustomerId(tenantId, created ? null : deviceById.getCustomerId(), device, deviceUpdateMsg);
deviceValidator.validate(device, Device::getTenantId);
if (created) {
device.setId(deviceId);
deviceValidator.validate(device, Device::getTenantId);
if (created) {
device.setId(deviceId);
}
Device savedDevice = edgeCtx.getDeviceService().saveDevice(device, false);
edgeCtx.getClusterService().onDeviceUpdated(savedDevice, created ? null : device);
}
Device savedDevice = edgeCtx.getDeviceService().saveDevice(device, false);
edgeCtx.getClusterService().onDeviceUpdated(savedDevice, created ? null : device);
} catch (Exception e) {
log.error("[{}] Failed to process device update msg [{}]", tenantId, deviceUpdateMsg, e);
throw e;
@ -80,6 +73,15 @@ public abstract class BaseDeviceProcessor extends BaseEdgeProcessor {
return Pair.of(created, deviceNameUpdated);
}
private boolean updateDeviceNameIfDuplicateExists(TenantId tenantId, DeviceId deviceId, Device device) {
Device deviceByName = edgeCtx.getDeviceService().findDeviceByTenantIdAndName(tenantId, device.getName());
return generateUniqueNameIfDuplicateExists(tenantId, deviceId, device, deviceByName).map(uniqueName -> {
device.setName(uniqueName);
return true;
}).orElse(false);
}
protected void updateDeviceCredentials(TenantId tenantId, DeviceCredentialsUpdateMsg deviceCredentialsUpdateMsg) {
DeviceCredentials deviceCredentials = JacksonUtil.fromString(deviceCredentialsUpdateMsg.getEntity(), DeviceCredentials.class, true);
if (deviceCredentials == null) {

33
application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/entityview/BaseEntityViewProcessor.java
View File

@ -21,7 +21,9 @@ import org.springframework.data.util.Pair;
import org.thingsboard.common.util.JacksonUtil;
import org.thingsboard.server.common.data.EntityView;
import org.thingsboard.server.common.data.StringUtils;
import org.thingsboard.server.common.data.asset.Asset;
import org.thingsboard.server.common.data.edge.Edge;
import org.thingsboard.server.common.data.id.AssetId;
import org.thingsboard.server.common.data.id.CustomerId;
import org.thingsboard.server.common.data.id.EntityViewId;
import org.thingsboard.server.common.data.id.TenantId;
@ -50,25 +52,28 @@ public abstract class BaseEntityViewProcessor extends BaseEdgeProcessor {
} else {
entityView.setId(entityViewId);
}
String entityViewName = entityView.getName();
EntityView entityViewByName = edgeCtx.getEntityViewService().findEntityViewByTenantIdAndName(tenantId, entityViewName);
if (entityViewByName != null && !entityViewByName.getId().equals(entityViewId)) {
entityViewName = entityViewName + "_" + StringUtils.randomAlphanumeric(15);
log.warn("[{}] Entity view with name {} already exists. Renaming entity view name to {}",
tenantId, entityView.getName(), entityViewName);
entityViewNameUpdated = true;
}
entityView.setName(entityViewName);
setCustomerId(tenantId, created ? null : entityViewById.getCustomerId(), entityView, entityViewUpdateMsg);
if (isSaveRequired(entityViewById, entityView)) {
entityViewNameUpdated = updateEntityViewNameIfDuplicateExists(tenantId, entityViewId, entityView);
setCustomerId(tenantId, created ? null : entityViewById.getCustomerId(), entityView, entityViewUpdateMsg);
entityViewValidator.validate(entityView, EntityView::getTenantId);
if (created) {
entityView.setId(entityViewId);
entityViewValidator.validate(entityView, EntityView::getTenantId);
if (created) {
entityView.setId(entityViewId);
}
edgeCtx.getEntityViewService().saveEntityView(entityView, false);
}
edgeCtx.getEntityViewService().saveEntityView(entityView, false);
return Pair.of(created, entityViewNameUpdated);
}
private boolean updateEntityViewNameIfDuplicateExists(TenantId tenantId, EntityViewId entityViewId, EntityView entityView) {
EntityView entityViewByName = edgeCtx.getEntityViewService().findEntityViewByTenantIdAndName(tenantId, entityView.getName());
return generateUniqueNameIfDuplicateExists(tenantId, entityViewId, entityView, entityViewByName).map(uniqueName -> {
entityView.setName(uniqueName);
return true;
}).orElse(false);
}
protected abstract void setCustomerId(TenantId tenantId, CustomerId customerId, EntityView entityView, EntityViewUpdateMsg entityViewUpdateMsg);
protected void deleteEntityView(TenantId tenantId, EntityViewId entityViewId) {

40
application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/user/BaseUserProcessor.java
View File

@ -22,7 +22,6 @@ import org.thingsboard.common.util.JacksonUtil;
import org.thingsboard.server.common.data.StringUtils;
import org.thingsboard.server.common.data.User;
import org.thingsboard.server.common.data.edge.Edge;
import org.thingsboard.server.common.data.exception.ThingsboardException;
import org.thingsboard.server.common.data.id.CustomerId;
import org.thingsboard.server.common.data.id.TenantId;
import org.thingsboard.server.common.data.id.UserId;
@ -57,27 +56,18 @@ public abstract class BaseUserProcessor extends BaseEdgeProcessor {
} else {
user.setId(userId);
}
if (isSaveRequired(userById, user)) {
userEmailUpdated = updateUserEmailIfDuplicateExists(tenantId, userId, user);
setCustomerId(tenantId, isCreated ? null : userById.getCustomerId(), user, userUpdateMsg);
String userEmail = user.getEmail();
User existing = edgeCtx.getUserService().findUserByTenantIdAndEmail(tenantId, user.getEmail());
userValidator.validate(user, User::getTenantId);
if (existing != null && !existing.getId().equals(user.getId())) {
String[] splitEmail = userEmail.split("@");
userEmail = splitEmail[0] + "_" + StringUtils.randomAlphanumeric(15) + "@" + splitEmail[1];
log.warn("[{}] User with email {} already exists. Renaming User email to {}",
tenantId, user.getEmail(), userEmail);
userEmailUpdated = true;
}
user.setEmail(userEmail);
setCustomerId(tenantId, isCreated ? null : userById.getCustomerId(), user, userUpdateMsg);
userValidator.validate(user, User::getTenantId);
if (isCreated) {
user.setId(userId);
}
if (isCreated) {
user.setId(userId);
edgeCtx.getUserService().saveUser(tenantId, user, false);
}
edgeCtx.getUserService().saveUser(tenantId, user, false);
} catch (Exception e) {
log.error("[{}] Failed to process user update msg [{}]", tenantId, userUpdateMsg, e);
throw e;
@ -86,6 +76,20 @@ public abstract class BaseUserProcessor extends BaseEdgeProcessor {
return Pair.of(isCreated, userEmailUpdated);
}
private boolean updateUserEmailIfDuplicateExists(TenantId tenantId, UserId userId, User user) {
String email = user.getEmail();
User userByEmail = edgeCtx.getUserService().findUserByTenantIdAndEmail(tenantId, email);
if (userByEmail != null && !userByEmail.getId().equals(user.getId())) {
String[] splitEmail = email.split("@");
String newEmail = splitEmail[0] + "_" + StringUtils.randomAlphanumeric(15) + "@" + splitEmail[1];
log.warn("[{}] User with email {} already exists. Renaming User email to {}", tenantId, user.getEmail(), newEmail);
user.setEmail(newEmail);
return true;
}
return false;
}
protected void deleteUserAndPushEntityDeletedEventToRuleEngine(TenantId tenantId, UserId userId) {
deleteUserAndPushEntityDeletedEventToRuleEngine(tenantId, userId, null);
}

Write Preview
Loading…
Cancel
Save