diff --git a/application/src/main/java/org/thingsboard/server/actors/ActorSystemContext.java b/application/src/main/java/org/thingsboard/server/actors/ActorSystemContext.java index 8a6c17726c..d000b46c89 100644 --- a/application/src/main/java/org/thingsboard/server/actors/ActorSystemContext.java +++ b/application/src/main/java/org/thingsboard/server/actors/ActorSystemContext.java @@ -116,6 +116,7 @@ import org.thingsboard.server.service.apiusage.TbApiUsageStateService; import org.thingsboard.server.service.cf.CalculatedFieldProcessingService; import org.thingsboard.server.service.cf.CalculatedFieldQueueService; import org.thingsboard.server.service.cf.CalculatedFieldStateService; +import org.thingsboard.server.service.cf.OwnerService; import org.thingsboard.server.service.cf.ctx.state.ArgumentEntry; import org.thingsboard.server.service.component.ComponentDiscoveryService; import org.thingsboard.server.service.edge.rpc.EdgeRpcService; @@ -566,6 +567,10 @@ public class ActorSystemContext { @Getter private JobManager jobManager; + @Autowired + @Getter + private OwnerService ownerService; + @Value("${actors.session.max_concurrent_sessions_per_device:1}") @Getter private int maxConcurrentSessionsPerDevice; diff --git a/application/src/main/java/org/thingsboard/server/actors/calculatedField/CalculatedFieldArgumentResetMsg.java b/application/src/main/java/org/thingsboard/server/actors/calculatedField/CalculatedFieldArgumentResetMsg.java new file mode 100644 index 0000000000..8b5927827e --- /dev/null +++ b/application/src/main/java/org/thingsboard/server/actors/calculatedField/CalculatedFieldArgumentResetMsg.java @@ -0,0 +1,37 @@ +/** + * Copyright © 2016-2025 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.actors.calculatedField; + +import lombok.Data; +import org.thingsboard.server.common.data.id.TenantId; +import org.thingsboard.server.common.msg.MsgType; +import org.thingsboard.server.common.msg.ToCalculatedFieldSystemMsg; +import org.thingsboard.server.common.msg.queue.TbCallback; +import org.thingsboard.server.service.cf.ctx.state.CalculatedFieldCtx; + +@Data +public class CalculatedFieldArgumentResetMsg implements ToCalculatedFieldSystemMsg { + + private final TenantId tenantId; + private final CalculatedFieldCtx ctx; + private final TbCallback callback; + + @Override + public MsgType getMsgType() { + return MsgType.CF_ARGUMENT_RESET_MSG; + } + +} diff --git a/application/src/main/java/org/thingsboard/server/actors/calculatedField/CalculatedFieldEntityActor.java b/application/src/main/java/org/thingsboard/server/actors/calculatedField/CalculatedFieldEntityActor.java index e0f70509a4..0db3cfdb4c 100644 --- a/application/src/main/java/org/thingsboard/server/actors/calculatedField/CalculatedFieldEntityActor.java +++ b/application/src/main/java/org/thingsboard/server/actors/calculatedField/CalculatedFieldEntityActor.java @@ -84,6 +84,9 @@ public class CalculatedFieldEntityActor extends AbstractCalculatedFieldActor { case CF_ALARM_ACTION_MSG: processor.process((CalculatedFieldAlarmActionMsg) msg); break; + case CF_ARGUMENT_RESET_MSG: + processor.process((CalculatedFieldArgumentResetMsg) msg); + break; default: return false; } diff --git a/application/src/main/java/org/thingsboard/server/actors/calculatedField/CalculatedFieldEntityMessageProcessor.java b/application/src/main/java/org/thingsboard/server/actors/calculatedField/CalculatedFieldEntityMessageProcessor.java index ee425bbf90..ecfb258cd1 100644 --- a/application/src/main/java/org/thingsboard/server/actors/calculatedField/CalculatedFieldEntityMessageProcessor.java +++ b/application/src/main/java/org/thingsboard/server/actors/calculatedField/CalculatedFieldEntityMessageProcessor.java @@ -155,6 +155,24 @@ public class CalculatedFieldEntityMessageProcessor extends AbstractContextAwareM } } + public void process(CalculatedFieldArgumentResetMsg msg) throws CalculatedFieldException { + log.debug("[{}] Processing CF argument reset msg.", entityId); + var ctx = msg.getCtx(); + var callback = new MultipleTbCallback(CALLBACKS_PER_CF, msg.getCallback()); + try { + Map dynamicSourceArgs = ctx.getArguments().entrySet().stream() + .filter(entry -> entry.getValue().hasOwnerSource()) + .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue)); + + Map fetchedArgs = cfService.fetchArgsFromDb(tenantId, entityId, dynamicSourceArgs); + fetchedArgs.values().forEach(arg -> arg.setForceResetPrevious(true)); + + processArgumentValuesUpdate(ctx, Collections.singletonList(ctx.getCfId()), callback, fetchedArgs, null, null); + } catch (Exception e) { + throw CalculatedFieldException.builder().ctx(ctx).eventEntity(entityId).cause(e).build(); + } + } + public void process(CalculatedFieldEntityDeleteMsg msg) { log.debug("[{}] Processing CF entity delete msg.", msg.getEntityId()); if (this.entityId.equals(msg.getEntityId())) { @@ -422,11 +440,7 @@ public class CalculatedFieldEntityMessageProcessor extends AbstractContextAwareM } private Map mapToArguments(CalculatedFieldCtx ctx, EntityId entityId, List data) { - var argNames = ctx.getLinkedEntityArguments().get(entityId); - if (argNames.isEmpty()) { - return Collections.emptyMap(); - } - return mapToArguments(argNames, data); + return mapToArguments(ctx.getLinkedAndDynamicArgs(entityId), data); } private Map mapToArguments(Map argNames, List data) { @@ -454,11 +468,11 @@ public class CalculatedFieldEntityMessageProcessor extends AbstractContextAwareM } private Map mapToArguments(CalculatedFieldCtx ctx, EntityId entityId, AttributeScopeProto scope, List attrDataList) { - var argNames = ctx.getLinkedEntityArguments().get(entityId); + var argNames = ctx.getLinkedAndDynamicArgs(entityId); if (argNames.isEmpty()) { return Collections.emptyMap(); } - List geofencingArgumentNames = ctx.getLinkedEntityGeofencingArgumentNames(); + List geofencingArgumentNames = ctx.getLinkedEntityAndCurrentOwnerGeofencingArgumentNames(); return mapToArguments(entityId, argNames, geofencingArgumentNames, scope, attrDataList); } @@ -480,11 +494,7 @@ public class CalculatedFieldEntityMessageProcessor extends AbstractContextAwareM } private Map mapToArgumentsWithDefaultValue(CalculatedFieldCtx ctx, EntityId entityId, AttributeScopeProto scope, List removedAttrKeys) { - var argNames = ctx.getLinkedEntityArguments().get(entityId); - if (argNames.isEmpty()) { - return Collections.emptyMap(); - } - return mapToArgumentsWithDefaultValue(argNames, ctx.getArguments(), scope, removedAttrKeys); + return mapToArgumentsWithDefaultValue(ctx.getLinkedAndDynamicArgs(entityId), ctx.getArguments(), scope, removedAttrKeys); } private Map mapToArgumentsWithDefaultValue(CalculatedFieldCtx ctx, AttributeScopeProto scope, List removedAttrKeys) { diff --git a/application/src/main/java/org/thingsboard/server/actors/calculatedField/CalculatedFieldManagerMessageProcessor.java b/application/src/main/java/org/thingsboard/server/actors/calculatedField/CalculatedFieldManagerMessageProcessor.java index 299a2bc8b9..6aa501cb10 100644 --- a/application/src/main/java/org/thingsboard/server/actors/calculatedField/CalculatedFieldManagerMessageProcessor.java +++ b/application/src/main/java/org/thingsboard/server/actors/calculatedField/CalculatedFieldManagerMessageProcessor.java @@ -49,6 +49,7 @@ import org.thingsboard.server.dao.device.DeviceService; import org.thingsboard.server.queue.settings.TbQueueCalculatedFieldSettings; import org.thingsboard.server.service.cf.CalculatedFieldProcessingService; import org.thingsboard.server.service.cf.CalculatedFieldStateService; +import org.thingsboard.server.service.cf.OwnerService; import org.thingsboard.server.service.cf.cache.TenantEntityProfileCache; import org.thingsboard.server.service.cf.ctx.CalculatedFieldEntityCtxId; import org.thingsboard.server.service.cf.ctx.state.CalculatedFieldCtx; @@ -58,8 +59,10 @@ import org.thingsboard.server.service.profile.TbDeviceProfileCache; import java.util.ArrayList; import java.util.Collections; import java.util.HashMap; +import java.util.HashSet; import java.util.List; import java.util.Map; +import java.util.Set; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.CopyOnWriteArrayList; import java.util.concurrent.ScheduledFuture; @@ -77,6 +80,7 @@ public class CalculatedFieldManagerMessageProcessor extends AbstractContextAware private final Map calculatedFields = new HashMap<>(); private final Map> entityIdCalculatedFields = new HashMap<>(); private final Map> entityIdCalculatedFieldLinks = new HashMap<>(); + private final Map> ownerEntities = new HashMap<>(); private final Map> cfDynamicArgumentsRefreshTasks = new ConcurrentHashMap<>(); private ScheduledFuture cfsReevaluationTask; @@ -88,6 +92,7 @@ public class CalculatedFieldManagerMessageProcessor extends AbstractContextAware private final TbAssetProfileCache assetProfileCache; private final TbDeviceProfileCache deviceProfileCache; private final TenantEntityProfileCache entityProfileCache; + private final OwnerService ownerService; private final TbQueueCalculatedFieldSettings cfSettings; protected final TenantId tenantId; @@ -103,6 +108,7 @@ public class CalculatedFieldManagerMessageProcessor extends AbstractContextAware this.assetProfileCache = systemContext.getAssetProfileCache(); this.deviceProfileCache = systemContext.getDeviceProfileCache(); this.entityProfileCache = new TenantEntityProfileCache(); + this.ownerService = systemContext.getOwnerService(); this.cfSettings = systemContext.getCalculatedFieldSettings(); this.tenantId = tenantId; } @@ -128,7 +134,7 @@ public class CalculatedFieldManagerMessageProcessor extends AbstractContextAware public void onCacheInitMsg(CalculatedFieldCacheInitMsg msg) { log.debug("[{}] Processing CF actor init message.", msg.getTenantId().getId()); - initEntityProfileCache(); + initEntitiesCache(); initCalculatedFields(); scheduleCfsReevaluation(); msg.getCallback().onSuccess(); @@ -251,6 +257,7 @@ public class CalculatedFieldManagerMessageProcessor extends AbstractContextAware if (profileId != null) { entityProfileCache.add(profileId, entityId); } + updateEntityOwner(entityId); if (!isMyPartition(entityId, callback)) { return; } @@ -283,11 +290,16 @@ public class CalculatedFieldManagerMessageProcessor extends AbstractContextAware } else { callback.onSuccess(); } + } else if (msg.isOwnerChanged()) { + onEntityOwnerChanged(msg, callback); + } else { + callback.onSuccess(); } } private void onEntityDeleted(ComponentLifecycleMsg msg, TbCallback callback) { entityProfileCache.removeEntityId(msg.getEntityId()); + ownerEntities.values().forEach(entities -> entities.remove(msg.getEntityId())); if (isMyPartition(msg.getEntityId(), callback)) { log.debug("Pushing entity lifecycle msg to specific actor [{}]", msg.getEntityId()); getOrCreateActor(msg.getEntityId()).tell(new CalculatedFieldEntityDeleteMsg(tenantId, msg.getEntityId(), callback)); @@ -415,8 +427,8 @@ public class CalculatedFieldManagerMessageProcessor extends AbstractContextAware public void onTelemetryMsg(CalculatedFieldTelemetryMsg msg) { EntityId entityId = msg.getEntityId(); log.debug("Received telemetry msg from entity [{}]", entityId); - // 2 = 1 for CF processing + 1 for links processing - MultipleTbCallback callback = new MultipleTbCallback(2, msg.getCallback()); + // 3 = 1 for CF processing + 1 for links processing + 1 for owner entity processing + MultipleTbCallback callback = new MultipleTbCallback(3, msg.getCallback()); // process all cfs related to entity, or it's profile; var entityIdFields = getCalculatedFieldsByEntityId(entityId); var profileIdFields = getCalculatedFieldsByEntityId(getProfileId(tenantId, entityId)); @@ -434,6 +446,17 @@ public class CalculatedFieldManagerMessageProcessor extends AbstractContextAware } else { callback.onSuccess(); } + // process all cfs related to owner entity + if (entityId.getEntityType().isOneOf(EntityType.TENANT, EntityType.CUSTOMER)) { + List ownerCFs = filterOwnerEntitiesCFs(msg); + if (!ownerCFs.isEmpty()) { + cfExecService.pushMsgToLinks(msg, ownerCFs, callback); + } else { + callback.onSuccess(); + } + } else { + callback.onSuccess(); + } } public void onLinkedTelemetryMsg(CalculatedFieldLinkedTelemetryMsg msg) { @@ -456,6 +479,31 @@ public class CalculatedFieldManagerMessageProcessor extends AbstractContextAware } } + private void onEntityOwnerChanged(ComponentLifecycleMsg msg, TbCallback msgCallback) { + EntityId entityId = msg.getEntityId(); + log.debug("Received changed owner msg from entity [{}]", entityId); + updateEntityOwner(entityId); + List cfs = new ArrayList<>(); + cfs.addAll(getCalculatedFieldsByEntityId(entityId)); + cfs.addAll(getCalculatedFieldsByEntityId(getProfileId(tenantId, entityId))); + if (cfs.isEmpty()) { + msgCallback.onSuccess(); + return; + } + MultipleTbCallback callback = new MultipleTbCallback(cfs.size(), msgCallback); + cfs.forEach(cf -> { + if (isMyPartition(entityId, callback)) { + if (cf.hasCurrentOwnerSourceArguments()) { + CalculatedFieldArgumentResetMsg argResetMsg = new CalculatedFieldArgumentResetMsg(tenantId, cf, callback); + log.debug("Pushing CF argument reset msg to specific actor [{}]", entityId); + getOrCreateActor(entityId).tell(argResetMsg); + } else { + callback.onSuccess(); + } + } + }); + } + private List filterCalculatedFieldLinks(CalculatedFieldTelemetryMsg msg) { EntityId entityId = msg.getEntityId(); var proto = msg.getProto(); @@ -469,6 +517,27 @@ public class CalculatedFieldManagerMessageProcessor extends AbstractContextAware return result; } + private List filterOwnerEntitiesCFs(CalculatedFieldTelemetryMsg msg) { + Set entities = getOwnerEntities(msg.getEntityId()); + var proto = msg.getProto(); + List result = new ArrayList<>(); + for (var entityId : entities) { + var ownerEntityCFs = getCalculatedFieldsByEntityId(entityId); + for (var ctx : ownerEntityCFs) { + if (ctx.dynamicSourceMatches(proto)) { + result.add(new CalculatedFieldEntityCtxId(tenantId, ctx.getCfId(), entityId)); + } + } + var ownerEntityProfileCFs = getCalculatedFieldsByEntityId(getProfileId(tenantId, entityId)); + for (var ctx : ownerEntityProfileCFs) { + if (ctx.dynamicSourceMatches(proto)) { + result.add(new CalculatedFieldEntityCtxId(tenantId, ctx.getCfId(), entityId)); + } + } + } + return result; + } + private List getCalculatedFieldsByEntityId(EntityId entityId) { if (entityId == null) { return Collections.emptyList(); @@ -491,6 +560,17 @@ public class CalculatedFieldManagerMessageProcessor extends AbstractContextAware return result; } + private Set getOwnerEntities(EntityId entityId) { + if (entityId == null) { + return Collections.emptySet(); + } + var result = ownerEntities.get(entityId); + if (result == null) { + result = Collections.emptySet(); + } + return result; + } + private void scheduleDynamicArgumentsRefreshTaskForCfIfNeeded(CalculatedFieldCtx cfCtx) { CalculatedField cf = cfCtx.getCalculatedField(); if (!(cf.getConfiguration() instanceof ScheduledUpdateSupportedCalculatedFieldConfiguration scheduledCfConfig)) { @@ -623,12 +703,13 @@ public class CalculatedFieldManagerMessageProcessor extends AbstractContextAware entityIdCalculatedFieldLinks.computeIfAbsent(link.getEntityId(), id -> new CopyOnWriteArrayList<>()).add(link); } - private void initEntityProfileCache() { + private void initEntitiesCache() { PageDataIterable deviceIdInfos = new PageDataIterable<>(pageLink -> deviceService.findProfileEntityIdInfosByTenantId(tenantId, pageLink), cfSettings.getInitTenantFetchPackSize()); for (ProfileEntityIdInfo idInfo : deviceIdInfos) { log.trace("Processing device record: {}", idInfo); try { entityProfileCache.add(idInfo.getProfileId(), idInfo.getEntityId()); + ownerEntities.computeIfAbsent(idInfo.getOwnerId(), ownerId -> new HashSet<>()).add(idInfo.getEntityId()); } catch (Exception e) { log.error("Failed to process device record: {}", idInfo, e); } @@ -638,12 +719,19 @@ public class CalculatedFieldManagerMessageProcessor extends AbstractContextAware log.trace("Processing asset record: {}", idInfo); try { entityProfileCache.add(idInfo.getProfileId(), idInfo.getEntityId()); + ownerEntities.computeIfAbsent(idInfo.getOwnerId(), ownerId -> new HashSet<>()).add(idInfo.getEntityId()); } catch (Exception e) { log.error("Failed to process asset record: {}", idInfo, e); } } } + private void updateEntityOwner(EntityId entityId) { + ownerEntities.values().forEach(entities -> entities.remove(entityId)); + EntityId owner = ownerService.getOwner(tenantId, entityId); + ownerEntities.computeIfAbsent(owner, ownerId -> new HashSet<>()).add(entityId); + } + private void applyToTargetCfEntityActors(CalculatedFieldCtx ctx, TbCallback callback, BiConsumer action) { diff --git a/application/src/main/java/org/thingsboard/server/service/cf/AbstractCalculatedFieldProcessingService.java b/application/src/main/java/org/thingsboard/server/service/cf/AbstractCalculatedFieldProcessingService.java index 6ecf7c5974..343db5286f 100644 --- a/application/src/main/java/org/thingsboard/server/service/cf/AbstractCalculatedFieldProcessingService.java +++ b/application/src/main/java/org/thingsboard/server/service/cf/AbstractCalculatedFieldProcessingService.java @@ -19,11 +19,13 @@ import com.google.common.util.concurrent.Futures; import com.google.common.util.concurrent.ListenableFuture; import com.google.common.util.concurrent.ListeningExecutorService; import com.google.common.util.concurrent.MoreExecutors; +import jakarta.annotation.Nullable; import jakarta.annotation.PostConstruct; import jakarta.annotation.PreDestroy; import lombok.Data; import lombok.extern.slf4j.Slf4j; import org.thingsboard.common.util.ThingsBoardExecutors; +import org.thingsboard.server.common.data.EntityType; import org.thingsboard.server.common.data.cf.configuration.Argument; import org.thingsboard.server.common.data.cf.configuration.ArgumentType; import org.thingsboard.server.common.data.cf.configuration.RelationQueryDynamicSourceConfiguration; @@ -45,6 +47,7 @@ import org.thingsboard.server.dao.usagerecord.ApiLimitService; import org.thingsboard.server.service.cf.ctx.state.ArgumentEntry; import org.thingsboard.server.service.cf.ctx.state.CalculatedFieldCtx; +import java.util.Collections; import java.util.HashMap; import java.util.List; import java.util.Map; @@ -55,6 +58,7 @@ import java.util.stream.Collectors; import static org.thingsboard.server.common.data.cf.configuration.geofencing.EntityCoordinates.ENTITY_ID_LATITUDE_ARGUMENT_KEY; import static org.thingsboard.server.common.data.cf.configuration.geofencing.EntityCoordinates.ENTITY_ID_LONGITUDE_ARGUMENT_KEY; +import static org.thingsboard.server.utils.CalculatedFieldArgumentUtils.createDefaultAttributeEntry; import static org.thingsboard.server.utils.CalculatedFieldArgumentUtils.createDefaultKvEntry; import static org.thingsboard.server.utils.CalculatedFieldArgumentUtils.transformSingleValueArgument; @@ -66,6 +70,7 @@ public abstract class AbstractCalculatedFieldProcessingService { protected final TimeseriesService timeseriesService; protected final ApiLimitService apiLimitService; protected final RelationService relationService; + protected final OwnerService ownerService; protected ListeningExecutorService calculatedFieldCallbackExecutor; @@ -84,14 +89,14 @@ public abstract class AbstractCalculatedFieldProcessingService { protected abstract String getExecutorNamePrefix(); - protected ListenableFuture> fetchArguments(CalculatedFieldCtx ctx, EntityId entityId) { + protected ListenableFuture> fetchArguments(CalculatedFieldCtx ctx, EntityId entityId, long ts) { Map> argFutures = switch (ctx.getCalculatedField().getType()) { - case GEOFENCING -> fetchGeofencingCalculatedFieldArguments(ctx, entityId, false); + case GEOFENCING -> fetchGeofencingCalculatedFieldArguments(ctx, entityId, false, ts); case SIMPLE, SCRIPT, ALARM -> { Map> futures = new HashMap<>(); for (var entry : ctx.getArguments().entrySet()) { - var argEntityId = resolveEntityId(entityId, entry.getValue()); - var argValueFuture = fetchArgumentValue(ctx.getTenantId(), argEntityId, entry.getValue(), System.currentTimeMillis()); + var argEntityId = resolveEntityId(ctx.getTenantId(), entityId, entry.getValue()); + var argValueFuture = fetchArgumentValue(ctx.getTenantId(), argEntityId, entry.getValue(), ts); futures.put(entry.getKey(), argValueFuture); } yield futures; @@ -102,8 +107,14 @@ public abstract class AbstractCalculatedFieldProcessingService { MoreExecutors.directExecutor()); } - protected EntityId resolveEntityId(EntityId entityId, Argument argument) { - return argument.getRefEntityId() != null ? argument.getRefEntityId() : entityId; + protected EntityId resolveEntityId(TenantId tenantId, EntityId entityId, Argument argument) { + if (argument.getRefEntityId() != null) { + return argument.getRefEntityId(); + } + if (!argument.hasOwnerSource()) { + return entityId; + } + return resolveOwnerArgument(tenantId, entityId, argument); } protected Map resolveArgumentFutures(Map> argFutures) { @@ -123,18 +134,18 @@ public abstract class AbstractCalculatedFieldProcessingService { )); } - protected Map> fetchGeofencingCalculatedFieldArguments(CalculatedFieldCtx ctx, EntityId entityId, boolean dynamicArgumentsOnly) { + protected Map> fetchGeofencingCalculatedFieldArguments(CalculatedFieldCtx ctx, EntityId entityId, boolean dynamicArgumentsOnly, long startTs) { Map> argFutures = new HashMap<>(); Set> entries = ctx.getArguments().entrySet(); if (dynamicArgumentsOnly) { entries = entries.stream() - .filter(entry -> entry.getValue().hasDynamicSource()) + .filter(entry -> entry.getValue().hasRelationQuerySource()) .collect(Collectors.toSet()); } for (var entry : entries) { switch (entry.getKey()) { case ENTITY_ID_LATITUDE_ARGUMENT_KEY, ENTITY_ID_LONGITUDE_ARGUMENT_KEY -> - argFutures.put(entry.getKey(), fetchArgumentValue(ctx.getTenantId(), entityId, entry.getValue(), System.currentTimeMillis())); + argFutures.put(entry.getKey(), fetchArgumentValue(ctx.getTenantId(), entityId, entry.getValue(), startTs)); default -> { var resolvedEntityIdsFuture = resolveGeofencingEntityIds(ctx.getTenantId(), entityId, entry); argFutures.put(entry.getKey(), Futures.transformAsync(resolvedEntityIdsFuture, resolvedEntityIds -> @@ -155,6 +166,14 @@ public abstract class AbstractCalculatedFieldProcessingService { } var refDynamicSourceConfiguration = value.getRefDynamicSourceConfiguration(); return switch (refDynamicSourceConfiguration.getType()) { + case CURRENT_CUSTOMER -> { + EntityId resolved = resolveOwnerArgument(tenantId, entityId, value); + if (resolved != null) { + yield Futures.immediateFuture(List.of(resolved)); + } else { + yield Futures.immediateFuture(Collections.emptyList()); + } + } case RELATION_QUERY -> { var configuration = (RelationQueryDynamicSourceConfiguration) refDynamicSourceConfiguration; if (configuration.isSimpleRelation()) { @@ -170,7 +189,23 @@ public abstract class AbstractCalculatedFieldProcessingService { yield Futures.transform(relationService.findByQuery(tenantId, configuration.toEntityRelationsQuery(entityId)), configuration::resolveEntityIds, calculatedFieldCallbackExecutor); } - case CURRENT_CUSTOMER -> throw new UnsupportedOperationException(); // fixme implement + }; + } + + @Nullable + private EntityId resolveOwnerArgument(TenantId tenantId, EntityId entityId, Argument argument) { + return switch (argument.getRefDynamicSourceConfiguration().getType()) { + case CURRENT_CUSTOMER -> { + EntityId ownerId = ownerService.getOwner(tenantId, entityId); + if (ownerId.getEntityType() == EntityType.TENANT) { + // todo: if inherit is true - use customer id + // fixme: WTF do we need it at all? + yield null; + } else { + yield ownerId; + } + } + default -> throw new UnsupportedOperationException(); }; } @@ -187,8 +222,7 @@ public abstract class AbstractCalculatedFieldProcessingService { argument.getRefEntityKey().getKey() ); return Futures.transform(attributesFuture, resultOpt -> - Map.entry(entityId, resultOpt.orElseGet(() -> - new BaseAttributeKvEntry(createDefaultKvEntry(argument), System.currentTimeMillis(), 0L))), + Map.entry(entityId, resultOpt.orElseGet(() -> createDefaultAttributeEntry(argument, System.currentTimeMillis()))), calculatedFieldCallbackExecutor ); }).collect(Collectors.toList()); @@ -200,6 +234,9 @@ public abstract class AbstractCalculatedFieldProcessingService { } protected ListenableFuture fetchArgumentValue(TenantId tenantId, EntityId entityId, Argument argument, long startTs) { + if (entityId == null) { + return Futures.immediateFuture(transformSingleValueArgument(Optional.empty())); + } return switch (argument.getRefEntityKey().getType()) { case TS_ROLLING -> fetchTsRolling(tenantId, entityId, argument, startTs); case ATTRIBUTE -> fetchAttribute(tenantId, entityId, argument, startTs); diff --git a/application/src/main/java/org/thingsboard/server/service/cf/CalculatedFieldCache.java b/application/src/main/java/org/thingsboard/server/service/cf/CalculatedFieldCache.java index 8dd3491942..cc77913f4b 100644 --- a/application/src/main/java/org/thingsboard/server/service/cf/CalculatedFieldCache.java +++ b/application/src/main/java/org/thingsboard/server/service/cf/CalculatedFieldCache.java @@ -23,6 +23,7 @@ import org.thingsboard.server.common.data.id.TenantId; import org.thingsboard.server.service.cf.ctx.state.CalculatedFieldCtx; import java.util.List; +import java.util.Set; import java.util.function.Predicate; public interface CalculatedFieldCache { @@ -47,4 +48,14 @@ public interface CalculatedFieldCache { EntityId getProfileId(TenantId tenantId, EntityId entityId); + Set getDynamicEntities(TenantId tenantId, EntityId entityId); + + void updateOwnerEntity(TenantId tenantId, EntityId entityId); + + void addOwnerEntity(TenantId tenantId, EntityId entityId); + + void evictEntity(EntityId entityId); + + void evictOwner(EntityId owner); + } diff --git a/application/src/main/java/org/thingsboard/server/service/cf/DefaultCalculatedFieldCache.java b/application/src/main/java/org/thingsboard/server/service/cf/DefaultCalculatedFieldCache.java index 0ef62c3568..36210d7302 100644 --- a/application/src/main/java/org/thingsboard/server/service/cf/DefaultCalculatedFieldCache.java +++ b/application/src/main/java/org/thingsboard/server/service/cf/DefaultCalculatedFieldCache.java @@ -23,6 +23,7 @@ import org.springframework.context.annotation.Lazy; import org.springframework.stereotype.Service; import org.springframework.util.ConcurrentReferenceHashMap; import org.thingsboard.server.actors.ActorSystemContext; +import org.thingsboard.server.common.data.EntityType; import org.thingsboard.server.common.data.cf.CalculatedField; import org.thingsboard.server.common.data.cf.CalculatedFieldLink; import org.thingsboard.server.common.data.cf.configuration.CalculatedFieldConfiguration; @@ -40,6 +41,7 @@ import org.thingsboard.server.service.profile.TbDeviceProfileCache; import java.util.Collections; import java.util.List; +import java.util.Set; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentMap; import java.util.concurrent.CopyOnWriteArrayList; @@ -59,6 +61,7 @@ public class DefaultCalculatedFieldCache implements CalculatedFieldCache { private final TbDeviceProfileCache deviceProfileCache; @Lazy private final ActorSystemContext systemContext; + private final OwnerService ownerService; private final ConcurrentMap calculatedFields = new ConcurrentHashMap<>(); private final ConcurrentMap> entityIdCalculatedFields = new ConcurrentHashMap<>(); @@ -66,6 +69,8 @@ public class DefaultCalculatedFieldCache implements CalculatedFieldCache { private final ConcurrentMap> entityIdCalculatedFieldLinks = new ConcurrentHashMap<>(); private final ConcurrentMap calculatedFieldsCtx = new ConcurrentHashMap<>(); + private final ConcurrentMap> ownerEntities = new ConcurrentHashMap<>(); + @Value("${queue.calculated_fields.init_fetch_pack_size:50000}") @Getter private int initFetchPackSize; @@ -220,6 +225,44 @@ public class DefaultCalculatedFieldCache implements CalculatedFieldCache { }; } + @Override + public Set getDynamicEntities(TenantId tenantId, EntityId entityId) { + if (entityId != null && entityId.getEntityType().isOneOf(EntityType.CUSTOMER, EntityType.TENANT)) { + return getOwnedEntities(tenantId, entityId); + } + return Collections.emptySet(); + } + + @Override + public void addOwnerEntity(TenantId tenantId, EntityId entityId) { + EntityId owner = ownerService.getOwner(tenantId, entityId); + getOwnedEntities(tenantId, owner).add(entityId); + } + + @Override + public void updateOwnerEntity(TenantId tenantId, EntityId entityId) { + evictEntity(entityId); + addOwnerEntity(tenantId, entityId); + } + + @Override + public void evictEntity(EntityId entityId) { + ownerEntities.values().forEach(entities -> entities.remove(entityId)); + } + + @Override + public void evictOwner(EntityId owner) { + ownerEntities.remove(owner); + } + + private Set getOwnedEntities(TenantId tenantId, EntityId ownerId) { + return ownerEntities.computeIfAbsent(ownerId, owner -> { + Set entities = ConcurrentHashMap.newKeySet(); + entities.addAll(ownerService.getOwnedEntities(tenantId, ownerId)); + return entities; + }); + } + private Lock getFetchLock(CalculatedFieldId id) { return calculatedFieldFetchLocks.computeIfAbsent(id, __ -> new ReentrantLock()); } diff --git a/application/src/main/java/org/thingsboard/server/service/cf/DefaultCalculatedFieldProcessingService.java b/application/src/main/java/org/thingsboard/server/service/cf/DefaultCalculatedFieldProcessingService.java index f5c39ed288..9b2964a736 100644 --- a/application/src/main/java/org/thingsboard/server/service/cf/DefaultCalculatedFieldProcessingService.java +++ b/application/src/main/java/org/thingsboard/server/service/cf/DefaultCalculatedFieldProcessingService.java @@ -69,9 +69,10 @@ public class DefaultCalculatedFieldProcessingService extends AbstractCalculatedF TimeseriesService timeseriesService, ApiLimitService apiLimitService, RelationService relationService, + OwnerService ownerService, TbClusterService clusterService, PartitionService partitionService) { - super(attributesService, timeseriesService, apiLimitService, relationService); + super(attributesService, timeseriesService, apiLimitService, relationService, ownerService); this.clusterService = clusterService; this.partitionService = partitionService; } @@ -83,26 +84,26 @@ public class DefaultCalculatedFieldProcessingService extends AbstractCalculatedF @Override public ListenableFuture> fetchArguments(CalculatedFieldCtx ctx, EntityId entityId) { - return super.fetchArguments(ctx, entityId); + return super.fetchArguments(ctx, entityId, System.currentTimeMillis()); } @Override public Map fetchDynamicArgsFromDb(CalculatedFieldCtx ctx, EntityId entityId) { - // only geofencing calculated fields supports dynamic arguments scheduled updates + // only scheduledSupported CF instances supports dynamic arguments scheduled updates if (!ctx.getCalculatedField().getType().equals(CalculatedFieldType.GEOFENCING)) { return Map.of(); } - return resolveArgumentFutures(fetchGeofencingCalculatedFieldArguments(ctx, entityId, true)); + return resolveArgumentFutures(fetchGeofencingCalculatedFieldArguments(ctx, entityId, true, System.currentTimeMillis())); } @Override public Map fetchArgsFromDb(TenantId tenantId, EntityId entityId, Map arguments) { Map> argFutures = new HashMap<>(); for (var entry : arguments.entrySet()) { - if (entry.getValue().hasDynamicSource()) { + if (entry.getValue().hasRelationQuerySource()) { continue; } - var argEntityId = resolveEntityId(entityId, entry.getValue()); + var argEntityId = resolveEntityId(tenantId, entityId, entry.getValue()); var argValueFuture = fetchArgumentValue(tenantId, argEntityId, entry.getValue(), System.currentTimeMillis()); argFutures.put(entry.getKey(), argValueFuture); } diff --git a/application/src/main/java/org/thingsboard/server/service/cf/DefaultCalculatedFieldQueueService.java b/application/src/main/java/org/thingsboard/server/service/cf/DefaultCalculatedFieldQueueService.java index a3e50812fa..ab06349e3d 100644 --- a/application/src/main/java/org/thingsboard/server/service/cf/DefaultCalculatedFieldQueueService.java +++ b/application/src/main/java/org/thingsboard/server/service/cf/DefaultCalculatedFieldQueueService.java @@ -85,6 +85,7 @@ public class DefaultCalculatedFieldQueueService implements CalculatedFieldQueueS checkEntityAndPushToQueue(tenantId, entityId, cf -> cf.matches(entries), cf -> cf.linkMatches(entityId, entries), + cf -> cf.dynamicSourceMatches(request.getEntries()), () -> toCalculatedFieldTelemetryMsgProto(request, result), callback); } @@ -102,6 +103,7 @@ public class DefaultCalculatedFieldQueueService implements CalculatedFieldQueueS checkEntityAndPushToQueue(tenantId, entityId, cf -> cf.matches(entries, scope), cf -> cf.linkMatches(entityId, entries, scope), + cf -> cf.dynamicSourceMatches(request.getEntries(), request.getScope()), () -> toCalculatedFieldTelemetryMsgProto(request, result), callback); } @@ -118,6 +120,7 @@ public class DefaultCalculatedFieldQueueService implements CalculatedFieldQueueS checkEntityAndPushToQueue(tenantId, entityId, cf -> cf.matchesKeys(result, scope), cf -> cf.linkMatchesAttrKeys(entityId, result, scope), + cf -> cf.matchesDynamicSourceKeys(result, request.getScope()), () -> toCalculatedFieldTelemetryMsgProto(request, result), callback); } @@ -128,16 +131,19 @@ public class DefaultCalculatedFieldQueueService implements CalculatedFieldQueueS checkEntityAndPushToQueue(tenantId, entityId, cf -> cf.matchesKeys(result), cf -> cf.linkMatchesTsKeys(entityId, result), + cf -> cf.matchesDynamicSourceKeys(result), () -> toCalculatedFieldTelemetryMsgProto(request, result), callback); } private void checkEntityAndPushToQueue(TenantId tenantId, EntityId entityId, - Predicate mainEntityFilter, Predicate linkedEntityFilter, + Predicate mainEntityFilter, + Predicate linkedEntityFilter, + Predicate dynamicSourceFilter, Supplier msg, FutureCallback callback) { if (EntityType.TENANT.equals(entityId.getEntityType())) { tenantId = (TenantId) entityId; } - boolean send = checkEntityForCalculatedFields(tenantId, entityId, mainEntityFilter, linkedEntityFilter); + boolean send = checkEntityForCalculatedFields(tenantId, entityId, mainEntityFilter, linkedEntityFilter, dynamicSourceFilter); if (send) { ToCalculatedFieldMsg calculatedFieldMsg = msg.get(); clusterService.pushMsgToCalculatedFields(tenantId, entityId, calculatedFieldMsg, wrap(callback)); @@ -148,7 +154,7 @@ public class DefaultCalculatedFieldQueueService implements CalculatedFieldQueueS } } - private boolean checkEntityForCalculatedFields(TenantId tenantId, EntityId entityId, Predicate filter, Predicate linkedEntityFilter) { + private boolean checkEntityForCalculatedFields(TenantId tenantId, EntityId entityId, Predicate filter, Predicate linkedEntityFilter, Predicate dynamicSourceFilter) { if (!supportedReferencedEntities.contains(entityId.getEntityType())) { return false; } @@ -165,6 +171,16 @@ public class DefaultCalculatedFieldQueueService implements CalculatedFieldQueueS } } + for (EntityId dynamicEntity : calculatedFieldCache.getDynamicEntities(tenantId, entityId)) { + if (calculatedFieldCache.getCalculatedFieldCtxsByEntityId(dynamicEntity).stream().anyMatch(dynamicSourceFilter)) { + return true; + } + EntityId dynamicEntityProfileId = calculatedFieldCache.getProfileId(tenantId, dynamicEntity); + if (calculatedFieldCache.getCalculatedFieldCtxsByEntityId(dynamicEntityProfileId).stream().anyMatch(dynamicSourceFilter)) { + return true; + } + } + return false; } diff --git a/application/src/main/java/org/thingsboard/server/service/cf/OwnerService.java b/application/src/main/java/org/thingsboard/server/service/cf/OwnerService.java new file mode 100644 index 0000000000..84c46715e7 --- /dev/null +++ b/application/src/main/java/org/thingsboard/server/service/cf/OwnerService.java @@ -0,0 +1,67 @@ +/** + * Copyright © 2016-2025 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.cf; + +import lombok.RequiredArgsConstructor; +import org.springframework.stereotype.Service; +import org.thingsboard.server.common.data.DeviceInfo; +import org.thingsboard.server.common.data.DeviceInfoFilter; +import org.thingsboard.server.common.data.EntityType; +import org.thingsboard.server.common.data.asset.Asset; +import org.thingsboard.server.common.data.id.AssetId; +import org.thingsboard.server.common.data.id.CustomerId; +import org.thingsboard.server.common.data.id.DeviceId; +import org.thingsboard.server.common.data.id.EntityId; +import org.thingsboard.server.common.data.id.TenantId; +import org.thingsboard.server.common.data.page.PageDataIterable; +import org.thingsboard.server.dao.asset.AssetService; +import org.thingsboard.server.dao.device.DeviceService; + +import java.util.HashSet; +import java.util.Set; + +@Service +@RequiredArgsConstructor +public class OwnerService { + + private final DeviceService deviceService; + private final AssetService assetService; + + public EntityId getOwner(TenantId tenantId, EntityId entityId) { + return switch (entityId.getEntityType()) { + case DEVICE -> deviceService.findDeviceById(tenantId, (DeviceId) entityId).getOwnerId(); + case ASSET -> assetService.findAssetById(tenantId, (AssetId) entityId).getOwnerId(); + default -> throw new UnsupportedOperationException(); + }; + } + + public Set getOwnedEntities(TenantId tenantId, EntityId ownerId) { + Set ownerEntities = new HashSet<>(); + if (EntityType.CUSTOMER.equals(ownerId.getEntityType())) { + PageDataIterable deviceIdInfos = new PageDataIterable<>(pageLink -> deviceService.findDeviceInfosByFilter(DeviceInfoFilter.builder().tenantId(tenantId).customerId((CustomerId) ownerId).build(), pageLink), 1000); + deviceIdInfos.forEach(deviceInfo -> ownerEntities.add(deviceInfo.getId())); + PageDataIterable assets = new PageDataIterable<>(pageLink -> assetService.findAssetsByTenantIdAndCustomerId(tenantId, (CustomerId) ownerId, pageLink), 1000); + assets.forEach(asset -> ownerEntities.add(asset.getId())); + } else if (EntityType.TENANT.equals(ownerId.getEntityType())) { + PageDataIterable deviceIdInfos = new PageDataIterable<>(pageLink -> deviceService.findDeviceInfosByFilter(DeviceInfoFilter.builder().tenantId((TenantId) ownerId).customerId(new CustomerId(CustomerId.NULL_UUID)).build(), pageLink), 1000); + deviceIdInfos.forEach(deviceInfo -> ownerEntities.add(deviceInfo.getId())); + PageDataIterable assets = new PageDataIterable<>(pageLink -> assetService.findAssetsByTenantIdAndCustomerId((TenantId) ownerId, new CustomerId(CustomerId.NULL_UUID), pageLink), 1000); + assets.forEach(asset -> ownerEntities.add(asset.getId())); + } + return ownerEntities; + } + +} diff --git a/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/CalculatedFieldCtx.java b/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/CalculatedFieldCtx.java index 13f5c8e6c7..5c288173de 100644 --- a/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/CalculatedFieldCtx.java +++ b/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/CalculatedFieldCtx.java @@ -73,6 +73,7 @@ public class CalculatedFieldCtx { private final Map arguments; private final Map mainEntityArguments; private final Map> linkedEntityArguments; + private final Map dynamicEntityArguments; private final List argNames; private Output output; private String expression; @@ -93,7 +94,7 @@ public class CalculatedFieldCtx { private long maxSingleValueArgumentSize; private List mainEntityGeofencingArgumentNames; - private List linkedEntityGeofencingArgumentNames; + private List linkedEntityAndCurrentOwnerGeofencingArgumentNames; public CalculatedFieldCtx(CalculatedField calculatedField, ActorSystemContext systemContext) { @@ -106,19 +107,27 @@ public class CalculatedFieldCtx { this.arguments = new HashMap<>(); this.mainEntityArguments = new HashMap<>(); this.linkedEntityArguments = new HashMap<>(); + this.dynamicEntityArguments = new HashMap<>(); this.argNames = new ArrayList<>(); this.mainEntityGeofencingArgumentNames = new ArrayList<>(); - this.linkedEntityGeofencingArgumentNames = new ArrayList<>(); + this.linkedEntityAndCurrentOwnerGeofencingArgumentNames = new ArrayList<>(); this.output = calculatedField.getConfiguration().getOutput(); if (calculatedField.getConfiguration() instanceof ArgumentsBasedCalculatedFieldConfiguration argBasedConfig) { this.arguments.putAll(argBasedConfig.getArguments()); for (Map.Entry entry : arguments.entrySet()) { var refId = entry.getValue().getRefEntityId(); var refKey = entry.getValue().getRefEntityKey(); - if (refId == null && entry.getValue().hasDynamicSource()) { - continue; - } - if (refId == null || refId.equals(calculatedField.getEntityId())) { + if (refId == null) { + // TODO: no matchers for this type of source exists yet, so no reason to add to dynamicEntityArguments map. + if (entry.getValue().hasRelationQuerySource()) { + continue; + } + if (entry.getValue().hasOwnerSource()) { + dynamicEntityArguments.put(refKey, entry.getKey()); + } else { + mainEntityArguments.put(refKey, entry.getKey()); + } + } else if (refId.equals(calculatedField.getEntityId())) { mainEntityArguments.put(refKey, entry.getKey()); } else { linkedEntityArguments.computeIfAbsent(refId, key -> new HashMap<>()).put(refKey, entry.getKey()); @@ -135,8 +144,8 @@ public class CalculatedFieldCtx { mainEntityGeofencingArgumentNames.add(zoneGroupName); return; } - if (config.isLinkedCfEntitySource(entityId)) { - linkedEntityGeofencingArgumentNames.add(zoneGroupName); + if (config.isLinkedCfEntitySource(entityId) || config.hasCurrentOwnerSource()) { + linkedEntityAndCurrentOwnerGeofencingArgumentNames.add(zoneGroupName); } }); } @@ -292,6 +301,14 @@ public class CalculatedFieldCtx { return map != null && matchesTimeSeries(map, values); } + public boolean dynamicSourceMatches(List values) { + return matchesTimeSeries(dynamicEntityArguments, values); + } + + public boolean dynamicSourceMatches(List values, AttributeScope scope) { + return matchesAttributes(dynamicEntityArguments, values, scope); + } + private boolean matchesAttributes(Map argMap, List values, AttributeScope scope) { if (argMap.isEmpty() || values.isEmpty()) { return false; @@ -335,6 +352,14 @@ public class CalculatedFieldCtx { return matchesTimeSeriesKeys(mainEntityArguments, keys); } + public boolean matchesDynamicSourceKeys(List keys, AttributeScope scope) { + return matchesAttributesKeys(dynamicEntityArguments, keys, scope); + } + + public boolean matchesDynamicSourceKeys(List keys) { + return matchesTimeSeriesKeys(dynamicEntityArguments, keys); + } + private boolean matchesAttributesKeys(Map argMap, List keys, AttributeScope scope) { if (argMap.isEmpty() || keys.isEmpty()) { return false; @@ -381,6 +406,25 @@ public class CalculatedFieldCtx { return map != null && matchesTimeSeriesKeys(map, keys); } + public boolean dynamicSourceMatches(CalculatedFieldTelemetryMsgProto proto) { + if (!proto.getTsDataList().isEmpty()) { + List updatedTelemetry = proto.getTsDataList().stream() + .map(ProtoUtils::fromProto) + .toList(); + return dynamicSourceMatches(updatedTelemetry); + } else if (!proto.getAttrDataList().isEmpty()) { + AttributeScope scope = AttributeScope.valueOf(proto.getScope().name()); + List updatedTelemetry = proto.getAttrDataList().stream() + .map(ProtoUtils::fromProto) + .toList(); + return dynamicSourceMatches(updatedTelemetry, scope); + } else if (!proto.getRemovedTsKeysList().isEmpty()) { + return matchesDynamicSourceKeys(proto.getRemovedTsKeysList()); + } else { + return matchesDynamicSourceKeys(proto.getRemovedAttrKeysList(), AttributeScope.valueOf(proto.getScope().name())); + } + } + public boolean linkMatches(EntityId entityId, CalculatedFieldTelemetryMsgProto proto) { if (!proto.getTsDataList().isEmpty()) { List updatedTelemetry = proto.getTsDataList().stream() @@ -400,6 +444,18 @@ public class CalculatedFieldCtx { } } + public Map getLinkedAndDynamicArgs(EntityId entityId) { + var argNames = new HashMap(); + var linkedArgNames = linkedEntityArguments.get(entityId); + if (linkedArgNames != null && !linkedArgNames.isEmpty()) { + argNames.putAll(linkedArgNames); + } + if (dynamicEntityArguments != null && !dynamicEntityArguments.isEmpty()) { + argNames.putAll(dynamicEntityArguments); + } + return argNames; + } + public CalculatedFieldEntityCtxId toCalculatedFieldEntityCtxId() { return new CalculatedFieldEntityCtxId(tenantId, cfId, entityId); } @@ -457,6 +513,10 @@ public class CalculatedFieldCtx { return "Failed to init CF state. State size exceeds limit of " + (maxStateSize / 1024) + "Kb!"; } + public boolean hasCurrentOwnerSourceArguments() { + return !dynamicEntityArguments.isEmpty(); + } + @Override public String toString() { return "CalculatedFieldCtx{" + diff --git a/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbClusterService.java b/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbClusterService.java index 265f14c4e2..418cf2c362 100644 --- a/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbClusterService.java +++ b/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbClusterService.java @@ -668,6 +668,7 @@ public class DefaultTbClusterService implements TbClusterService { } msg.event(ComponentLifecycleEvent.UPDATED) .oldProfileId(old.getDeviceProfileId()) + .ownerChanged(!entity.getOwnerId().equals(old.getOwnerId())) .oldName(old.getName()); } broadcast(msg.build()); @@ -688,6 +689,7 @@ public class DefaultTbClusterService implements TbClusterService { } else { msg.event(ComponentLifecycleEvent.UPDATED) .oldProfileId(old.getAssetProfileId()) + .ownerChanged(!entity.getOwnerId().equals(old.getOwnerId())) .oldName(old.getName()); } broadcast(msg.build()); diff --git a/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbEdgeConsumerService.java b/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbEdgeConsumerService.java index 3dd6993962..b7d561f005 100644 --- a/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbEdgeConsumerService.java +++ b/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbEdgeConsumerService.java @@ -87,8 +87,7 @@ public class DefaultTbEdgeConsumerService extends AbstractConsumerService new SimpleCalculatedFieldState(entityId); diff --git a/application/src/test/java/org/thingsboard/server/cf/AlarmRulesTest.java b/application/src/test/java/org/thingsboard/server/cf/AlarmRulesTest.java index 1bfb2bf875..581fad9a27 100644 --- a/application/src/test/java/org/thingsboard/server/cf/AlarmRulesTest.java +++ b/application/src/test/java/org/thingsboard/server/cf/AlarmRulesTest.java @@ -25,6 +25,7 @@ import org.springframework.test.context.bean.override.mockito.MockitoSpyBean; import org.thingsboard.common.util.JacksonUtil; import org.thingsboard.rule.engine.action.TbAlarmResult; import org.thingsboard.server.actors.ActorSystemContext; +import org.thingsboard.server.common.data.AttributeScope; import org.thingsboard.server.common.data.Device; import org.thingsboard.server.common.data.alarm.Alarm; import org.thingsboard.server.common.data.alarm.AlarmSeverity; @@ -40,6 +41,7 @@ import org.thingsboard.server.common.data.cf.CalculatedFieldType; import org.thingsboard.server.common.data.cf.configuration.AlarmCalculatedFieldConfiguration; import org.thingsboard.server.common.data.cf.configuration.Argument; import org.thingsboard.server.common.data.cf.configuration.ArgumentType; +import org.thingsboard.server.common.data.cf.configuration.CurrentCustomerDynamicSourceConfiguration; import org.thingsboard.server.common.data.cf.configuration.ReferencedEntityKey; import org.thingsboard.server.common.data.debug.DebugSettings; import org.thingsboard.server.common.data.event.CalculatedFieldDebugEvent; @@ -74,13 +76,14 @@ public class AlarmRulesTest extends AbstractControllerTest { @Autowired private EventDao eventDao; + private Device device; private DeviceId deviceId; private EventId latestEventId; @Before public void beforeEach() throws Exception { loginTenantAdmin(); - Device device = createDevice("Device A", "aaa"); + device = createDevice("Device A", "aaa"); deviceId = device.getId(); } @@ -207,6 +210,40 @@ public class AlarmRulesTest extends AbstractControllerTest { }); } + @Test + public void testCreateAlarm_currentOwnerArgument() throws Exception { + Argument temperatureArgument = new Argument(); + temperatureArgument.setRefEntityKey(new ReferencedEntityKey("temperature", ArgumentType.TS_LATEST, null)); + temperatureArgument.setDefaultValue("0"); + + Argument temperatureThresholdArgument = new Argument(); + temperatureThresholdArgument.setRefEntityKey(new ReferencedEntityKey("temperatureThreshold", ArgumentType.ATTRIBUTE, AttributeScope.SERVER_SCOPE)); + temperatureThresholdArgument.setRefDynamicSourceConfiguration(new CurrentCustomerDynamicSourceConfiguration()); + temperatureThresholdArgument.setDefaultValue("1000"); + + Map arguments = Map.of( + "temperature", temperatureArgument, + "temperatureThreshold", temperatureThresholdArgument + ); + + Map createRules = Map.of( + AlarmSeverity.CRITICAL, new Condition("return temperature >= temperatureThreshold;", null, null) + ); + + device.setCustomerId(customerId); + device = doPost("/api/device", device, Device.class); + CalculatedField calculatedField = createAlarmCf(deviceId, "High Temperature Alarm", + arguments, createRules, null); + postAttributes(customerId, AttributeScope.SERVER_SCOPE, "{\"temperatureThreshold\":50}"); + + postTelemetry(deviceId, "{\"temperature\":51}"); + checkAlarmResult(calculatedField, alarmResult -> { + assertThat(alarmResult.isCreated()).isTrue(); + assertThat(alarmResult.getAlarm().getSeverity()).isEqualTo(AlarmSeverity.CRITICAL); + assertThat(alarmResult.getAlarm().getStatus()).isEqualTo(AlarmStatus.ACTIVE_UNACK); + }); + } + private void checkAlarmResult(CalculatedField calculatedField, Consumer assertion) { await().atMost(TIMEOUT, TimeUnit.SECONDS).untilAsserted(() -> { TbAlarmResult alarmResult = getLatestAlarmResult(calculatedField.getId()); diff --git a/application/src/test/java/org/thingsboard/server/cf/CalculatedFieldIntegrationTest.java b/application/src/test/java/org/thingsboard/server/cf/CalculatedFieldIntegrationTest.java index 28e9d5bb1a..6db510192a 100644 --- a/application/src/test/java/org/thingsboard/server/cf/CalculatedFieldIntegrationTest.java +++ b/application/src/test/java/org/thingsboard/server/cf/CalculatedFieldIntegrationTest.java @@ -19,7 +19,6 @@ import com.fasterxml.jackson.databind.JsonNode; import com.fasterxml.jackson.databind.node.ArrayNode; import com.fasterxml.jackson.databind.node.ObjectNode; import org.junit.Test; -import org.junit.jupiter.api.BeforeEach; import org.thingsboard.common.util.JacksonUtil; import org.thingsboard.server.common.data.AttributeScope; import org.thingsboard.server.common.data.DataConstants; @@ -67,11 +66,6 @@ public class CalculatedFieldIntegrationTest extends CalculatedFieldControllerTes public static final int TIMEOUT = 60; public static final int POLL_INTERVAL = 1; - @BeforeEach - void setUp() throws Exception { - loginTenantAdmin(); - } - @Test public void testSimpleCalculatedFieldWhenAllTelemetryPresent() throws Exception { Device testDevice = createDevice("Test device", "1234567890"); diff --git a/application/src/test/java/org/thingsboard/server/controller/AbstractWebTest.java b/application/src/test/java/org/thingsboard/server/controller/AbstractWebTest.java index 8e7d2bb5ff..d2ac4741f4 100644 --- a/application/src/test/java/org/thingsboard/server/controller/AbstractWebTest.java +++ b/application/src/test/java/org/thingsboard/server/controller/AbstractWebTest.java @@ -77,6 +77,7 @@ import org.thingsboard.server.actors.device.DeviceActorMessageProcessor; import org.thingsboard.server.actors.device.SessionInfo; import org.thingsboard.server.actors.device.ToDeviceRpcRequestMetadata; import org.thingsboard.server.actors.service.DefaultActorService; +import org.thingsboard.server.common.data.AttributeScope; import org.thingsboard.server.common.data.Customer; import org.thingsboard.server.common.data.DataConstants; import org.thingsboard.server.common.data.Device; @@ -1317,8 +1318,13 @@ public abstract class AbstractWebTest extends AbstractInMemoryStorageTest { } protected void postTelemetry(EntityId entityId, String payload) throws Exception { - doPost("/api/plugins/telemetry/" + entityId.getEntityType() + "/" + entityId.getId() + - "/timeseries/" + DataConstants.SERVER_SCOPE, JacksonUtil.toJsonNode(payload)).andExpect(status().isOk()); + doPostAsync("/api/plugins/telemetry/" + entityId.getEntityType() + "/" + entityId.getId() + + "/timeseries/" + DataConstants.SERVER_SCOPE, JacksonUtil.toJsonNode(payload), 30_000L).andExpect(status().isOk()); + } + + protected void postAttributes(EntityId entityId, AttributeScope scope, String payload) throws Exception { + doPostAsync("/api/plugins/telemetry/" + entityId.getEntityType() + "/" + entityId.getId() + + "/attributes/" + scope, JacksonUtil.toJsonNode(payload), 30_000L).andExpect(status().isOk()); } protected CalculatedField saveCalculatedField(CalculatedField calculatedField) { diff --git a/common/data/src/main/java/org/thingsboard/server/common/data/Device.java b/common/data/src/main/java/org/thingsboard/server/common/data/Device.java index 57a0d24466..e170dbc467 100644 --- a/common/data/src/main/java/org/thingsboard/server/common/data/Device.java +++ b/common/data/src/main/java/org/thingsboard/server/common/data/Device.java @@ -28,6 +28,7 @@ import org.thingsboard.server.common.data.device.data.DeviceData; import org.thingsboard.server.common.data.id.CustomerId; import org.thingsboard.server.common.data.id.DeviceId; import org.thingsboard.server.common.data.id.DeviceProfileId; +import org.thingsboard.server.common.data.id.EntityId; import org.thingsboard.server.common.data.id.OtaPackageId; import org.thingsboard.server.common.data.id.TenantId; import org.thingsboard.server.common.data.validation.Length; @@ -142,6 +143,11 @@ public class Device extends BaseDataWithAdditionalInfo implements HasL this.customerId = customerId; } + @JsonIgnore + public EntityId getOwnerId() { + return customerId != null && !customerId.isNullUid() ? customerId : tenantId; + } + @Schema(requiredMode = Schema.RequiredMode.REQUIRED, description = "Unique Device Name in scope of Tenant", example = "A4B72CCDFF33") @Override public String getName() { diff --git a/common/data/src/main/java/org/thingsboard/server/common/data/ProfileEntityIdInfo.java b/common/data/src/main/java/org/thingsboard/server/common/data/ProfileEntityIdInfo.java index 22934de813..1fb49a46f7 100644 --- a/common/data/src/main/java/org/thingsboard/server/common/data/ProfileEntityIdInfo.java +++ b/common/data/src/main/java/org/thingsboard/server/common/data/ProfileEntityIdInfo.java @@ -34,21 +34,23 @@ public class ProfileEntityIdInfo implements Serializable, HasTenantId { private static final long serialVersionUID = 8532058281983868003L; private final TenantId tenantId; + private final EntityId ownerId; private final EntityId profileId; private final EntityId entityId; - private ProfileEntityIdInfo(UUID tenantId, EntityId profileId, EntityId entityId) { + private ProfileEntityIdInfo(UUID tenantId, EntityId ownerId, EntityId profileId, EntityId entityId) { this.tenantId = TenantId.fromUUID(tenantId); + this.ownerId = ownerId; this.profileId = profileId; this.entityId = entityId; } - public static ProfileEntityIdInfo create(UUID tenantId, DeviceProfileId profileId, DeviceId entityId) { - return new ProfileEntityIdInfo(tenantId, profileId, entityId); + public static ProfileEntityIdInfo create(UUID tenantId, EntityId ownerId, DeviceProfileId profileId, DeviceId entityId) { + return new ProfileEntityIdInfo(tenantId, ownerId, profileId, entityId); } - public static ProfileEntityIdInfo create(UUID tenantId, AssetProfileId profileId, AssetId entityId) { - return new ProfileEntityIdInfo(tenantId, profileId, entityId); + public static ProfileEntityIdInfo create(UUID tenantId, EntityId ownerId, AssetProfileId profileId, AssetId entityId) { + return new ProfileEntityIdInfo(tenantId, ownerId, profileId, entityId); } } diff --git a/common/data/src/main/java/org/thingsboard/server/common/data/asset/Asset.java b/common/data/src/main/java/org/thingsboard/server/common/data/asset/Asset.java index e732049118..a34b58a4da 100644 --- a/common/data/src/main/java/org/thingsboard/server/common/data/asset/Asset.java +++ b/common/data/src/main/java/org/thingsboard/server/common/data/asset/Asset.java @@ -15,6 +15,7 @@ */ package org.thingsboard.server.common.data.asset; +import com.fasterxml.jackson.annotation.JsonIgnore; import com.fasterxml.jackson.databind.JsonNode; import io.swagger.v3.oas.annotations.media.Schema; import lombok.EqualsAndHashCode; @@ -29,6 +30,7 @@ import org.thingsboard.server.common.data.HasVersion; import org.thingsboard.server.common.data.id.AssetId; import org.thingsboard.server.common.data.id.AssetProfileId; import org.thingsboard.server.common.data.id.CustomerId; +import org.thingsboard.server.common.data.id.EntityId; import org.thingsboard.server.common.data.id.TenantId; import org.thingsboard.server.common.data.validation.Length; import org.thingsboard.server.common.data.validation.NoXss; @@ -125,6 +127,11 @@ public class Asset extends BaseDataWithAdditionalInfo implements HasLab this.customerId = customerId; } + @JsonIgnore + public EntityId getOwnerId() { + return customerId != null && !customerId.isNullUid() ? customerId : tenantId; + } + @Schema(requiredMode = Schema.RequiredMode.REQUIRED, description = "Unique Asset Name in scope of Tenant", example = "Empire State Building") @Override public String getName() { diff --git a/common/data/src/main/java/org/thingsboard/server/common/data/cf/configuration/Argument.java b/common/data/src/main/java/org/thingsboard/server/common/data/cf/configuration/Argument.java index 52935c3411..0aad0737b8 100644 --- a/common/data/src/main/java/org/thingsboard/server/common/data/cf/configuration/Argument.java +++ b/common/data/src/main/java/org/thingsboard/server/common/data/cf/configuration/Argument.java @@ -37,4 +37,12 @@ public class Argument { return refDynamicSourceConfiguration != null; } + public boolean hasRelationQuerySource() { + return hasDynamicSource() && CFArgumentDynamicSourceType.RELATION_QUERY.equals(refDynamicSourceConfiguration.getType()); + } + + public boolean hasOwnerSource() { + return hasDynamicSource() && refDynamicSourceConfiguration.getType() == CFArgumentDynamicSourceType.CURRENT_CUSTOMER; + } + } diff --git a/common/data/src/main/java/org/thingsboard/server/common/data/cf/configuration/BaseCalculatedFieldConfiguration.java b/common/data/src/main/java/org/thingsboard/server/common/data/cf/configuration/BaseCalculatedFieldConfiguration.java index 535febf3a0..b72cdad60a 100644 --- a/common/data/src/main/java/org/thingsboard/server/common/data/cf/configuration/BaseCalculatedFieldConfiguration.java +++ b/common/data/src/main/java/org/thingsboard/server/common/data/cf/configuration/BaseCalculatedFieldConfiguration.java @@ -31,8 +31,8 @@ public abstract class BaseCalculatedFieldConfiguration implements ExpressionBase if (arguments.containsKey("ctx")) { throw new IllegalArgumentException("Argument name 'ctx' is reserved and cannot be used."); } - if (arguments.values().stream().anyMatch(Argument::hasDynamicSource)) { - throw new IllegalArgumentException("Calculated field with type: '" + getType() + "' doesn't support dynamic source configuration!"); + if (arguments.values().stream().anyMatch(Argument::hasRelationQuerySource)) { + throw new IllegalArgumentException("Calculated field with type: '" + getType() + "' doesn't support relation query source configuration!"); } } diff --git a/common/data/src/main/java/org/thingsboard/server/common/data/cf/configuration/geofencing/GeofencingCalculatedFieldConfiguration.java b/common/data/src/main/java/org/thingsboard/server/common/data/cf/configuration/geofencing/GeofencingCalculatedFieldConfiguration.java index dc331f5876..ff251a3ad3 100644 --- a/common/data/src/main/java/org/thingsboard/server/common/data/cf/configuration/geofencing/GeofencingCalculatedFieldConfiguration.java +++ b/common/data/src/main/java/org/thingsboard/server/common/data/cf/configuration/geofencing/GeofencingCalculatedFieldConfiguration.java @@ -63,7 +63,7 @@ public class GeofencingCalculatedFieldConfiguration implements ArgumentsBasedCal @Override public boolean isScheduledUpdateEnabled() { - return scheduledUpdateInterval > 0 && zoneGroups.values().stream().anyMatch(ZoneGroupConfiguration::hasDynamicSource); + return scheduledUpdateInterval > 0 && zoneGroups.values().stream().anyMatch(ZoneGroupConfiguration::hasRelationQuerySource); } @Override diff --git a/common/data/src/main/java/org/thingsboard/server/common/data/cf/configuration/geofencing/ZoneGroupConfiguration.java b/common/data/src/main/java/org/thingsboard/server/common/data/cf/configuration/geofencing/ZoneGroupConfiguration.java index 2feb6e49d0..775f711a5e 100644 --- a/common/data/src/main/java/org/thingsboard/server/common/data/cf/configuration/geofencing/ZoneGroupConfiguration.java +++ b/common/data/src/main/java/org/thingsboard/server/common/data/cf/configuration/geofencing/ZoneGroupConfiguration.java @@ -54,7 +54,7 @@ public class ZoneGroupConfiguration { if (reportStrategy == null) { throw new IllegalArgumentException("Report strategy must be specified for '" + name + "' zone group!"); } - if (hasDynamicSource()) { + if (refDynamicSourceConfiguration != null) { refDynamicSourceConfiguration.validate(); } if (!createRelationsWithMatchedZones) { @@ -68,8 +68,12 @@ public class ZoneGroupConfiguration { } } - public boolean hasDynamicSource() { - return refDynamicSourceConfiguration != null; + public boolean hasRelationQuerySource() { + return toArgument().hasRelationQuerySource(); + } + + public boolean hasCurrentOwnerSource() { + return toArgument().hasOwnerSource(); } @JsonIgnore @@ -92,4 +96,5 @@ public class ZoneGroupConfiguration { argument.setRefEntityKey(new ReferencedEntityKey(perimeterKeyName, ArgumentType.ATTRIBUTE, AttributeScope.SERVER_SCOPE)); return argument; } + } diff --git a/common/data/src/test/java/org/thingsboard/server/common/data/cf/configuration/ArgumentTest.java b/common/data/src/test/java/org/thingsboard/server/common/data/cf/configuration/ArgumentTest.java index fd59317649..260a39a8bc 100644 --- a/common/data/src/test/java/org/thingsboard/server/common/data/cf/configuration/ArgumentTest.java +++ b/common/data/src/test/java/org/thingsboard/server/common/data/cf/configuration/ArgumentTest.java @@ -29,10 +29,21 @@ public class ArgumentTest { } @Test - void validateShouldReturnTrueIfDynamicSourceConfigurationIsNotNull() { + void validateWhenRelationQuerySourceConfigurationIsNotNull() { var argument = new Argument(); argument.setRefDynamicSourceConfiguration(new RelationQueryDynamicSourceConfiguration()); assertThat(argument.hasDynamicSource()).isTrue(); + assertThat(argument.hasRelationQuerySource()).isTrue(); + assertThat(argument.hasOwnerSource()).isFalse(); + } + + @Test + void validateWhenCurrentCustomerSourceConfigurationIsNotNull() { + var argument = new Argument(); + argument.setRefDynamicSourceConfiguration(new CurrentCustomerDynamicSourceConfiguration()); + assertThat(argument.hasDynamicSource()).isTrue(); + assertThat(argument.hasOwnerSource()).isTrue(); + assertThat(argument.hasRelationQuerySource()).isFalse(); } } diff --git a/common/data/src/test/java/org/thingsboard/server/common/data/cf/configuration/geofencing/GeofencingCalculatedFieldConfigurationTest.java b/common/data/src/test/java/org/thingsboard/server/common/data/cf/configuration/geofencing/GeofencingCalculatedFieldConfigurationTest.java index 91a47aac57..9f2f49bc20 100644 --- a/common/data/src/test/java/org/thingsboard/server/common/data/cf/configuration/geofencing/GeofencingCalculatedFieldConfigurationTest.java +++ b/common/data/src/test/java/org/thingsboard/server/common/data/cf/configuration/geofencing/GeofencingCalculatedFieldConfigurationTest.java @@ -24,14 +24,12 @@ import org.thingsboard.server.common.data.cf.configuration.Argument; import org.thingsboard.server.common.data.cf.configuration.ArgumentType; import org.thingsboard.server.common.data.cf.configuration.ReferencedEntityKey; -import java.util.List; import java.util.Map; import static org.assertj.core.api.Assertions.assertThat; import static org.assertj.core.api.Assertions.assertThatCode; import static org.assertj.core.api.Assertions.assertThatThrownBy; import static org.mockito.Mockito.mock; -import static org.mockito.Mockito.never; import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; import static org.thingsboard.server.common.data.cf.configuration.geofencing.EntityCoordinates.ENTITY_ID_LATITUDE_ARGUMENT_KEY; @@ -112,7 +110,7 @@ public class GeofencingCalculatedFieldConfigurationTest { void scheduledUpdateDisabledWhenIntervalIsGreaterThanZeroButNoZonesWithDynamicArguments() { var cfg = new GeofencingCalculatedFieldConfiguration(); var zoneGroupConfigurationMock = mock(ZoneGroupConfiguration.class); - when(zoneGroupConfigurationMock.hasDynamicSource()).thenReturn(false); + when(zoneGroupConfigurationMock.hasRelationQuerySource()).thenReturn(false); cfg.setZoneGroups(Map.of("someGroupName", zoneGroupConfigurationMock)); cfg.setScheduledUpdateInterval(60); assertThat(cfg.isScheduledUpdateEnabled()).isFalse(); @@ -122,7 +120,7 @@ public class GeofencingCalculatedFieldConfigurationTest { void scheduledUpdateEnabledWhenIntervalIsGreaterThanZeroAndDynamicArgumentsPresent() { var cfg = new GeofencingCalculatedFieldConfiguration(); var zoneGroupConfigurationMock = mock(ZoneGroupConfiguration.class); - when(zoneGroupConfigurationMock.hasDynamicSource()).thenReturn(true); + when(zoneGroupConfigurationMock.hasRelationQuerySource()).thenReturn(true); cfg.setZoneGroups(Map.of("someGroupName", zoneGroupConfigurationMock)); cfg.setScheduledUpdateInterval(60); assertThat(cfg.isScheduledUpdateEnabled()).isTrue(); diff --git a/common/data/src/test/java/org/thingsboard/server/common/data/cf/configuration/geofencing/ZoneGroupConfigurationTest.java b/common/data/src/test/java/org/thingsboard/server/common/data/cf/configuration/geofencing/ZoneGroupConfigurationTest.java index 4eb822d93c..7bb657fb33 100644 --- a/common/data/src/test/java/org/thingsboard/server/common/data/cf/configuration/geofencing/ZoneGroupConfigurationTest.java +++ b/common/data/src/test/java/org/thingsboard/server/common/data/cf/configuration/geofencing/ZoneGroupConfigurationTest.java @@ -22,6 +22,7 @@ import org.junit.jupiter.params.provider.ValueSource; import org.thingsboard.server.common.data.AttributeScope; import org.thingsboard.server.common.data.cf.configuration.Argument; import org.thingsboard.server.common.data.cf.configuration.ArgumentType; +import org.thingsboard.server.common.data.cf.configuration.CurrentCustomerDynamicSourceConfiguration; import org.thingsboard.server.common.data.cf.configuration.ReferencedEntityKey; import org.thingsboard.server.common.data.cf.configuration.RelationQueryDynamicSourceConfiguration; import org.thingsboard.server.common.data.relation.EntityRelation; @@ -98,19 +99,25 @@ public class ZoneGroupConfigurationTest { } @Test - void whenHasDynamicSourceCalled_shouldReturnTrueIfDynamicSourceConfigurationIsNotNull() { + void whenHasRelationQuerySourceCalled_shouldReturnTrueIfRelationQuerySourceConfigurationIsNotNull() { var zoneGroupConfiguration = new ZoneGroupConfiguration("perimeter", REPORT_TRANSITION_EVENTS_AND_PRESENCE_STATUS, false); zoneGroupConfiguration.setRefDynamicSourceConfiguration(new RelationQueryDynamicSourceConfiguration()); - assertThat(zoneGroupConfiguration.hasDynamicSource()).isTrue(); + assertThat(zoneGroupConfiguration.hasRelationQuerySource()).isTrue(); } @Test - void whenHasDynamicSourceCalled_shouldReturnTrueIfDynamicSourceConfigurationIsNull() { + void whenHasRelationQuerySourceCalled_shouldReturnFalseIfRelationQuerySourceConfigurationIsNull() { var zoneGroupConfiguration = mock(ZoneGroupConfiguration.class); assertThat(zoneGroupConfiguration.getRefDynamicSourceConfiguration()).isNull(); - assertThat(zoneGroupConfiguration.hasDynamicSource()).isFalse(); + assertThat(zoneGroupConfiguration.hasRelationQuerySource()).isFalse(); } + @Test + void whenHasRelationQuerySourceCalled_shouldReturnFalseIfCurrentCustomerSourceConfigured() { + var zoneGroupConfiguration = mock(ZoneGroupConfiguration.class); + zoneGroupConfiguration.setRefDynamicSourceConfiguration(new CurrentCustomerDynamicSourceConfiguration()); + assertThat(zoneGroupConfiguration.hasRelationQuerySource()).isFalse(); + } @Test void validateToArgumentsMethodCallWithoutRefEntityId() { diff --git a/common/message/src/main/java/org/thingsboard/server/common/msg/MsgType.java b/common/message/src/main/java/org/thingsboard/server/common/msg/MsgType.java index fca3632ee8..6451ea0a6f 100644 --- a/common/message/src/main/java/org/thingsboard/server/common/msg/MsgType.java +++ b/common/message/src/main/java/org/thingsboard/server/common/msg/MsgType.java @@ -151,6 +151,7 @@ public enum MsgType { CF_ENTITY_INIT_CF_MSG, CF_ENTITY_DELETE_MSG, + CF_ARGUMENT_RESET_MSG, // Sent to reset argument; CF_DYNAMIC_ARGUMENTS_REFRESH_MSG, CF_ENTITY_DYNAMIC_ARGUMENTS_REFRESH_MSG, CF_REEVALUATE_MSG; diff --git a/common/message/src/main/java/org/thingsboard/server/common/msg/plugin/ComponentLifecycleMsg.java b/common/message/src/main/java/org/thingsboard/server/common/msg/plugin/ComponentLifecycleMsg.java index d57301fd10..23b9fe08e3 100644 --- a/common/message/src/main/java/org/thingsboard/server/common/msg/plugin/ComponentLifecycleMsg.java +++ b/common/message/src/main/java/org/thingsboard/server/common/msg/plugin/ComponentLifecycleMsg.java @@ -46,14 +46,15 @@ public class ComponentLifecycleMsg implements TenantAwareMsg, ToAllNodesMsg { private final String name; private final EntityId oldProfileId; private final EntityId profileId; + private final boolean ownerChanged; private final JsonNode info; public ComponentLifecycleMsg(TenantId tenantId, EntityId entityId, ComponentLifecycleEvent event) { - this(tenantId, entityId, event, null, null, null, null, null); + this(tenantId, entityId, event, null, null, null, null, false, null); } @Builder - private ComponentLifecycleMsg(TenantId tenantId, EntityId entityId, ComponentLifecycleEvent event, String oldName, String name, EntityId oldProfileId, EntityId profileId, JsonNode info) { + private ComponentLifecycleMsg(TenantId tenantId, EntityId entityId, ComponentLifecycleEvent event, String oldName, String name, EntityId oldProfileId, EntityId profileId, boolean ownerChanged, JsonNode info) { this.tenantId = tenantId; this.entityId = entityId; this.event = event; @@ -61,6 +62,7 @@ public class ComponentLifecycleMsg implements TenantAwareMsg, ToAllNodesMsg { this.name = name; this.oldProfileId = oldProfileId; this.profileId = profileId; + this.ownerChanged = ownerChanged; this.info = info; } diff --git a/common/proto/src/main/java/org/thingsboard/server/common/util/ProtoUtils.java b/common/proto/src/main/java/org/thingsboard/server/common/util/ProtoUtils.java index f5a07cf07e..26a64c7f8a 100644 --- a/common/proto/src/main/java/org/thingsboard/server/common/util/ProtoUtils.java +++ b/common/proto/src/main/java/org/thingsboard/server/common/util/ProtoUtils.java @@ -129,6 +129,7 @@ public class ProtoUtils { builder.setOldProfileIdMSB(msg.getOldProfileId().getId().getMostSignificantBits()); builder.setOldProfileIdLSB(msg.getOldProfileId().getId().getLeastSignificantBits()); } + builder.setOwnerChanged(msg.isOwnerChanged()); if (msg.getName() != null) { builder.setName(msg.getName()); } @@ -165,6 +166,7 @@ public class ProtoUtils { var profileType = EntityType.DEVICE.equals(entityId.getEntityType()) ? EntityType.DEVICE_PROFILE : EntityType.ASSET_PROFILE; builder.oldProfileId(EntityIdFactory.getByTypeAndUuid(profileType, new UUID(proto.getOldProfileIdMSB(), proto.getOldProfileIdLSB()))); } + builder.ownerChanged(proto.getOwnerChanged()); if (proto.hasInfo()) { builder.info(JacksonUtil.toJsonNode(proto.getInfo())); } diff --git a/common/proto/src/main/proto/queue.proto b/common/proto/src/main/proto/queue.proto index 5a8e348f5f..fac1116a30 100644 --- a/common/proto/src/main/proto/queue.proto +++ b/common/proto/src/main/proto/queue.proto @@ -1291,6 +1291,7 @@ message ComponentLifecycleMsgProto { int64 profileIdMSB = 11; int64 profileIdLSB = 12; optional string info = 13; + bool ownerChanged = 100; } message EdgeEventMsgProto { diff --git a/dao/src/main/java/org/thingsboard/server/dao/service/validator/CalculatedFieldDataValidator.java b/dao/src/main/java/org/thingsboard/server/dao/service/validator/CalculatedFieldDataValidator.java index 05b782c26c..2c6e9160ae 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/service/validator/CalculatedFieldDataValidator.java +++ b/dao/src/main/java/org/thingsboard/server/dao/service/validator/CalculatedFieldDataValidator.java @@ -100,7 +100,7 @@ public class CalculatedFieldDataValidator extends DataValidator } Map relationQueryBasedArguments = argumentsBasedCfg.getArguments().entrySet() .stream() - .filter(entry -> entry.getValue().hasDynamicSource()) + .filter(entry -> entry.getValue().hasRelationQuerySource()) .collect(Collectors.toMap(Map.Entry::getKey, entry -> (RelationQueryDynamicSourceConfiguration) entry.getValue().getRefDynamicSourceConfiguration())); if (relationQueryBasedArguments.isEmpty()) { return; diff --git a/dao/src/main/java/org/thingsboard/server/dao/sql/device/DefaultNativeAssetRepository.java b/dao/src/main/java/org/thingsboard/server/dao/sql/device/DefaultNativeAssetRepository.java index ec47da3499..3057b0d30f 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/sql/device/DefaultNativeAssetRepository.java +++ b/dao/src/main/java/org/thingsboard/server/dao/sql/device/DefaultNativeAssetRepository.java @@ -23,9 +23,12 @@ import org.springframework.transaction.support.TransactionTemplate; import org.thingsboard.server.common.data.ProfileEntityIdInfo; import org.thingsboard.server.common.data.id.AssetId; import org.thingsboard.server.common.data.id.AssetProfileId; +import org.thingsboard.server.common.data.id.CustomerId; +import org.thingsboard.server.common.data.id.EntityId; import org.thingsboard.server.common.data.id.TenantId; import org.thingsboard.server.common.data.page.PageData; +import java.util.Map; import java.util.UUID; @Repository @@ -40,23 +43,24 @@ public class DefaultNativeAssetRepository extends AbstractNativeRepository imple @Override public PageData findProfileEntityIdInfos(Pageable pageable) { - String PROFILE_ASSET_ID_INFO_QUERY = "SELECT tenant_id as tenantId, asset_profile_id as profileId, id as id FROM asset ORDER BY created_time ASC LIMIT %s OFFSET %s"; - return find(COUNT_QUERY, PROFILE_ASSET_ID_INFO_QUERY, pageable, row -> { - AssetId id = new AssetId((UUID) row.get("id")); - AssetProfileId profileId = new AssetProfileId((UUID) row.get("profileId")); - var tenantIdObj = row.get("tenantId"); - return ProfileEntityIdInfo.create(tenantIdObj != null ? (UUID) tenantIdObj : TenantId.SYS_TENANT_ID.getId(), profileId, id); - }); + String PROFILE_ASSET_ID_INFO_QUERY = "SELECT tenant_id as tenantId, customer_id as customerId, asset_profile_id as profileId, id as id FROM asset ORDER BY created_time ASC LIMIT %s OFFSET %s"; + return find(COUNT_QUERY, PROFILE_ASSET_ID_INFO_QUERY, pageable, DefaultNativeAssetRepository::toInfo); } @Override public PageData findProfileEntityIdInfosByTenantId(UUID tenantId, Pageable pageable) { - String PROFILE_ASSET_ID_INFO_QUERY = String.format("SELECT tenant_id as tenantId, asset_profile_id as profileId, id as id FROM asset WHERE tenant_id = '%s' ORDER BY created_time ASC LIMIT %%s OFFSET %%s", tenantId); - return find(COUNT_QUERY, PROFILE_ASSET_ID_INFO_QUERY, pageable, row -> { - AssetId id = new AssetId((UUID) row.get("id")); - AssetProfileId profileId = new AssetProfileId((UUID) row.get("profileId")); - var tenantIdObj = row.get("tenantId"); - return ProfileEntityIdInfo.create(tenantIdObj != null ? (UUID) tenantIdObj : TenantId.SYS_TENANT_ID.getId(), profileId, id); - }); + String PROFILE_ASSET_ID_INFO_QUERY = String.format("SELECT tenant_id as tenantId, customer_id as customerId, asset_profile_id as profileId, id as id FROM asset WHERE tenant_id = '%s' ORDER BY created_time ASC LIMIT %%s OFFSET %%s", tenantId); + return find(COUNT_QUERY, PROFILE_ASSET_ID_INFO_QUERY, pageable, DefaultNativeAssetRepository::toInfo); } + + private static ProfileEntityIdInfo toInfo(Map row) { + var tenantIdObj = row.get("tenantId"); + UUID tenantId = tenantIdObj != null ? (UUID) tenantIdObj : TenantId.SYS_TENANT_ID.getId(); + AssetId id = new AssetId((UUID) row.get("id")); + CustomerId customerId = new CustomerId((UUID) row.get("customerId")); + EntityId ownerId = !customerId.isNullUid() ? customerId : TenantId.fromUUID(tenantId); + AssetProfileId profileId = new AssetProfileId((UUID) row.get("profileId")); + return ProfileEntityIdInfo.create(tenantId, ownerId, profileId, id); + } + } diff --git a/dao/src/main/java/org/thingsboard/server/dao/sql/device/DefaultNativeDeviceRepository.java b/dao/src/main/java/org/thingsboard/server/dao/sql/device/DefaultNativeDeviceRepository.java index 78ee2795b0..49062f829f 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/sql/device/DefaultNativeDeviceRepository.java +++ b/dao/src/main/java/org/thingsboard/server/dao/sql/device/DefaultNativeDeviceRepository.java @@ -22,11 +22,14 @@ import org.springframework.stereotype.Repository; import org.springframework.transaction.support.TransactionTemplate; import org.thingsboard.server.common.data.DeviceIdInfo; import org.thingsboard.server.common.data.ProfileEntityIdInfo; +import org.thingsboard.server.common.data.id.CustomerId; import org.thingsboard.server.common.data.id.DeviceId; import org.thingsboard.server.common.data.id.DeviceProfileId; +import org.thingsboard.server.common.data.id.EntityId; import org.thingsboard.server.common.data.id.TenantId; import org.thingsboard.server.common.data.page.PageData; +import java.util.Map; import java.util.UUID; @Repository @@ -52,24 +55,24 @@ public class DefaultNativeDeviceRepository extends AbstractNativeRepository impl @Override public PageData findProfileEntityIdInfos(Pageable pageable) { - String PROFILE_DEVICE_ID_INFO_QUERY = "SELECT tenant_id as tenantId, device_profile_id as profileId, id as id FROM device ORDER BY created_time ASC LIMIT %s OFFSET %s"; - return find(COUNT_QUERY, PROFILE_DEVICE_ID_INFO_QUERY, pageable, row -> { - DeviceId id = new DeviceId((UUID) row.get("id")); - DeviceProfileId profileId = new DeviceProfileId((UUID) row.get("profileId")); - var tenantIdObj = row.get("tenantId"); - return ProfileEntityIdInfo.create(tenantIdObj != null ? (UUID) tenantIdObj : TenantId.SYS_TENANT_ID.getId(), profileId, id); - }); + String PROFILE_DEVICE_ID_INFO_QUERY = "SELECT tenant_id as tenantId, customer_id as customerId, device_profile_id as profileId, id as id FROM device ORDER BY created_time ASC LIMIT %s OFFSET %s"; + return find(COUNT_QUERY, PROFILE_DEVICE_ID_INFO_QUERY, pageable, DefaultNativeDeviceRepository::toInfo); } @Override public PageData findProfileEntityIdInfosByTenantId(UUID tenantId, Pageable pageable) { - String PROFILE_DEVICE_ID_INFO_QUERY = String.format("SELECT tenant_id as tenantId, device_profile_id as profileId, id as id FROM device WHERE tenant_id = '%s' ORDER BY created_time ASC LIMIT %%s OFFSET %%s", tenantId); - return find(COUNT_QUERY, PROFILE_DEVICE_ID_INFO_QUERY, pageable, row -> { - DeviceId id = new DeviceId((UUID) row.get("id")); - DeviceProfileId profileId = new DeviceProfileId((UUID) row.get("profileId")); - var tenantIdObj = row.get("tenantId"); - return ProfileEntityIdInfo.create(tenantIdObj != null ? (UUID) tenantIdObj : TenantId.SYS_TENANT_ID.getId(), profileId, id); - }); + String PROFILE_DEVICE_ID_INFO_QUERY = String.format("SELECT tenant_id as tenantId, customer_id as customerId, device_profile_id as profileId, id as id FROM device WHERE tenant_id = '%s' ORDER BY created_time ASC LIMIT %%s OFFSET %%s", tenantId); + return find(COUNT_QUERY, PROFILE_DEVICE_ID_INFO_QUERY, pageable, DefaultNativeDeviceRepository::toInfo); + } + + private static ProfileEntityIdInfo toInfo(Map row) { + var tenantIdObj = row.get("tenantId"); + UUID tenantId = tenantIdObj != null ? (UUID) tenantIdObj : TenantId.SYS_TENANT_ID.getId(); + DeviceId id = new DeviceId((UUID) row.get("id")); + CustomerId customerId = new CustomerId((UUID) row.get("customerId")); + EntityId ownerId = !customerId.isNullUid() ? customerId : TenantId.fromUUID(tenantId); + DeviceProfileId profileId = new DeviceProfileId((UUID) row.get("profileId")); + return ProfileEntityIdInfo.create(tenantId, ownerId, profileId, id); } }