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

Remove findDeviceByIdAsync from DeviceService

Browse Source
This commit is contained in:
ViacheslavKlimov 2023-04-18 18:25:35 +03:00
parent a1e816ada4
commit 079b2a5ab7
14 changed files with 117 additions and 188 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

53
application/src/main/java/org/thingsboard/server/service/device/ClaimDevicesServiceImpl.java
View File

@ -90,35 +90,32 @@ public class ClaimDevicesServiceImpl implements ClaimDevicesService {
@Override
public ListenableFuture<Void> registerClaimingInfo(TenantId tenantId, DeviceId deviceId, String secretKey, long durationMs) {
ListenableFuture<Device> deviceFuture = deviceService.findDeviceByIdAsync(tenantId, deviceId);
return Futures.transformAsync(deviceFuture, device -> {
Cache cache = cacheManager.getCache(CLAIM_DEVICES_CACHE);
List<Object> key = constructCacheKey(device.getId());
if (isAllowedClaimingByDefault) {
if (device.getCustomerId().getId().equals(ModelConstants.NULL_UUID)) {
persistInCache(secretKey, durationMs, cache, key);
return Futures.immediateFuture(null);
}
log.warn("The device [{}] has been already claimed!", device.getName());
throw new IllegalArgumentException();
} else {
ListenableFuture<List<AttributeKvEntry>> claimingAllowedFuture = attributesService.find(tenantId, device.getId(),
DataConstants.SERVER_SCOPE, Collections.singletonList(CLAIM_ATTRIBUTE_NAME));
return Futures.transform(claimingAllowedFuture, list -> {
if (list != null && !list.isEmpty()) {
Optional<Boolean> claimingAllowedOptional = list.get(0).getBooleanValue();
if (claimingAllowedOptional.isPresent() && claimingAllowedOptional.get()
&& device.getCustomerId().getId().equals(ModelConstants.NULL_UUID)) {
persistInCache(secretKey, durationMs, cache, key);
return null;
}
}
log.warn("Failed to find claimingAllowed attribute for device or it is already claimed![{}]", device.getName());
throw new IllegalArgumentException();
}, MoreExecutors.directExecutor());
Device device = deviceService.findDeviceById(tenantId, deviceId);
Cache cache = cacheManager.getCache(CLAIM_DEVICES_CACHE);
List<Object> key = constructCacheKey(device.getId());
if (isAllowedClaimingByDefault) {
if (device.getCustomerId().getId().equals(ModelConstants.NULL_UUID)) {
persistInCache(secretKey, durationMs, cache, key);
return Futures.immediateFuture(null);
}
}, MoreExecutors.directExecutor());
log.warn("The device [{}] has been already claimed!", device.getName());
return Futures.immediateFailedFuture(new IllegalArgumentException());
} else {
ListenableFuture<List<AttributeKvEntry>> claimingAllowedFuture = attributesService.find(tenantId, device.getId(),
DataConstants.SERVER_SCOPE, Collections.singletonList(CLAIM_ATTRIBUTE_NAME));
return Futures.transform(claimingAllowedFuture, list -> {
if (list != null && !list.isEmpty()) {
Optional<Boolean> claimingAllowedOptional = list.get(0).getBooleanValue();
if (claimingAllowedOptional.isPresent() && claimingAllowedOptional.get()
&& device.getCustomerId().getId().equals(ModelConstants.NULL_UUID)) {
persistInCache(secretKey, durationMs, cache, key);
return null;
}
}
log.warn("Failed to find claimingAllowed attribute for device or it is already claimed![{}]", device.getName());
throw new IllegalArgumentException();
}, MoreExecutors.directExecutor());
}
}
private ListenableFuture<ClaimDataInfo> getClaimData(Cache cache, Device device) {

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

@ -16,7 +16,6 @@
package org.thingsboard.server.service.edge.rpc.processor.device;
import com.datastax.oss.driver.api.core.uuid.Uuids;
import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.ListenableFuture;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
@ -108,8 +107,8 @@ public abstract class BaseDeviceProcessor extends BaseEdgeProcessor {
public ListenableFuture<Void> processDeviceCredentialsMsg(TenantId tenantId, DeviceCredentialsUpdateMsg deviceCredentialsUpdateMsg) {
log.debug("[{}] Executing processDeviceCredentialsMsg, deviceCredentialsUpdateMsg [{}]", tenantId, deviceCredentialsUpdateMsg);
DeviceId deviceId = new DeviceId(new UUID(deviceCredentialsUpdateMsg.getDeviceIdMSB(), deviceCredentialsUpdateMsg.getDeviceIdLSB()));
ListenableFuture<Device> deviceFuture = deviceService.findDeviceByIdAsync(tenantId, deviceId);
return Futures.transform(deviceFuture, device -> {
return dbCallbackExecutorService.submit(() -> {
Device device = deviceService.findDeviceById(tenantId, deviceId);
if (device != null) {
log.debug("Updating device credentials for device [{}]. New device credentials Id [{}], value [{}]",
device.getName(), deviceCredentialsUpdateMsg.getCredentialsId(), deviceCredentialsUpdateMsg.getCredentialsValue());
@ -129,6 +128,6 @@ public abstract class BaseDeviceProcessor extends BaseEdgeProcessor {
log.warn("Can't find device by id [{}], deviceCredentialsUpdateMsg [{}]", deviceId, deviceCredentialsUpdateMsg);
}
return null;
}, dbCallbackExecutorService);
});
}
}

3
application/src/main/java/org/thingsboard/server/service/security/AccessValidator.java
View File

@ -264,8 +264,7 @@ public class AccessValidator {
if (currentUser.isSystemAdmin()) {
callback.onSuccess(ValidationResult.accessDenied(SYSTEM_ADMINISTRATOR_IS_NOT_ALLOWED_TO_PERFORM_THIS_OPERATION));
} else {
ListenableFuture<Device> deviceFuture = deviceService.findDeviceByIdAsync(currentUser.getTenantId(), new DeviceId(entityId.getId()));
Futures.addCallback(deviceFuture, getCallback(callback, device -> {
Futures.addCallback(Futures.immediateFuture(deviceService.findDeviceById(currentUser.getTenantId(), new DeviceId(entityId.getId()))), getCallback(callback, device -> {
if (device == null) {
return ValidationResult.entityNotFound(DEVICE_WITH_REQUESTED_ID_NOT_FOUND);
} else {

51
application/src/main/java/org/thingsboard/server/service/transport/DefaultTransportApiService.java
View File

@ -315,8 +315,8 @@ public class DefaultTransportApiService implements TransportApiService {
private ListenableFuture<TransportApiResponseMsg> handle(GetOrCreateDeviceFromGatewayRequestMsg requestMsg) {
DeviceId gatewayId = new DeviceId(new UUID(requestMsg.getGatewayIdMSB(), requestMsg.getGatewayIdLSB()));
ListenableFuture<Device> gatewayFuture = deviceService.findDeviceByIdAsync(TenantId.SYS_TENANT_ID, gatewayId);
return Futures.transform(gatewayFuture, gateway -> {
return dbCallbackExecutorService.submit(() -> {
Device gateway = deviceService.findDeviceById(TenantId.SYS_TENANT_ID, gatewayId);
Lock deviceCreationLock = deviceCreationLocks.computeIfAbsent(requestMsg.getDeviceName(), id -> new ReentrantLock());
deviceCreationLock.lock();
try {
@ -382,7 +382,7 @@ public class DefaultTransportApiService implements TransportApiService {
} finally {
deviceCreationLock.unlock();
}
}, dbCallbackExecutorService);
});
}
private ListenableFuture<TransportApiResponseMsg> handle(ProvisionDeviceRequestMsg requestMsg) {
@ -521,30 +521,31 @@ public class DefaultTransportApiService implements TransportApiService {
}
private ListenableFuture<TransportApiResponseMsg> getDeviceInfo(DeviceCredentials credentials) {
return Futures.transform(deviceService.findDeviceByIdAsync(TenantId.SYS_TENANT_ID, credentials.getDeviceId()), device -> {
if (device == null) {
log.trace("[{}] Failed to lookup device by id", credentials.getDeviceId());
return getEmptyTransportApiResponse();
Device device = deviceService.findDeviceById(TenantId.SYS_TENANT_ID, credentials.getDeviceId());
if (device == null) {
log.trace("[{}] Failed to lookup device by id", credentials.getDeviceId());
return getEmptyTransportApiResponseFuture();
}
try {
ValidateDeviceCredentialsResponseMsg.Builder builder = ValidateDeviceCredentialsResponseMsg.newBuilder();
builder.setDeviceInfo(getDeviceInfoProto(device));
DeviceProfile deviceProfile = deviceProfileCache.get(device.getTenantId(), device.getDeviceProfileId());
if (deviceProfile != null) {
builder.setProfileBody(ByteString.copyFrom(dataDecodingEncodingService.encode(deviceProfile)));
} else {
log.warn("[{}] Failed to find device profile [{}] for device. ", device.getId(), device.getDeviceProfileId());
}
try {
ValidateDeviceCredentialsResponseMsg.Builder builder = ValidateDeviceCredentialsResponseMsg.newBuilder();
builder.setDeviceInfo(getDeviceInfoProto(device));
DeviceProfile deviceProfile = deviceProfileCache.get(device.getTenantId(), device.getDeviceProfileId());
if (deviceProfile != null) {
builder.setProfileBody(ByteString.copyFrom(dataDecodingEncodingService.encode(deviceProfile)));
} else {
log.warn("[{}] Failed to find device profile [{}] for device. ", device.getId(), device.getDeviceProfileId());
}
if (!StringUtils.isEmpty(credentials.getCredentialsValue())) {
builder.setCredentialsBody(credentials.getCredentialsValue());
}
return TransportApiResponseMsg.newBuilder()
.setValidateCredResponseMsg(builder.build()).build();
} catch (JsonProcessingException e) {
log.warn("[{}] Failed to lookup device by id", credentials.getDeviceId(), e);
return getEmptyTransportApiResponse();
if (!StringUtils.isEmpty(credentials.getCredentialsValue())) {
builder.setCredentialsBody(credentials.getCredentialsValue());
}
}, MoreExecutors.directExecutor());
return Futures.immediateFuture(TransportApiResponseMsg.newBuilder()
.setValidateCredResponseMsg(builder.build()).build());
} catch (JsonProcessingException e) {
log.warn("[{}] Failed to lookup device by id", credentials.getDeviceId(), e);
return getEmptyTransportApiResponseFuture();
} catch (Exception e) {
return Futures.immediateFailedFuture(e);
}
}
private DeviceInfoProto getDeviceInfoProto(Device device) throws JsonProcessingException {

6
application/src/test/java/org/thingsboard/server/service/transport/DefaultTransportApiServiceTest.java
View File

@ -123,7 +123,7 @@ public class DefaultTransportApiServiceTest {
@Test
public void validateExistingDeviceByX509CertificateStrategy() {
var device = createDevice();
when(deviceService.findDeviceByIdAsync(any(), any())).thenReturn(Futures.immediateFuture(device));
when(deviceService.findDeviceById(any(), any())).thenReturn(device);
var deviceCredentials = createDeviceCredentials(chain[0], device.getId());
when(deviceCredentialsService.findDeviceCredentialsByCredentialsId(any())).thenReturn(deviceCredentials);
@ -139,7 +139,7 @@ public class DefaultTransportApiServiceTest {
var device = createDevice();
when(deviceService.findDeviceByTenantIdAndName(any(), any())).thenReturn(device);
when(deviceService.findDeviceByIdAsync(any(), any())).thenReturn(Futures.immediateFuture(device));
when(deviceService.findDeviceById(any(), any())).thenReturn(device);
var deviceCredentials = createDeviceCredentials(chain[0], device.getId());
when(deviceCredentialsService.findDeviceCredentialsByCredentialsId(any())).thenReturn(null);
@ -150,7 +150,7 @@ public class DefaultTransportApiServiceTest {
service.validateOrCreateDeviceX509Certificate(certificateChain);
verify(deviceProfileService, times(1)).findDeviceProfileByProvisionDeviceKey(any());
verify(deviceService, times(1)).findDeviceByIdAsync(any(), any());
verify(deviceService, times(1)).findDeviceById(any(), any());
verify(deviceCredentialsService, times(1)).findDeviceCredentialsByCredentialsId(any());
verify(deviceProvisionService, times(1)).provisionDeviceViaX509Chain(any(), any());
}

2
common/dao-api/src/main/java/org/thingsboard/server/dao/device/DeviceService.java
View File

@ -44,8 +44,6 @@ public interface DeviceService extends EntityDaoService {
Device findDeviceById(TenantId tenantId, DeviceId deviceId);
ListenableFuture<Device> findDeviceByIdAsync(TenantId tenantId, DeviceId deviceId);
Device findDeviceByTenantIdAndName(TenantId tenantId, String name);
Device saveDevice(Device device, boolean doValidate);

36
dao/src/main/java/org/thingsboard/server/dao/device/DeviceServiceImpl.java
View File

@ -15,7 +15,6 @@
*/
package org.thingsboard.server.dao.device;
import com.google.common.base.Function;
import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.ListenableFuture;
import com.google.common.util.concurrent.MoreExecutors;
@ -73,14 +72,11 @@ import org.thingsboard.server.dao.exception.DataValidationException;
import org.thingsboard.server.dao.service.DataValidator;
import org.thingsboard.server.dao.service.PaginatedRemover;
import javax.annotation.Nullable;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Comparator;
import java.util.List;
import java.util.Optional;
import java.util.UUID;
import java.util.stream.Collectors;
import static org.thingsboard.server.dao.DaoUtil.toUUIDs;
import static org.thingsboard.server.dao.service.Validator.validateId;
@ -139,17 +135,6 @@ public class DeviceServiceImpl extends AbstractCachedEntityService<DeviceCacheKe
}, true);
}
@Override
public ListenableFuture<Device> findDeviceByIdAsync(TenantId tenantId, DeviceId deviceId) {
log.trace("Executing findDeviceById [{}]", deviceId);
validateId(deviceId, INCORRECT_DEVICE_ID + deviceId);
if (TenantId.SYS_TENANT_ID.equals(tenantId)) {
return deviceDao.findByIdAsync(tenantId, deviceId.getId());
} else {
return deviceDao.findDeviceByTenantIdAndIdAsync(tenantId, deviceId.getId());
}
}
@Override
public Device findDeviceByTenantIdAndName(TenantId tenantId, String name) {
log.trace("Executing findDeviceByTenantIdAndName [{}][{}]", tenantId, name);
@ -508,27 +493,20 @@ public class DeviceServiceImpl extends AbstractCachedEntityService<DeviceCacheKe
@Override
public ListenableFuture<List<Device>> findDevicesByQuery(TenantId tenantId, DeviceSearchQuery query) {
ListenableFuture<List<EntityRelation>> relations = relationService.findByQuery(tenantId, query.toEntitySearchQuery());
ListenableFuture<List<Device>> devices = Futures.transformAsync(relations, r -> {
return Futures.transform(relations, r -> {
EntitySearchDirection direction = query.toEntitySearchQuery().getParameters().getDirection();
List<ListenableFuture<Device>> futures = new ArrayList<>();
List<Device> devices = new ArrayList<>();
for (EntityRelation relation : r) {
EntityId entityId = direction == EntitySearchDirection.FROM ? relation.getTo() : relation.getFrom();
if (entityId.getEntityType() == EntityType.DEVICE) {
futures.add(findDeviceByIdAsync(tenantId, new DeviceId(entityId.getId())));
Device device = findDeviceById(tenantId, new DeviceId(entityId.getId()));
if (query.getDeviceTypes().contains(device.getType())) {
devices.add(device);
}
}
}
return Futures.successfulAsList(futures);
return devices;
}, MoreExecutors.directExecutor());
devices = Futures.transform(devices, new Function<>() {
@Nullable
@Override
public List<Device> apply(@Nullable List<Device> deviceList) {
return deviceList == null ? Collections.emptyList() : deviceList.stream().filter(device -> query.getDeviceTypes().contains(device.getType())).collect(Collectors.toList());
}
}, MoreExecutors.directExecutor());
return devices;
}
@Override

14
rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/action/TbCreateRelationNode.java
View File

@ -25,6 +25,7 @@ import org.thingsboard.rule.engine.api.TbNodeConfiguration;
import org.thingsboard.rule.engine.api.TbNodeException;
import org.thingsboard.rule.engine.api.util.TbNodeUtils;
import org.thingsboard.rule.engine.util.EntityContainer;
import org.thingsboard.server.common.data.Device;
import org.thingsboard.server.common.data.id.AssetId;
import org.thingsboard.server.common.data.id.CustomerId;
import org.thingsboard.server.common.data.id.DashboardId;
@ -171,13 +172,12 @@ public class TbCreateRelationNode extends TbAbstractRelationActionNode<TbCreateR
}
private ListenableFuture<Boolean> processDevice(TbContext ctx, EntityContainer entityContainer, SearchDirectionIds sdId, String relationType) {
return Futures.transformAsync(ctx.getDeviceService().findDeviceByIdAsync(ctx.getTenantId(), new DeviceId(entityContainer.getEntityId().getId())), device -> {
if (device != null) {
return processSave(ctx, sdId, relationType);
} else {
return Futures.immediateFuture(true);
}
}, ctx.getDbCallbackExecutor());
Device device = ctx.getDeviceService().findDeviceById(ctx.getTenantId(), new DeviceId(entityContainer.getEntityId().getId()));
if (device != null) {
return processSave(ctx, sdId, relationType);
} else {
return Futures.immediateFuture(true);
}
}
private ListenableFuture<Boolean> processAsset(TbContext ctx, EntityContainer entityContainer, SearchDirectionIds sdId, String relationType) {

7
rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/metadata/TbAbstractGetEntityDetailsNode.java
View File

@ -39,7 +39,6 @@ import java.lang.reflect.Type;
import java.util.Map;
import static org.thingsboard.common.util.DonAsynchron.withCallback;
import static org.thingsboard.rule.engine.api.TbRelationTypes.SUCCESS;
@Slf4j
public abstract class TbAbstractGetEntityDetailsNode<C extends TbAbstractGetEntityDetailsNodeConfiguration> implements TbNode {
@ -67,7 +66,7 @@ public abstract class TbAbstractGetEntityDetailsNode<C extends TbAbstractGetEnti
protected abstract ListenableFuture<TbMsg> getDetails(TbContext ctx, TbMsg msg);
protected abstract ListenableFuture<ContactBased> getContactBasedListenableFuture(TbContext ctx, TbMsg msg);
protected abstract ListenableFuture<? extends ContactBased> getContactBasedListenableFuture(TbContext ctx, TbMsg msg);
protected MessageData getDataAsJson(TbMsg msg) {
if (this.config.isAddToMetadata()) {
@ -79,7 +78,7 @@ public abstract class TbAbstractGetEntityDetailsNode<C extends TbAbstractGetEnti
protected ListenableFuture<TbMsg> getTbMsgListenableFuture(TbContext ctx, TbMsg msg, MessageData messageData, String prefix) {
if (!this.config.getDetailsList().isEmpty()) {
ListenableFuture<ContactBased> contactBasedListenableFuture = getContactBasedListenableFuture(ctx, msg);
ListenableFuture<? extends ContactBased> contactBasedListenableFuture = getContactBasedListenableFuture(ctx, msg);
ListenableFuture<JsonElement> resultObject = addContactProperties(messageData.getData(), contactBasedListenableFuture, prefix);
return transformMsg(ctx, msg, resultObject, messageData);
} else {
@ -102,7 +101,7 @@ public abstract class TbAbstractGetEntityDetailsNode<C extends TbAbstractGetEnti
}, MoreExecutors.directExecutor());
}
private ListenableFuture<JsonElement> addContactProperties(JsonElement data, ListenableFuture<ContactBased> entityFuture, String prefix) {
private ListenableFuture<JsonElement> addContactProperties(JsonElement data, ListenableFuture<? extends ContactBased> entityFuture, String prefix) {
return Futures.transformAsync(entityFuture, contactBased -> {
if (contactBased != null) {
JsonElement jsonElement = null;

107
rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/metadata/TbGetCustomerDetailsNode.java
View File

@ -26,6 +26,8 @@ import org.thingsboard.rule.engine.api.TbNodeException;
import org.thingsboard.rule.engine.api.util.TbNodeUtils;
import org.thingsboard.server.common.data.ContactBased;
import org.thingsboard.server.common.data.Customer;
import org.thingsboard.server.common.data.HasCustomerId;
import org.thingsboard.server.common.data.HasName;
import org.thingsboard.server.common.data.id.AssetId;
import org.thingsboard.server.common.data.id.DeviceId;
import org.thingsboard.server.common.data.id.EdgeId;
@ -59,81 +61,44 @@ public class TbGetCustomerDetailsNode extends TbAbstractGetEntityDetailsNode<TbG
}
@Override
protected ListenableFuture<ContactBased> getContactBasedListenableFuture(TbContext ctx, TbMsg msg) {
return Futures.transformAsync(getCustomer(ctx, msg), customer -> {
if (customer != null) {
return Futures.immediateFuture(customer);
protected ListenableFuture<? extends ContactBased> getContactBasedListenableFuture(TbContext ctx, TbMsg msg) {
return getCustomer(ctx, msg);
}
private ListenableFuture<Customer> getCustomer(TbContext ctx, TbMsg msg) {
ListenableFuture<? extends HasCustomerId> entityFuture;
switch (msg.getOriginator().getEntityType()) { // TODO: use EntityServiceRegistry
case DEVICE:
entityFuture = Futures.immediateFuture(ctx.getDeviceService().findDeviceById(ctx.getTenantId(), (DeviceId) msg.getOriginator()));
break;
case ASSET:
entityFuture = ctx.getAssetService().findAssetByIdAsync(ctx.getTenantId(), (AssetId) msg.getOriginator());
break;
case ENTITY_VIEW:
entityFuture = ctx.getEntityViewService().findEntityViewByIdAsync(ctx.getTenantId(), (EntityViewId) msg.getOriginator());
break;
case USER:
entityFuture = ctx.getUserService().findUserByIdAsync(ctx.getTenantId(), (UserId) msg.getOriginator());
break;
case EDGE:
entityFuture = ctx.getEdgeService().findEdgeByIdAsync(ctx.getTenantId(), (EdgeId) msg.getOriginator());
break;
default:
throw new RuntimeException(msg.getOriginator().getEntityType().getNormalName() + " entities not supported");
}
return Futures.transformAsync(entityFuture, entity -> {
if (entity != null) {
if (!entity.getCustomerId().isNullUid()) {
return ctx.getCustomerService().findCustomerByIdAsync(ctx.getTenantId(), entity.getCustomerId());
} else {
throw new RuntimeException(msg.getOriginator().getEntityType().getNormalName() +
(entity instanceof HasName ? " with name '" + ((HasName) entity).getName() + "'" : "")
+ " is not assigned to Customer");
}
} else {
return Futures.immediateFuture(null);
}
}, MoreExecutors.directExecutor());
}
private ListenableFuture<Customer> getCustomer(TbContext ctx, TbMsg msg) {
switch (msg.getOriginator().getEntityType()) {
case DEVICE:
return Futures.transformAsync(ctx.getDeviceService().findDeviceByIdAsync(ctx.getTenantId(), new DeviceId(msg.getOriginator().getId())), device -> {
if (device != null) {
if (!device.getCustomerId().isNullUid()) {
return ctx.getCustomerService().findCustomerByIdAsync(ctx.getTenantId(), device.getCustomerId());
} else {
throw new RuntimeException("Device with name '" + device.getName() + "' is not assigned to Customer.");
}
} else {
return Futures.immediateFuture(null);
}
}, MoreExecutors.directExecutor());
case ASSET:
return Futures.transformAsync(ctx.getAssetService().findAssetByIdAsync(ctx.getTenantId(), new AssetId(msg.getOriginator().getId())), asset -> {
if (asset != null) {
if (!asset.getCustomerId().isNullUid()) {
return ctx.getCustomerService().findCustomerByIdAsync(ctx.getTenantId(), asset.getCustomerId());
} else {
throw new RuntimeException("Asset with name '" + asset.getName() + "' is not assigned to Customer.");
}
} else {
return Futures.immediateFuture(null);
}
}, MoreExecutors.directExecutor());
case ENTITY_VIEW:
return Futures.transformAsync(ctx.getEntityViewService().findEntityViewByIdAsync(ctx.getTenantId(), new EntityViewId(msg.getOriginator().getId())), entityView -> {
if (entityView != null) {
if (!entityView.getCustomerId().isNullUid()) {
return ctx.getCustomerService().findCustomerByIdAsync(ctx.getTenantId(), entityView.getCustomerId());
} else {
throw new RuntimeException("EntityView with name '" + entityView.getName() + "' is not assigned to Customer.");
}
} else {
return Futures.immediateFuture(null);
}
}, MoreExecutors.directExecutor());
case USER:
return Futures.transformAsync(ctx.getUserService().findUserByIdAsync(ctx.getTenantId(), new UserId(msg.getOriginator().getId())), user -> {
if (user != null) {
if (!user.getCustomerId().isNullUid()) {
return ctx.getCustomerService().findCustomerByIdAsync(ctx.getTenantId(), user.getCustomerId());
} else {
throw new RuntimeException("User with name '" + user.getName() + "' is not assigned to Customer.");
}
} else {
return Futures.immediateFuture(null);
}
}, MoreExecutors.directExecutor());
case EDGE:
return Futures.transformAsync(ctx.getEdgeService().findEdgeByIdAsync(ctx.getTenantId(), new EdgeId(msg.getOriginator().getId())), edge -> {
if (edge != null) {
if (!edge.getCustomerId().isNullUid()) {
return ctx.getCustomerService().findCustomerByIdAsync(ctx.getTenantId(), edge.getCustomerId());
} else {
throw new RuntimeException("Edge with name '" + edge.getName() + "' is not assigned to Customer.");
}
} else {
return Futures.immediateFuture(null);
}
}, MoreExecutors.directExecutor());
default:
throw new RuntimeException("Entity with entityType '" + msg.getOriginator().getEntityType() + "' is not supported.");
}
}
}

10
rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/metadata/TbGetTenantDetailsNode.java
View File

@ -53,13 +53,7 @@ public class TbGetTenantDetailsNode extends TbAbstractGetEntityDetailsNode<TbGet
}
@Override
protected ListenableFuture<ContactBased> getContactBasedListenableFuture(TbContext ctx, TbMsg msg) {
return Futures.transformAsync(ctx.getTenantService().findTenantByIdAsync(ctx.getTenantId(), ctx.getTenantId()), tenant -> {
if (tenant != null) {
return Futures.immediateFuture(tenant);
} else {
return Futures.immediateFuture(null);
}
}, MoreExecutors.directExecutor());
protected ListenableFuture<? extends ContactBased> getContactBasedListenableFuture(TbContext ctx, TbMsg msg) {
return ctx.getTenantService().findTenantByIdAsync(ctx.getTenantId(), ctx.getTenantId());
}
}

3
rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/util/EntitiesCustomerIdAsyncLoader.java
View File

@ -31,7 +31,6 @@ public class EntitiesCustomerIdAsyncLoader {
public static ListenableFuture<CustomerId> findEntityIdAsync(TbContext ctx, EntityId original) {
switch (original.getEntityType()) {
case CUSTOMER:
return Futures.immediateFuture((CustomerId) original);
@ -40,7 +39,7 @@ public class EntitiesCustomerIdAsyncLoader {
case ASSET:
return getCustomerAsync(ctx.getAssetService().findAssetByIdAsync(ctx.getTenantId(), (AssetId) original));
case DEVICE:
return getCustomerAsync(ctx.getDeviceService().findDeviceByIdAsync(ctx.getTenantId(), (DeviceId) original));
return getCustomerAsync(Futures.immediateFuture(ctx.getDeviceService().findDeviceById(ctx.getTenantId(), (DeviceId) original)));
default:
return Futures.immediateFailedFuture(new TbNodeException("Unexpected original EntityType " + original.getEntityType()));
}

4
rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/util/EntitiesFieldsAsyncLoader.java
View File

@ -37,7 +37,7 @@ import java.util.function.Function;
public class EntitiesFieldsAsyncLoader {
public static ListenableFuture<EntityFieldsData> findAsync(TbContext ctx, EntityId original) {
switch (original.getEntityType()) {
switch (original.getEntityType()) { // TODO: use EntityServiceRegistry
case TENANT:
return getAsync(ctx.getTenantService().findTenantByIdAsync(ctx.getTenantId(), (TenantId) original),
EntityFieldsData::new);
@ -51,7 +51,7 @@ public class EntitiesFieldsAsyncLoader {
return getAsync(ctx.getAssetService().findAssetByIdAsync(ctx.getTenantId(), (AssetId) original),
EntityFieldsData::new);
case DEVICE:
return getAsync(ctx.getDeviceService().findDeviceByIdAsync(ctx.getTenantId(), (DeviceId) original),
return getAsync(Futures.immediateFuture(ctx.getDeviceService().findDeviceById(ctx.getTenantId(), (DeviceId) original)),
EntityFieldsData::new);
case ALARM:
return getAsync(ctx.getAlarmService().findAlarmByIdAsync(ctx.getTenantId(), (AlarmId) original),

2
rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/metadata/AbstractAttributeNodeTest.java
View File

@ -219,7 +219,7 @@ public abstract class AbstractAttributeNodeTest {
void mockFindDevice(Device device) {
when(ctx.getDeviceService()).thenReturn(deviceService);
when(deviceService.findDeviceByIdAsync(any(), eq(device.getId()))).thenReturn(Futures.immediateFuture(device));
when(deviceService.findDeviceById(any(), eq(device.getId()))).thenReturn(device);
}
void mockFindAsset(Asset asset) {