dendi-wang/thingsboard
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge branch 'fix/bulk-import' of https://github.com/ViacheslavKlimov/thingsboard

Browse Source
This commit is contained in:
Andrii Shvaika 2021-11-05 17:46:03 +02:00
commit 9aecb1463a
5 changed files with 167 additions and 167 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

73
application/src/main/java/org/thingsboard/server/service/asset/AssetBulkImportService.java
View File

@ -17,78 +17,69 @@ package org.thingsboard.server.service.asset;
import com.fasterxml.jackson.databind.node.ObjectNode;
import com.fasterxml.jackson.databind.node.TextNode;
import lombok.RequiredArgsConstructor;
import org.springframework.stereotype.Service;
import org.thingsboard.common.util.JacksonUtil;
import org.thingsboard.server.cluster.TbClusterService;
import org.thingsboard.server.common.data.EntityType;
import org.thingsboard.server.common.data.asset.Asset;
import org.thingsboard.server.common.data.id.TenantId;
import org.thingsboard.server.dao.asset.AssetService;
import org.thingsboard.server.dao.tenant.TbTenantProfileCache;
import org.thingsboard.server.queue.util.TbCoreComponent;
import org.thingsboard.server.service.action.EntityActionService;
import org.thingsboard.server.service.importing.AbstractBulkImportService;
import org.thingsboard.server.service.importing.BulkImportColumnType;
import org.thingsboard.server.service.importing.BulkImportRequest;
import org.thingsboard.server.service.importing.ImportedEntityInfo;
import org.thingsboard.server.service.security.AccessValidator;
import org.thingsboard.server.service.security.model.SecurityUser;
import org.thingsboard.server.service.security.permission.AccessControlService;
import org.thingsboard.server.service.telemetry.TelemetrySubscriptionService;
import java.util.Map;
import java.util.Optional;
@Service
@TbCoreComponent
@RequiredArgsConstructor
public class AssetBulkImportService extends AbstractBulkImportService<Asset> {
private final AssetService assetService;
public AssetBulkImportService(TelemetrySubscriptionService tsSubscriptionService, TbTenantProfileCache tenantProfileCache,
AccessControlService accessControlService, AccessValidator accessValidator,
EntityActionService entityActionService, TbClusterService clusterService, AssetService assetService) {
super(tsSubscriptionService, tenantProfileCache, accessControlService, accessValidator, entityActionService, clusterService);
this.assetService = assetService;
}
@Override
protected ImportedEntityInfo<Asset> saveEntity(BulkImportRequest importRequest, Map<BulkImportColumnType, String> fields, SecurityUser user) {
ImportedEntityInfo<Asset> importedEntityInfo = new ImportedEntityInfo<>();
Asset asset = new Asset();
asset.setTenantId(user.getTenantId());
setAssetFields(asset, fields);
Asset existingAsset = assetService.findAssetByTenantIdAndName(user.getTenantId(), asset.getName());
if (existingAsset != null && importRequest.getMapping().getUpdate()) {
importedEntityInfo.setOldEntity(new Asset(existingAsset));
importedEntityInfo.setUpdated(true);
existingAsset.update(asset);
asset = existingAsset;
}
asset = assetService.saveAsset(asset);
importedEntityInfo.setEntity(asset);
return importedEntityInfo;
}
private void setAssetFields(Asset asset, Map<BulkImportColumnType, String> fields) {
ObjectNode additionalInfo = (ObjectNode) Optional.ofNullable(asset.getAdditionalInfo()).orElseGet(JacksonUtil::newObjectNode);
protected void setEntityFields(Asset entity, Map<BulkImportColumnType, String> fields) {
ObjectNode additionalInfo = (ObjectNode) Optional.ofNullable(entity.getAdditionalInfo()).orElseGet(JacksonUtil::newObjectNode);
fields.forEach((columnType, value) -> {
switch (columnType) {
case NAME:
asset.setName(value);
entity.setName(value);
break;
case TYPE:
asset.setType(value);
entity.setType(value);
break;
case LABEL:
asset.setLabel(value);
entity.setLabel(value);
break;
case DESCRIPTION:
additionalInfo.set("description", new TextNode(value));
break;
}
});
asset.setAdditionalInfo(additionalInfo);
entity.setAdditionalInfo(additionalInfo);
}
@Override
protected Asset saveEntity(Asset entity, Map<BulkImportColumnType, String> fields) {
return assetService.saveAsset(entity);
}
@Override
protected Asset findOrCreateEntity(TenantId tenantId, String name) {
return Optional.ofNullable(assetService.findAssetByTenantIdAndName(tenantId, name))
.orElseGet(Asset::new);
}
@Override
protected void setOwners(Asset entity, SecurityUser user) {
entity.setTenantId(user.getTenantId());
entity.setCustomerId(user.getCustomerId());
}
@Override
protected EntityType getEntityType() {
return EntityType.ASSET;
}
}

109
application/src/main/java/org/thingsboard/server/service/device/DeviceBulkImportService.java
View File

@ -18,18 +18,19 @@ package org.thingsboard.server.service.device;
import com.fasterxml.jackson.databind.node.BooleanNode;
import com.fasterxml.jackson.databind.node.ObjectNode;
import com.fasterxml.jackson.databind.node.TextNode;
import lombok.RequiredArgsConstructor;
import lombok.SneakyThrows;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.lang3.RandomStringUtils;
import org.apache.commons.lang3.StringUtils;
import org.springframework.stereotype.Service;
import org.thingsboard.common.util.JacksonUtil;
import org.thingsboard.server.cluster.TbClusterService;
import org.thingsboard.server.common.data.Device;
import org.thingsboard.server.common.data.DeviceProfile;
import org.thingsboard.server.common.data.DeviceProfileProvisionType;
import org.thingsboard.server.common.data.DeviceProfileType;
import org.thingsboard.server.common.data.DeviceTransportType;
import org.thingsboard.server.common.data.EntityType;
import org.thingsboard.server.common.data.device.credentials.BasicMqttCredentials;
import org.thingsboard.server.common.data.device.credentials.lwm2m.LwM2MClientCredentials;
import org.thingsboard.server.common.data.device.credentials.lwm2m.LwM2MSecurityMode;
@ -45,17 +46,10 @@ import org.thingsboard.server.dao.device.DeviceCredentialsService;
import org.thingsboard.server.dao.device.DeviceProfileService;
import org.thingsboard.server.dao.device.DeviceService;
import org.thingsboard.server.dao.exception.DeviceCredentialsValidationException;
import org.thingsboard.server.dao.tenant.TbTenantProfileCache;
import org.thingsboard.server.queue.util.TbCoreComponent;
import org.thingsboard.server.service.action.EntityActionService;
import org.thingsboard.server.service.importing.AbstractBulkImportService;
import org.thingsboard.server.service.importing.BulkImportColumnType;
import org.thingsboard.server.service.importing.BulkImportRequest;
import org.thingsboard.server.service.importing.ImportedEntityInfo;
import org.thingsboard.server.service.security.AccessValidator;
import org.thingsboard.server.service.security.model.SecurityUser;
import org.thingsboard.server.service.security.permission.AccessControlService;
import org.thingsboard.server.service.telemetry.TelemetrySubscriptionService;
import java.util.Collection;
import java.util.EnumSet;
@ -68,6 +62,7 @@ import java.util.concurrent.locks.ReentrantLock;
@Service
@TbCoreComponent
@RequiredArgsConstructor
public class DeviceBulkImportService extends AbstractBulkImportService<Device> {
protected final DeviceService deviceService;
protected final DeviceCredentialsService deviceCredentialsService;
@ -75,33 +70,33 @@ public class DeviceBulkImportService extends AbstractBulkImportService<Device> {
private final Lock findOrCreateDeviceProfileLock = new ReentrantLock();
public DeviceBulkImportService(TelemetrySubscriptionService tsSubscriptionService, TbTenantProfileCache tenantProfileCache,
AccessControlService accessControlService, AccessValidator accessValidator,
EntityActionService entityActionService, TbClusterService clusterService,
DeviceService deviceService, DeviceCredentialsService deviceCredentialsService,
DeviceProfileService deviceProfileService) {
super(tsSubscriptionService, tenantProfileCache, accessControlService, accessValidator, entityActionService, clusterService);
this.deviceService = deviceService;
this.deviceCredentialsService = deviceCredentialsService;
this.deviceProfileService = deviceProfileService;
@Override
protected void setEntityFields(Device entity, Map<BulkImportColumnType, String> fields) {
ObjectNode additionalInfo = (ObjectNode) Optional.ofNullable(entity.getAdditionalInfo()).orElseGet(JacksonUtil::newObjectNode);
fields.forEach((columnType, value) -> {
switch (columnType) {
case NAME:
entity.setName(value);
break;
case TYPE:
entity.setType(value);
break;
case LABEL:
entity.setLabel(value);
break;
case DESCRIPTION:
additionalInfo.set("description", new TextNode(value));
break;
case IS_GATEWAY:
additionalInfo.set("gateway", BooleanNode.valueOf(Boolean.parseBoolean(value)));
break;
}
entity.setAdditionalInfo(additionalInfo);
});
}
@Override
protected ImportedEntityInfo<Device> saveEntity(BulkImportRequest importRequest, Map<BulkImportColumnType, String> fields, SecurityUser user) {
ImportedEntityInfo<Device> importedEntityInfo = new ImportedEntityInfo<>();
Device device = new Device();
device.setTenantId(user.getTenantId());
setDeviceFields(device, fields);
Device existingDevice = deviceService.findDeviceByTenantIdAndName(user.getTenantId(), device.getName());
if (existingDevice != null && importRequest.getMapping().getUpdate()) {
importedEntityInfo.setOldEntity(new Device(existingDevice));
importedEntityInfo.setUpdated(true);
existingDevice.updateDevice(device);
device = existingDevice;
}
protected Device saveEntity(Device entity, Map<BulkImportColumnType, String> fields) {
DeviceCredentials deviceCredentials;
try {
deviceCredentials = createDeviceCredentials(fields);
@ -112,42 +107,27 @@ public class DeviceBulkImportService extends AbstractBulkImportService<Device> {
DeviceProfile deviceProfile;
if (deviceCredentials.getCredentialsType() == DeviceCredentialsType.LWM2M_CREDENTIALS) {
deviceProfile = setUpLwM2mDeviceProfile(user.getTenantId(), device);
} else if (StringUtils.isNotEmpty(device.getType())) {
deviceProfile = deviceProfileService.findOrCreateDeviceProfile(user.getTenantId(), device.getType());
deviceProfile = setUpLwM2mDeviceProfile(entity.getTenantId(), entity);
} else if (StringUtils.isNotEmpty(entity.getType())) {
deviceProfile = deviceProfileService.findOrCreateDeviceProfile(entity.getTenantId(), entity.getType());
} else {
deviceProfile = deviceProfileService.findDefaultDeviceProfile(user.getTenantId());
deviceProfile = deviceProfileService.findDefaultDeviceProfile(entity.getTenantId());
}
device.setDeviceProfileId(deviceProfile.getId());
entity.setDeviceProfileId(deviceProfile.getId());
device = deviceService.saveDeviceWithCredentials(device, deviceCredentials);
importedEntityInfo.setEntity(device);
return importedEntityInfo;
return deviceService.saveDeviceWithCredentials(entity, deviceCredentials);
}
private void setDeviceFields(Device device, Map<BulkImportColumnType, String> fields) {
ObjectNode additionalInfo = (ObjectNode) Optional.ofNullable(device.getAdditionalInfo()).orElseGet(JacksonUtil::newObjectNode);
fields.forEach((columnType, value) -> {
switch (columnType) {
case NAME:
device.setName(value);
break;
case TYPE:
device.setType(value);
break;
case LABEL:
device.setLabel(value);
break;
case DESCRIPTION:
additionalInfo.set("description", new TextNode(value));
break;
case IS_GATEWAY:
additionalInfo.set("gateway", BooleanNode.valueOf(Boolean.parseBoolean(value)));
break;
@Override
protected Device findOrCreateEntity(TenantId tenantId, String name) {
return Optional.ofNullable(deviceService.findDeviceByTenantIdAndName(tenantId, name))
.orElseGet(Device::new);
}
device.setAdditionalInfo(additionalInfo);
});
@Override
protected void setOwners(Device entity, SecurityUser user) {
entity.setTenantId(user.getTenantId());
entity.setCustomerId(user.getCustomerId());
}
@SneakyThrows
@ -273,4 +253,9 @@ public class DeviceBulkImportService extends AbstractBulkImportService<Device> {
}
}
@Override
protected EntityType getEntityType() {
return EntityType.DEVICE;
}
}

81
application/src/main/java/org/thingsboard/server/service/edge/EdgeBulkImportService.java
View File

@ -17,90 +17,81 @@ package org.thingsboard.server.service.edge;
import com.fasterxml.jackson.databind.node.ObjectNode;
import com.fasterxml.jackson.databind.node.TextNode;
import lombok.RequiredArgsConstructor;
import org.springframework.stereotype.Service;
import org.thingsboard.common.util.JacksonUtil;
import org.thingsboard.server.cluster.TbClusterService;
import org.thingsboard.server.common.data.EntityType;
import org.thingsboard.server.common.data.edge.Edge;
import org.thingsboard.server.common.data.id.TenantId;
import org.thingsboard.server.dao.edge.EdgeService;
import org.thingsboard.server.dao.tenant.TbTenantProfileCache;
import org.thingsboard.server.queue.util.TbCoreComponent;
import org.thingsboard.server.service.action.EntityActionService;
import org.thingsboard.server.service.importing.AbstractBulkImportService;
import org.thingsboard.server.service.importing.BulkImportColumnType;
import org.thingsboard.server.service.importing.BulkImportRequest;
import org.thingsboard.server.service.importing.ImportedEntityInfo;
import org.thingsboard.server.service.security.AccessValidator;
import org.thingsboard.server.service.security.model.SecurityUser;
import org.thingsboard.server.service.security.permission.AccessControlService;
import org.thingsboard.server.service.telemetry.TelemetrySubscriptionService;
import java.util.Map;
import java.util.Optional;
@Service
@TbCoreComponent
@RequiredArgsConstructor
public class EdgeBulkImportService extends AbstractBulkImportService<Edge> {
private final EdgeService edgeService;
public EdgeBulkImportService(TelemetrySubscriptionService tsSubscriptionService, TbTenantProfileCache tenantProfileCache,
AccessControlService accessControlService, AccessValidator accessValidator,
EntityActionService entityActionService, TbClusterService clusterService, EdgeService edgeService) {
super(tsSubscriptionService, tenantProfileCache, accessControlService, accessValidator, entityActionService, clusterService);
this.edgeService = edgeService;
}
@Override
protected ImportedEntityInfo<Edge> saveEntity(BulkImportRequest importRequest, Map<BulkImportColumnType, String> fields, SecurityUser user) {
ImportedEntityInfo<Edge> importedEntityInfo = new ImportedEntityInfo<>();
Edge edge = new Edge();
edge.setTenantId(user.getTenantId());
setEdgeFields(edge, fields);
Edge existingEdge = edgeService.findEdgeByTenantIdAndName(user.getTenantId(), edge.getName());
if (existingEdge != null && importRequest.getMapping().getUpdate()) {
importedEntityInfo.setOldEntity(new Edge(existingEdge));
importedEntityInfo.setUpdated(true);
existingEdge.update(edge);
edge = existingEdge;
}
edge = edgeService.saveEdge(edge, true);
importedEntityInfo.setEntity(edge);
return importedEntityInfo;
}
private void setEdgeFields(Edge edge, Map<BulkImportColumnType, String> fields) {
ObjectNode additionalInfo = (ObjectNode) Optional.ofNullable(edge.getAdditionalInfo()).orElseGet(JacksonUtil::newObjectNode);
protected void setEntityFields(Edge entity, Map<BulkImportColumnType, String> fields) {
ObjectNode additionalInfo = (ObjectNode) Optional.ofNullable(entity.getAdditionalInfo()).orElseGet(JacksonUtil::newObjectNode);
fields.forEach((columnType, value) -> {
switch (columnType) {
case NAME:
edge.setName(value);
entity.setName(value);
break;
case TYPE:
edge.setType(value);
entity.setType(value);
break;
case LABEL:
edge.setLabel(value);
entity.setLabel(value);
break;
case DESCRIPTION:
additionalInfo.set("description", new TextNode(value));
break;
case EDGE_LICENSE_KEY:
edge.setEdgeLicenseKey(value);
entity.setEdgeLicenseKey(value);
break;
case CLOUD_ENDPOINT:
edge.setCloudEndpoint(value);
entity.setCloudEndpoint(value);
break;
case ROUTING_KEY:
edge.setRoutingKey(value);
entity.setRoutingKey(value);
break;
case SECRET:
edge.setSecret(value);
entity.setSecret(value);
break;
}
});
edge.setAdditionalInfo(additionalInfo);
entity.setAdditionalInfo(additionalInfo);
}
@Override
protected Edge saveEntity(Edge entity, Map<BulkImportColumnType, String> fields) {
return edgeService.saveEdge(entity, true);
}
@Override
protected Edge findOrCreateEntity(TenantId tenantId, String name) {
return Optional.ofNullable(edgeService.findEdgeByTenantIdAndName(tenantId, name))
.orElseGet(Edge::new);
}
@Override
protected void setOwners(Edge entity, SecurityUser user) {
entity.setTenantId(user.getTenantId());
entity.setCustomerId(user.getCustomerId());
}
@Override
protected EntityType getEntityType() {
return EntityType.EDGE;
}
}

67
application/src/main/java/org/thingsboard/server/service/importing/AbstractBulkImportService.java
View File

@ -19,19 +19,21 @@ import com.google.common.util.concurrent.FutureCallback;
import com.google.gson.JsonObject;
import com.google.gson.JsonPrimitive;
import lombok.Data;
import lombok.RequiredArgsConstructor;
import lombok.SneakyThrows;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.exception.ExceptionUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.security.core.context.SecurityContext;
import org.springframework.security.core.context.SecurityContextHolder;
import org.thingsboard.common.util.DonAsynchron;
import org.thingsboard.common.util.ThingsBoardThreadFactory;
import org.thingsboard.server.cluster.TbClusterService;
import org.thingsboard.server.common.data.BaseData;
import org.thingsboard.server.common.data.EntityType;
import org.thingsboard.server.common.data.HasTenantId;
import org.thingsboard.server.common.data.TenantProfile;
import org.thingsboard.server.common.data.audit.ActionType;
import org.thingsboard.server.common.data.id.EntityId;
import org.thingsboard.server.common.data.id.HasId;
import org.thingsboard.server.common.data.id.TenantId;
import org.thingsboard.server.common.data.id.UUIDBased;
import org.thingsboard.server.common.data.kv.AttributeKvEntry;
import org.thingsboard.server.common.data.kv.BasicTsKvEntry;
@ -47,6 +49,7 @@ import org.thingsboard.server.service.security.AccessValidator;
import org.thingsboard.server.service.security.model.SecurityUser;
import org.thingsboard.server.service.security.permission.AccessControlService;
import org.thingsboard.server.service.security.permission.Operation;
import org.thingsboard.server.service.security.permission.Resource;
import org.thingsboard.server.service.telemetry.TelemetrySubscriptionService;
import org.thingsboard.server.utils.CsvUtils;
import org.thingsboard.server.utils.TypeCastUtil;
@ -68,14 +71,17 @@ import java.util.function.Consumer;
import java.util.stream.Collectors;
import java.util.stream.Stream;
@RequiredArgsConstructor
public abstract class AbstractBulkImportService<E extends BaseData<? extends EntityId>> {
protected final TelemetrySubscriptionService tsSubscriptionService;
protected final TbTenantProfileCache tenantProfileCache;
protected final AccessControlService accessControlService;
protected final AccessValidator accessValidator;
protected final EntityActionService entityActionService;
protected final TbClusterService clusterService;
public abstract class AbstractBulkImportService<E extends HasId<? extends EntityId> & HasTenantId> {
@Autowired
private TelemetrySubscriptionService tsSubscriptionService;
@Autowired
private TbTenantProfileCache tenantProfileCache;
@Autowired
private AccessControlService accessControlService;
@Autowired
private AccessValidator accessValidator;
@Autowired
private EntityActionService entityActionService;
private static ThreadPoolExecutor executor;
@ -100,7 +106,7 @@ public abstract class AbstractBulkImportService<E extends BaseData<? extends Ent
entitiesData.forEach(entityData -> DonAsynchron.submit(() -> {
SecurityContextHolder.setContext(securityContext);
ImportedEntityInfo<E> importedEntityInfo = saveEntity(request, entityData.getFields(), user);
ImportedEntityInfo<E> importedEntityInfo = saveEntity(entityData.getFields(), user);
E entity = importedEntityInfo.getEntity();
onEntityImported.accept(importedEntityInfo);
@ -127,12 +133,39 @@ public abstract class AbstractBulkImportService<E extends BaseData<? extends Ent
return result;
}
protected abstract ImportedEntityInfo<E> saveEntity(BulkImportRequest importRequest, Map<BulkImportColumnType, String> fields, SecurityUser user);
@SneakyThrows
private ImportedEntityInfo<E> saveEntity(Map<BulkImportColumnType, String> fields, SecurityUser user) {
ImportedEntityInfo<E> importedEntityInfo = new ImportedEntityInfo<>();
E entity = findOrCreateEntity(user.getTenantId(), fields.get(BulkImportColumnType.NAME));
if (entity.getId() != null) {
importedEntityInfo.setOldEntity((E) entity.getClass().getConstructor(entity.getClass()).newInstance(entity));
importedEntityInfo.setUpdated(true);
} else {
setOwners(entity, user);
}
setEntityFields(entity, fields);
accessControlService.checkPermission(user, Resource.of(getEntityType()), Operation.WRITE, entity.getId(), entity);
E savedEntity = saveEntity(entity, fields);
importedEntityInfo.setEntity(savedEntity);
return importedEntityInfo;
}
protected abstract E findOrCreateEntity(TenantId tenantId, String name);
protected abstract void setOwners(E entity, SecurityUser user);
protected abstract void setEntityFields(E entity, Map<BulkImportColumnType, String> fields);
protected abstract E saveEntity(E entity, Map<BulkImportColumnType, String> fields);
protected abstract EntityType getEntityType();
/*
* Attributes' values are firstly added to JsonObject in order to then make some type cast,
* because we get all values as strings from CSV
* */
private void saveKvs(SecurityUser user, E entity, Map<ColumnMapping, ParsedValue> data) {
Arrays.stream(BulkImportColumnType.values())
.filter(BulkImportColumnType::isKv)

2
application/src/main/java/org/thingsboard/server/service/security/permission/Resource.java
View File

@ -58,7 +58,7 @@ public enum Resource {
public static Resource of(EntityType entityType) {
for (Resource resource : Resource.values()) {
if (resource.getEntityType().get() == entityType) {
if (resource.getEntityType().orElse(null) == entityType) {
return resource;
}
}