diff --git a/application/src/main/data/upgrade/basic/schema_update.sql b/application/src/main/data/upgrade/basic/schema_update.sql index 3903bf1bf5..d01fa37312 100644 --- a/application/src/main/data/upgrade/basic/schema_update.sql +++ b/application/src/main/data/upgrade/basic/schema_update.sql @@ -45,7 +45,7 @@ SET profile_data = jsonb_set( CASE WHEN (profile_data -> 'configuration') ? 'minAllowedDeduplicationIntervalInSecForCF' THEN NULL - ELSE to_jsonb(3600) + ELSE to_jsonb(60) END, 'minAggregationIntervalInSecForCF', CASE 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 2050619212..89f5578760 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 @@ -415,6 +415,23 @@ public class CalculatedFieldEntityMessageProcessor extends AbstractContextAwareM } catch (Exception e) { throw CalculatedFieldException.builder().ctx(ctx).eventEntity(entityId).cause(e).build(); } + } else if (ctx.shouldFetchEntityRelations(state)) { + log.debug("[{}][{}] Going to update related entities for CF.", entityId, ctx.getCfId()); + try { + if (state instanceof RelatedEntitiesAggregationCalculatedFieldState relatedEntitiesState) { + List relatedEntities = cfService.fetchRelatedEntities(ctx, entityId); + List missingEntities = relatedEntitiesState.checkRelatedEntities(relatedEntities); + if (!missingEntities.isEmpty()) { + missingEntities.forEach(missingEntityId -> { + Map fetchedArgs = cfService.fetchArgsFromDb(tenantId, missingEntityId, ctx.getArguments()); + relatedEntitiesState.updateEntityData(setEntityIdToSingleEntityArguments(missingEntityId, fetchedArgs)); + }); + justRestored = true; + } + } + } catch (Exception e) { + throw CalculatedFieldException.builder().ctx(ctx).eventEntity(entityId).cause(e).build(); + } } if (state.isSizeOk()) { Map updatedArgs = state.update(newArgValues, ctx); 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 b4c7f3e997..a5b975d454 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 @@ -27,9 +27,11 @@ import org.thingsboard.server.actors.service.DefaultActorService; import org.thingsboard.server.actors.shared.AbstractContextAwareMsgProcessor; import org.thingsboard.server.common.data.Customer; import org.thingsboard.server.common.data.DataConstants; +import org.thingsboard.server.common.data.DeviceProfile; import org.thingsboard.server.common.data.EntityType; import org.thingsboard.server.common.data.ProfileEntityIdInfo; import org.thingsboard.server.common.data.alarm.Alarm; +import org.thingsboard.server.common.data.asset.AssetProfile; import org.thingsboard.server.common.data.audit.ActionType; import org.thingsboard.server.common.data.cf.CalculatedField; import org.thingsboard.server.common.data.cf.CalculatedFieldLink; @@ -74,6 +76,7 @@ import java.util.HashMap; import java.util.HashSet; import java.util.List; import java.util.Map; +import java.util.Optional; import java.util.Set; import java.util.concurrent.CopyOnWriteArrayList; import java.util.concurrent.ScheduledFuture; @@ -320,12 +323,16 @@ public class CalculatedFieldManagerMessageProcessor extends AbstractContextAware EntityId fromId = entityRelation.getFrom(); String relationType = entityRelation.getType(); + if (!(CalculatedField.isSupportedRefEntity(toId) || CalculatedField.isSupportedRefEntity(fromId))) { + callback.onSuccess(); + return; + } + MultipleTbCallback callbackForToAndFrom = new MultipleTbCallback(2, callback); processRelationByDirection(EntitySearchDirection.TO, relationType, toId, callbackForToAndFrom, relationAction.apply(fromId)); processRelationByDirection(EntitySearchDirection.FROM, relationType, fromId, callbackForToAndFrom, relationAction.apply(toId)); } - private void processRelationByDirection(EntitySearchDirection direction, String relationType, EntityId mainId, @@ -339,12 +346,11 @@ public class CalculatedFieldManagerMessageProcessor extends AbstractContextAware List matchingCfs = cfsByEntityIdAndProfile.stream() .filter(cf -> { - if (cf.getCalculatedField().getConfiguration() instanceof RelatedEntitiesAggregationCalculatedFieldConfiguration config ) { + if (cf.getCalculatedField().getConfiguration() instanceof RelatedEntitiesAggregationCalculatedFieldConfiguration config) { RelationPathLevel relation = config.getRelation(); return direction.equals(relation.direction()) && relationType.equals(relation.relationType()); - } else { - return false; } + return false; }) .toList(); @@ -717,8 +723,8 @@ public class CalculatedFieldManagerMessageProcessor extends AbstractContextAware private EntityId getProfileId(TenantId tenantId, EntityId entityId) { return switch (entityId.getEntityType()) { - case ASSET -> assetProfileCache.get(tenantId, (AssetId) entityId).getId(); - case DEVICE -> deviceProfileCache.get(tenantId, (DeviceId) entityId).getId(); + case ASSET -> Optional.ofNullable(assetProfileCache.get(tenantId, (AssetId) entityId)).map(AssetProfile::getId).orElse(null); + case DEVICE -> Optional.ofNullable(deviceProfileCache.get(tenantId, (DeviceId) entityId)).map(DeviceProfile::getId).orElse(null); default -> null; }; } 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 45d8e3b4e0..6914f6d10a 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 @@ -24,6 +24,7 @@ import jakarta.annotation.PreDestroy; import lombok.Data; import lombok.extern.slf4j.Slf4j; import org.thingsboard.common.util.ThingsBoardExecutors; +import org.thingsboard.server.common.data.cf.CalculatedField; 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.RelationPathQueryDynamicSourceConfiguration; @@ -62,6 +63,7 @@ import java.util.Optional; import java.util.Set; import java.util.concurrent.ExecutionException; import java.util.function.Function; +import java.util.function.Predicate; import java.util.stream.Collectors; import static org.thingsboard.server.common.data.cf.CalculatedFieldType.PROPAGATION; @@ -184,11 +186,12 @@ public abstract class AbstractCalculatedFieldProcessingService { } protected Map> fetchRelatedEntitiesAggArguments(CalculatedFieldCtx ctx, EntityId entityId, long ts) { - RelatedEntitiesAggregationCalculatedFieldConfiguration aggConfig = (RelatedEntitiesAggregationCalculatedFieldConfiguration) ctx.getCalculatedField().getConfiguration(); - - ListenableFuture> relatedEntitiesFut = resolveRelatedEntities(ctx.getTenantId(), entityId, aggConfig.getRelation()); + if (!(ctx.getCalculatedField().getConfiguration() instanceof RelatedEntitiesAggregationCalculatedFieldConfiguration config)) { + return Collections.emptyMap(); + } + ListenableFuture> relatedEntitiesFut = resolveRelatedEntities(ctx.getTenantId(), entityId, config.getRelation()); - return aggConfig.getArguments().entrySet().stream() + return config.getArguments().entrySet().stream() .collect(Collectors.toMap( Map.Entry::getKey, entry -> Futures.transformAsync(relatedEntitiesFut, relatedEntities -> fetchRelatedEntitiesArgumentEntry(ctx.getTenantId(), relatedEntities, entry.getValue(), ts), MoreExecutors.directExecutor()) @@ -205,8 +208,9 @@ public abstract class AbstractCalculatedFieldProcessingService { )); } - private ListenableFuture> resolveRelatedEntities(TenantId tenantId, EntityId entityId, RelationPathLevel relation) { - ListenableFuture> relationsFut = relationService.findByRelationPathQueryAsync(tenantId, new EntityRelationPathQuery(entityId, List.of(relation))); + protected ListenableFuture> resolveRelatedEntities(TenantId tenantId, EntityId entityId, RelationPathLevel relation) { + Predicate filter = entityRelation -> CalculatedField.isSupportedRefEntity(entityRelation.getFrom()) && CalculatedField.isSupportedRefEntity(entityRelation.getTo()); + ListenableFuture> relationsFut = relationService.findFilteredRelationsByPathQueryAsync(tenantId, new EntityRelationPathQuery(entityId, List.of(relation)), filter); return Futures.transform(relationsFut, relations -> { if (relations == null) { @@ -217,7 +221,11 @@ public abstract class AbstractCalculatedFieldProcessingService { case FROM -> relations.stream() .map(EntityRelation::getTo) .toList(); - case TO -> relations.isEmpty() ? List.of() : List.of(relations.get(0).getFrom()); + case TO -> relations.stream() + .map(EntityRelation::getFrom) + .findFirst() + .map(List::of) + .orElseGet(Collections::emptyList); }; }, calculatedFieldCallbackExecutor); } @@ -239,7 +247,8 @@ public abstract class AbstractCalculatedFieldProcessingService { case CURRENT_OWNER -> Futures.immediateFuture(List.of(resolveOwnerArgument(tenantId, entityId))); case RELATION_PATH_QUERY -> { var configuration = (RelationPathQueryDynamicSourceConfiguration) refDynamicSourceConfiguration; - yield Futures.transform(relationService.findByRelationPathQueryAsync(tenantId, configuration.toRelationPathQuery(entityId)), + Predicate filter = entityRelation -> CalculatedField.isSupportedRefEntity(entityRelation.getFrom()) && CalculatedField.isSupportedRefEntity(entityRelation.getTo()); + yield Futures.transform(relationService.findFilteredRelationsByPathQueryAsync(tenantId, configuration.toRelationPathQuery(entityId), filter), configuration::resolveEntityIds, calculatedFieldCallbackExecutor); } }; diff --git a/application/src/main/java/org/thingsboard/server/service/cf/CalculatedFieldProcessingService.java b/application/src/main/java/org/thingsboard/server/service/cf/CalculatedFieldProcessingService.java index 67bd2fec0c..804f94341b 100644 --- a/application/src/main/java/org/thingsboard/server/service/cf/CalculatedFieldProcessingService.java +++ b/application/src/main/java/org/thingsboard/server/service/cf/CalculatedFieldProcessingService.java @@ -37,6 +37,8 @@ public interface CalculatedFieldProcessingService { Map fetchDynamicArgsFromDb(CalculatedFieldCtx ctx, EntityId entityId); + List fetchRelatedEntities(CalculatedFieldCtx ctx, EntityId entityId); + Map fetchArgsFromDb(TenantId tenantId, EntityId entityId, Map arguments); ArgumentEntry fetchMetricDuringInterval(TenantId tenantId, EntityId entityId, String argKey, AggMetric metric, AggIntervalEntry interval); 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 c0114780b2..9033b21fd4 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 @@ -25,6 +25,7 @@ import org.thingsboard.server.common.data.DataConstants; import org.thingsboard.server.common.data.EntityType; import org.thingsboard.server.common.data.cf.configuration.Argument; import org.thingsboard.server.common.data.cf.configuration.aggregation.AggMetric; +import org.thingsboard.server.common.data.cf.configuration.aggregation.RelatedEntitiesAggregationCalculatedFieldConfiguration; import org.thingsboard.server.common.data.id.CalculatedFieldId; import org.thingsboard.server.common.data.id.EntityId; import org.thingsboard.server.common.data.id.TenantId; @@ -56,6 +57,7 @@ import java.util.HashMap; import java.util.List; import java.util.Map; import java.util.UUID; +import java.util.concurrent.ExecutionException; import static org.thingsboard.server.common.data.cf.configuration.PropagationCalculatedFieldConfiguration.PROPAGATION_CONFIG_ARGUMENT; import static org.thingsboard.server.utils.CalculatedFieldUtils.toProto; @@ -99,6 +101,19 @@ public class DefaultCalculatedFieldProcessingService extends AbstractCalculatedF }; } + @Override + public List fetchRelatedEntities(CalculatedFieldCtx ctx, EntityId entityId) { + try { + if (ctx.getCalculatedField().getConfiguration() instanceof RelatedEntitiesAggregationCalculatedFieldConfiguration config) { + return resolveRelatedEntities(ctx.getTenantId(), entityId, config.getRelation()).get(); + } + return Collections.emptyList(); + } catch (ExecutionException | InterruptedException e) { + Throwable cause = e.getCause(); + throw new RuntimeException("Failed to fetch related entities for entity [" + entityId + "]: " + cause.getMessage(), cause); + } + } + @Override public Map fetchArgsFromDb(TenantId tenantId, EntityId entityId, Map arguments) { Map> argFutures = new HashMap<>(); 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 09c6fc9c38..1bc1222d6b 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 @@ -162,7 +162,7 @@ public class DefaultCalculatedFieldQueueService implements CalculatedFieldQueueS } private boolean checkEntityForCalculatedFields(TenantId tenantId, EntityId entityId, Predicate filter, Predicate linkedEntityFilter, Predicate dynamicSourceFilter, Predicate relatedEntityFilter) { - if (!CalculatedField.SUPPORTED_REFERENCED_ENTITIES.contains(entityId.getEntityType())) { + if (!CalculatedField.isSupportedRefEntity(entityId)) { return false; } @@ -211,7 +211,6 @@ public class DefaultCalculatedFieldQueueService implements CalculatedFieldQueueS return true; } } - return false; } } } 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 cdf055fe7a..5c50f7e95e 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 @@ -61,6 +61,7 @@ import org.thingsboard.server.dao.relation.RelationService; import org.thingsboard.server.gen.transport.TransportProtos.CalculatedFieldTelemetryMsgProto; import org.thingsboard.server.service.cf.CalculatedFieldProcessingService; import org.thingsboard.server.service.cf.ctx.CalculatedFieldEntityCtxId; +import org.thingsboard.server.service.cf.ctx.state.aggregation.RelatedEntitiesAggregationCalculatedFieldState; import org.thingsboard.server.service.cf.ctx.state.geofencing.GeofencingCalculatedFieldState; import org.thingsboard.server.service.telemetry.AlarmSubscriptionService; @@ -703,6 +704,19 @@ public class CalculatedFieldCtx implements Closeable { }; } + public boolean shouldFetchEntityRelations(CalculatedFieldState state) { + if (!(state instanceof RelatedEntitiesAggregationCalculatedFieldState relatedEntitiesAggState)) { + return false; + } + if (!isScheduledUpdateEnabled()) { + return false; + } + if (relatedEntitiesAggState.getLastRelatedEntitiesRefreshTs() == -1L) { + return true; + } + return relatedEntitiesAggState.getLastRelatedEntitiesRefreshTs() < System.currentTimeMillis() - scheduledUpdateIntervalMillis; + } + @Override public void close() { try { diff --git a/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/aggregation/RelatedEntitiesAggregationCalculatedFieldState.java b/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/aggregation/RelatedEntitiesAggregationCalculatedFieldState.java index 89ca14c94b..ff50cd99b6 100644 --- a/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/aggregation/RelatedEntitiesAggregationCalculatedFieldState.java +++ b/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/aggregation/RelatedEntitiesAggregationCalculatedFieldState.java @@ -38,9 +38,12 @@ import org.thingsboard.server.service.cf.ctx.state.BaseCalculatedFieldState; import org.thingsboard.server.service.cf.ctx.state.CalculatedFieldCtx; import org.thingsboard.server.service.cf.ctx.state.aggregation.function.AggEntry; +import java.util.ArrayList; import java.util.HashMap; +import java.util.List; import java.util.Map; import java.util.Map.Entry; +import java.util.concurrent.ScheduledFuture; import static java.util.concurrent.TimeUnit.SECONDS; @@ -52,9 +55,13 @@ public class RelatedEntitiesAggregationCalculatedFieldState extends BaseCalculat private long lastArgsRefreshTs = -1; @Setter private long lastMetricsEvalTs = -1; + @Setter + private long lastRelatedEntitiesRefreshTs = -1; private long deduplicationIntervalMs = -1; private Map metrics; + private ScheduledFuture reevaluationFuture; + public RelatedEntitiesAggregationCalculatedFieldState(EntityId entityId) { super(entityId); } @@ -67,8 +74,13 @@ public class RelatedEntitiesAggregationCalculatedFieldState extends BaseCalculat deduplicationIntervalMs = SECONDS.toMillis(configuration.getDeduplicationIntervalInSec()); } - public void scheduleReevaluation() { - ctx.scheduleReevaluation(deduplicationIntervalMs, actorCtx); + @Override + public void close() { + super.close(); + if (reevaluationFuture != null) { + reevaluationFuture.cancel(true); + reevaluationFuture = null; + } } @Override @@ -76,9 +88,14 @@ public class RelatedEntitiesAggregationCalculatedFieldState extends BaseCalculat super.reset(); lastArgsRefreshTs = -1; lastMetricsEvalTs = -1; + lastRelatedEntitiesRefreshTs = -1; metrics = null; } + public void updateLastRelatedEntitiesRefreshTs() { + lastRelatedEntitiesRefreshTs = System.currentTimeMillis(); + } + @Override public CalculatedFieldType getType() { return CalculatedFieldType.RELATED_ENTITIES_AGGREGATION; @@ -90,6 +107,56 @@ public class RelatedEntitiesAggregationCalculatedFieldState extends BaseCalculat return super.update(argumentValues, ctx); } + public List checkRelatedEntities(List relatedEntities) { + Map> entityInputs = prepareInputs(); + findOutdatedEntities(entityInputs, relatedEntities).forEach(this::cleanupEntityData); + updateLastRelatedEntitiesRefreshTs(); + return findMissingEntities(entityInputs, relatedEntities); + } + + private List findMissingEntities(Map> entityInputs, List relatedEntities) { + List missing = new ArrayList<>(); + relatedEntities.forEach(entityId -> { + if (!entityInputs.containsKey(entityId)) { + missing.add(entityId); + log.warn("[{}] Missing related entity inputs for {}", ctx.getCfId(), entityId); + } + }); + return missing; + } + + private List findOutdatedEntities(Map> entityInputs, List relatedEntities) { + List outdated = new ArrayList<>(); + entityInputs.keySet().forEach(entityId -> { + if (!relatedEntities.contains(entityId)) { + outdated.add(entityId); + log.warn("[{}] CF state keeps outdated related entity {}", ctx.getCfId(), entityId); + } + }); + return outdated; + } + + public Map updateEntityData(Map fetchedArgs) { + lastMetricsEvalTs = -1; + return update(fetchedArgs, ctx); + } + + public void cleanupEntityData(EntityId relatedEntityId) { + arguments.values().forEach(argEntry -> { + RelatedEntitiesArgumentEntry aggEntry = (RelatedEntitiesArgumentEntry) argEntry; + aggEntry.getEntityInputs().remove(relatedEntityId); + }); + lastMetricsEvalTs = -1; + lastArgsRefreshTs = System.currentTimeMillis(); + } + + public void scheduleReevaluation() { + ScheduledFuture future = ctx.scheduleReevaluation(deduplicationIntervalMs, actorCtx); + if (future != null) { + reevaluationFuture = future; + } + } + @Override public ListenableFuture performCalculation(Map updatedArgs, CalculatedFieldCtx ctx) throws Exception { boolean cfUpdated = updatedArgs != null && updatedArgs.isEmpty(); @@ -97,7 +164,7 @@ public class RelatedEntitiesAggregationCalculatedFieldState extends BaseCalculat Output output = ctx.getOutput(); ObjectNode aggResult = aggregateMetrics(output); lastMetricsEvalTs = System.currentTimeMillis(); - ctx.scheduleReevaluation(deduplicationIntervalMs, actorCtx); + scheduleReevaluation(); return Futures.immediateFuture(TelemetryCalculatedFieldResult.builder() .type(output.getType()) .scope(output.getScope()) @@ -108,20 +175,6 @@ public class RelatedEntitiesAggregationCalculatedFieldState extends BaseCalculat } } - public Map updateEntityData(Map fetchedArgs) { - lastMetricsEvalTs = -1; - return update(fetchedArgs, ctx); - } - - public void cleanupEntityData(EntityId relatedEntityId) { - arguments.values().forEach(argEntry -> { - RelatedEntitiesArgumentEntry aggEntry = (RelatedEntitiesArgumentEntry) argEntry; - aggEntry.getEntityInputs().remove(relatedEntityId); - }); - lastMetricsEvalTs = -1; - lastArgsRefreshTs = System.currentTimeMillis(); - } - private boolean shouldRecalculate() { boolean intervalPassed = lastMetricsEvalTs <= System.currentTimeMillis() - deduplicationIntervalMs; boolean argsUpdatedDuringInterval = lastArgsRefreshTs > lastMetricsEvalTs; diff --git a/application/src/main/java/org/thingsboard/server/service/entitiy/EntityStateSourcingListener.java b/application/src/main/java/org/thingsboard/server/service/entitiy/EntityStateSourcingListener.java index 48a318f981..9a6a96155f 100644 --- a/application/src/main/java/org/thingsboard/server/service/entitiy/EntityStateSourcingListener.java +++ b/application/src/main/java/org/thingsboard/server/service/entitiy/EntityStateSourcingListener.java @@ -29,7 +29,6 @@ import org.thingsboard.server.common.data.Customer; import org.thingsboard.server.common.data.Device; import org.thingsboard.server.common.data.DeviceProfile; import org.thingsboard.server.common.data.EntityType; -import org.thingsboard.server.common.data.ObjectType; import org.thingsboard.server.common.data.TbResource; import org.thingsboard.server.common.data.TbResourceInfo; import org.thingsboard.server.common.data.Tenant; @@ -49,6 +48,7 @@ import org.thingsboard.server.common.data.job.Job; import org.thingsboard.server.common.data.msg.TbMsgType; import org.thingsboard.server.common.data.notification.NotificationRequest; import org.thingsboard.server.common.data.plugin.ComponentLifecycleEvent; +import org.thingsboard.server.common.data.relation.EntityRelation; import org.thingsboard.server.common.data.rule.RuleChain; import org.thingsboard.server.common.data.rule.RuleChainType; import org.thingsboard.server.common.data.security.DeviceCredentials; @@ -274,10 +274,13 @@ public class EntityStateSourcingListener { @TransactionalEventListener(fallbackExecution = true) public void handleEvent(RelationActionEvent relationEvent) { - if (relationEvent.getActionType() == ActionType.RELATION_ADD_OR_UPDATE) { - tbClusterService.onRelationUpdated(relationEvent.getTenantId(), relationEvent.getRelation(), TbQueueCallback.EMPTY); - } else if (relationEvent.getActionType() == ActionType.RELATION_DELETED) { - tbClusterService.onRelationDeleted(relationEvent.getTenantId(), relationEvent.getRelation(), TbQueueCallback.EMPTY); + EntityRelation relation = relationEvent.getRelation(); + if (CalculatedField.isSupportedRefEntity(relation.getFrom()) && CalculatedField.isSupportedRefEntity(relation.getTo())) { + if (relationEvent.getActionType() == ActionType.RELATION_ADD_OR_UPDATE) { + tbClusterService.onRelationUpdated(relationEvent.getTenantId(), relation, TbQueueCallback.EMPTY); + } else if (relationEvent.getActionType() == ActionType.RELATION_DELETED) { + tbClusterService.onRelationDeleted(relationEvent.getTenantId(), relation, TbQueueCallback.EMPTY); + } } } diff --git a/application/src/test/java/org/thingsboard/server/cf/RelatedEntitiesAggregationCalculatedFieldTest.java b/application/src/test/java/org/thingsboard/server/cf/RelatedEntitiesAggregationCalculatedFieldTest.java index a7b67baf3f..48c4e67608 100644 --- a/application/src/test/java/org/thingsboard/server/cf/RelatedEntitiesAggregationCalculatedFieldTest.java +++ b/application/src/test/java/org/thingsboard/server/cf/RelatedEntitiesAggregationCalculatedFieldTest.java @@ -51,6 +51,7 @@ import org.thingsboard.server.common.data.relation.EntityRelation; import org.thingsboard.server.common.data.relation.EntitySearchDirection; import org.thingsboard.server.common.data.relation.RelationPathLevel; import org.thingsboard.server.common.data.relation.RelationTypeGroup; +import org.thingsboard.server.common.data.rule.RuleChain; import org.thingsboard.server.common.data.security.Authority; import org.thingsboard.server.controller.AbstractControllerTest; import org.thingsboard.server.dao.service.DaoSqlTest; @@ -87,6 +88,7 @@ public class RelatedEntitiesAggregationCalculatedFieldTest extends AbstractContr updateDefaultTenantProfileConfig(tenantProfileConfig -> { tenantProfileConfig.setMinAllowedDeduplicationIntervalInSecForCF(1); + tenantProfileConfig.setMinAllowedScheduledUpdateIntervalInSecForCF(1); }); Tenant tenant = new Tenant(); @@ -177,7 +179,7 @@ public class RelatedEntitiesAggregationCalculatedFieldTest extends AbstractContr Asset asset2 = createAsset("Asset 2", assetProfile.getId()); - await().alias("add entity to profile with no related entities and perform aggregation").atMost(deduplicationInterval, TimeUnit.SECONDS) + await().alias("add entity to profile with no related entities and perform aggregation").atMost(deduplicationInterval * 2, TimeUnit.SECONDS) .pollInterval(POLL_INTERVAL, TimeUnit.SECONDS) .untilAsserted(() -> { ObjectNode occupancy = getLatestTelemetry(asset2.getId(), "freeSpaces", "occupiedSpaces", "totalSpaces"); @@ -190,7 +192,7 @@ public class RelatedEntitiesAggregationCalculatedFieldTest extends AbstractContr createEntityRelation(asset2.getId(), device3.getId(), "Contains"); createEntityRelation(asset2.getId(), device4.getId(), "Contains"); - await().alias("create relations and perform aggregation").atMost(deduplicationInterval, TimeUnit.SECONDS) + await().alias("create relations and perform aggregation").atMost(deduplicationInterval * 2, TimeUnit.SECONDS) .pollInterval(POLL_INTERVAL, TimeUnit.SECONDS) .untilAsserted(() -> { verifyTelemetry(asset2.getId(), Map.of( @@ -202,7 +204,7 @@ public class RelatedEntitiesAggregationCalculatedFieldTest extends AbstractContr postTelemetry(device3.getId(), "{\"occupied\":false}"); - await().alias("update telemetry and perform aggregation").atMost(deduplicationInterval, TimeUnit.SECONDS) + await().alias("update telemetry and perform aggregation").atMost(deduplicationInterval * 2, TimeUnit.SECONDS) .pollInterval(POLL_INTERVAL, TimeUnit.SECONDS) .untilAsserted(() -> { verifyTelemetry(asset2.getId(), Map.of( @@ -224,7 +226,7 @@ public class RelatedEntitiesAggregationCalculatedFieldTest extends AbstractContr createOccupancyCF(assetProfile.getId()); - await().alias("create CF and perform initial aggregation").atMost(deduplicationInterval, TimeUnit.SECONDS) + await().alias("create CF and perform initial aggregation").atMost(deduplicationInterval * 2, TimeUnit.SECONDS) .pollInterval(POLL_INTERVAL, TimeUnit.SECONDS) .untilAsserted(() -> { verifyTelemetry(asset.getId(), Map.of( @@ -246,7 +248,7 @@ public class RelatedEntitiesAggregationCalculatedFieldTest extends AbstractContr postTelemetry(device3.getId(), "{\"occupied\":true}"); - await().alias("change profile and no aggregation").atMost(deduplicationInterval, TimeUnit.SECONDS) + await().alias("change profile and no aggregation").atMost(deduplicationInterval * 2, TimeUnit.SECONDS) .pollInterval(POLL_INTERVAL, TimeUnit.SECONDS) .untilAsserted(() -> { verifyTelemetry(asset2.getId(), Map.of( @@ -268,7 +270,7 @@ public class RelatedEntitiesAggregationCalculatedFieldTest extends AbstractContr createOccupancyCF(asset2.getId()); - await().alias("create CF and perform aggregation with default values").atMost(deduplicationInterval, TimeUnit.SECONDS) + await().alias("create CF and perform aggregation with default values").atMost(deduplicationInterval * 2, TimeUnit.SECONDS) .pollInterval(POLL_INTERVAL, TimeUnit.SECONDS) .untilAsserted(() -> { verifyTelemetry(asset2.getId(), Map.of( @@ -299,6 +301,45 @@ public class RelatedEntitiesAggregationCalculatedFieldTest extends AbstractContr }); } + @Test + public void testCreateCfAndRelationToRuleChain_checkAggregation() throws Exception { + Asset asset2 = createAsset("Asset 2", assetProfile.getId()); + Device device3 = createDevice("Device 3", "1234567890333"); + postTelemetry(device3.getId(), "{\"occupied\":true}"); + + RuleChain ruleChain = new RuleChain(); + ruleChain.setName("RuleChain"); + ruleChain = doPost("/api/ruleChain", ruleChain, RuleChain.class); + postTelemetry(ruleChain.getId(), "{\"occupied\":true}"); + + createEntityRelation(asset2.getId(), device3.getId(), "Contains"); + createEntityRelation(asset2.getId(), ruleChain.getId(), "Contains"); + + createOccupancyCF(asset2.getId()); + + await().alias("create CF and perform initial aggregation").atMost(deduplicationInterval * 2, TimeUnit.SECONDS) + .pollInterval(POLL_INTERVAL, TimeUnit.SECONDS) + .untilAsserted(() -> { + verifyTelemetry(asset2.getId(), Map.of( + "freeSpaces", "0", + "occupiedSpaces", "1", + "totalSpaces", "1" + )); + }); + + postTelemetry(ruleChain.getId(), "{\"occupied\":true}"); + + await().alias("update telemetry on rule chain and no aggregation performed").atMost(deduplicationInterval * 2, TimeUnit.SECONDS) + .pollInterval(POLL_INTERVAL, TimeUnit.SECONDS) + .untilAsserted(() -> { + verifyTelemetry(asset2.getId(), Map.of( + "freeSpaces", "0", + "occupiedSpaces", "1", + "totalSpaces", "1" + )); + }); + } + @Test public void testDeleteCf_checkNoAggregation() throws Exception { CalculatedField cf = createOccupancyCF(asset.getId()); @@ -309,7 +350,7 @@ public class RelatedEntitiesAggregationCalculatedFieldTest extends AbstractContr postTelemetry(device1.getId(), "{\"occupied\":false}"); - await().alias("delete cf and update telemetry and no aggregation").atMost(deduplicationInterval, TimeUnit.SECONDS) + await().alias("delete cf and update telemetry and no aggregation").atMost(deduplicationInterval * 2, TimeUnit.SECONDS) .pollInterval(POLL_INTERVAL, TimeUnit.SECONDS) .untilAsserted(() -> { verifyTelemetry(asset.getId(), Map.of( @@ -364,7 +405,7 @@ public class RelatedEntitiesAggregationCalculatedFieldTest extends AbstractContr createOccupancyCF(asset2.getId()); - await().alias("create CF and perform aggregation").atMost(deduplicationInterval, TimeUnit.SECONDS) + await().alias("create CF and perform aggregation").atMost(deduplicationInterval * 2, TimeUnit.SECONDS) .pollInterval(POLL_INTERVAL, TimeUnit.SECONDS) .untilAsserted(() -> { verifyTelemetry(asset2.getId(), Map.of( @@ -402,7 +443,7 @@ public class RelatedEntitiesAggregationCalculatedFieldTest extends AbstractContr createOccupancyCFWithAttr(asset2.getId()); - await().alias("create CF and perform aggregation").atMost(deduplicationInterval, TimeUnit.SECONDS) + await().alias("create CF and perform aggregation").atMost(deduplicationInterval * 2, TimeUnit.SECONDS) .pollInterval(POLL_INTERVAL, TimeUnit.SECONDS) .untilAsserted(() -> { verifyTelemetry(asset2.getId(), Map.of( @@ -437,7 +478,7 @@ public class RelatedEntitiesAggregationCalculatedFieldTest extends AbstractContr createEntityRelation(asset.getId(), device3.getId(), "Contains"); - await().alias("create relation and perform aggregation").atMost(deduplicationInterval, TimeUnit.SECONDS) + await().alias("create relation and perform aggregation").atMost(deduplicationInterval * 2, TimeUnit.SECONDS) .pollInterval(POLL_INTERVAL, TimeUnit.SECONDS) .untilAsserted(() -> { verifyTelemetry(asset.getId(), Map.of( @@ -455,7 +496,25 @@ public class RelatedEntitiesAggregationCalculatedFieldTest extends AbstractContr deleteEntityRelation(new EntityRelation(asset.getId(), device1.getId(), "Contains", RelationTypeGroup.COMMON)); - await().alias("create relation and perform aggregation").atMost(deduplicationInterval, TimeUnit.SECONDS) + await().alias("create relation and perform aggregation").atMost(deduplicationInterval * 2, TimeUnit.SECONDS) + .pollInterval(POLL_INTERVAL, TimeUnit.SECONDS) + .untilAsserted(() -> { + verifyTelemetry(asset.getId(), Map.of( + "freeSpaces", "1", + "occupiedSpaces", "0", + "totalSpaces", "1" + )); + }); + } + + @Test + public void testDeleteEntityByRelation_checkAggregation() throws Exception { + createOccupancyCF(asset.getId()); + checkInitialCalculation(); + + doDelete("/api/device/" + device1.getId()).andExpect(status().isOk()); + + await().alias("create relation and perform aggregation").atMost(deduplicationInterval * 2, TimeUnit.SECONDS) .pollInterval(POLL_INTERVAL, TimeUnit.SECONDS) .untilAsserted(() -> { verifyTelemetry(asset.getId(), Map.of( @@ -479,7 +538,7 @@ public class RelatedEntitiesAggregationCalculatedFieldTest extends AbstractContr configuration.setRelation(new RelationPathLevel(EntitySearchDirection.FROM, "Has")); saveCalculatedField(cf); - await().alias("update relation path and perform aggregation").atMost(deduplicationInterval, TimeUnit.SECONDS) + await().alias("update relation path and perform aggregation").atMost(deduplicationInterval * 2, TimeUnit.SECONDS) .pollInterval(POLL_INTERVAL, TimeUnit.SECONDS) .untilAsserted(() -> { verifyTelemetry(asset.getId(), Map.of( @@ -505,7 +564,7 @@ public class RelatedEntitiesAggregationCalculatedFieldTest extends AbstractContr configuration.setArguments(Map.of("oc", argument)); saveCalculatedField(cf); - await().alias("update arguments and perform aggregation").atMost(deduplicationInterval, TimeUnit.SECONDS) + await().alias("update arguments and perform aggregation").atMost(deduplicationInterval * 2, TimeUnit.SECONDS) .pollInterval(POLL_INTERVAL, TimeUnit.SECONDS) .untilAsserted(() -> { verifyTelemetry(asset.getId(), Map.of( @@ -560,7 +619,7 @@ public class RelatedEntitiesAggregationCalculatedFieldTest extends AbstractContr postTelemetry(device2.getId(), "{\"temperature\":19.6}"); CalculatedField cf = createAvgTemperatureCF(asset.getId()); - await().alias("create avg temp cf and perform initial aggregation").atMost(deduplicationInterval, TimeUnit.SECONDS) + await().alias("create avg temp cf and perform initial aggregation").atMost(deduplicationInterval * 2, TimeUnit.SECONDS) .pollInterval(POLL_INTERVAL, TimeUnit.SECONDS) .untilAsserted(() -> { verifyTelemetry(asset.getId(), Map.of("avgTemperature", "24")); @@ -573,7 +632,7 @@ public class RelatedEntitiesAggregationCalculatedFieldTest extends AbstractContr configuration.setOutput(output); saveCalculatedField(cf); - await().alias("update output and perform aggregation").atMost(deduplicationInterval, TimeUnit.SECONDS) + await().alias("update output and perform aggregation").atMost(deduplicationInterval * 2, TimeUnit.SECONDS) .pollInterval(POLL_INTERVAL, TimeUnit.SECONDS) .untilAsserted(() -> { ArrayNode avgTemperature = getServerAttributes(asset.getId(), "avgTemperature"); @@ -589,7 +648,7 @@ public class RelatedEntitiesAggregationCalculatedFieldTest extends AbstractContr postTelemetry(device2.getId(), "{\"temperature\":19.6}"); CalculatedField cf = createAvgTemperatureCF(asset.getId()); - await().alias("create avg temp cf and perform initial aggregation").atMost(deduplicationInterval, TimeUnit.SECONDS) + await().alias("create avg temp cf and perform initial aggregation").atMost(deduplicationInterval * 2, TimeUnit.SECONDS) .pollInterval(POLL_INTERVAL, TimeUnit.SECONDS) .untilAsserted(() -> { verifyTelemetry(asset.getId(), Map.of("avgTemperature", "24")); @@ -607,7 +666,7 @@ public class RelatedEntitiesAggregationCalculatedFieldTest extends AbstractContr postTelemetry(device2.getId(), "{\"temperature\":32.1}"); - await().alias("update telemetry and perform aggregation").atMost(2 * deduplicationInterval, TimeUnit.SECONDS) + await().alias("update telemetry and perform aggregation").atMost(2 * deduplicationInterval + 10, TimeUnit.SECONDS) .pollInterval(POLL_INTERVAL, TimeUnit.SECONDS) .untilAsserted(() -> { verifyTelemetry(asset.getId(), Map.of("avgTemperature", "28")); @@ -615,7 +674,7 @@ public class RelatedEntitiesAggregationCalculatedFieldTest extends AbstractContr } private void checkInitialCalculation() { - await().alias("create CF and perform initial aggregation").atMost(deduplicationInterval, TimeUnit.SECONDS) + await().alias("create CF and perform initial aggregation").atMost(deduplicationInterval * 2, TimeUnit.SECONDS) .pollInterval(POLL_INTERVAL, TimeUnit.SECONDS) .untilAsserted(this::checkInitialCalculationValues); } @@ -743,6 +802,7 @@ public class RelatedEntitiesAggregationCalculatedFieldTest extends AbstractContr configuration.setRelation(relation); configuration.setArguments(inputs); configuration.setDeduplicationIntervalInSec(deduplicationInterval); + configuration.setScheduledUpdateInterval(10); configuration.setMetrics(metrics); configuration.setOutput(output); diff --git a/application/src/test/java/org/thingsboard/server/controller/AbstractNotifyEntityTest.java b/application/src/test/java/org/thingsboard/server/controller/AbstractNotifyEntityTest.java index 01c24ad493..d4eec0d648 100644 --- a/application/src/test/java/org/thingsboard/server/controller/AbstractNotifyEntityTest.java +++ b/application/src/test/java/org/thingsboard/server/controller/AbstractNotifyEntityTest.java @@ -128,11 +128,17 @@ public abstract class AbstractNotifyEntityTest extends AbstractWebTest { protected void testNotifyEntityAllOneTimeLogEntityActionEntityEqClass(HasName entity, EntityId entityId, EntityId originatorId, TenantId tenantId, CustomerId customerId, UserId userId, String userName, ActionType actionType, ActionType actionTypeEdge, Object... additionalInfo) { + testNotifyEntityAllOneTimeLogEntityActionEntityEqClass(tenantId, entity, entityId, originatorId, tenantId, customerId, userId, userName, actionType, actionTypeEdge, additionalInfo); + } + + protected void testNotifyEntityAllOneTimeLogEntityActionEntityEqClass(TenantId entityTenantId, HasName entity, EntityId entityId, EntityId originatorId, + TenantId authTenantId, CustomerId customerId, UserId userId, String userName, + ActionType actionType, ActionType actionTypeEdge, Object... additionalInfo) { int cntTime = 1; - testNotificationMsgToEdgeServiceTime(entityId, tenantId, actionTypeEdge, cntTime); - testLogEntityActionEntityEqClass(entity, originatorId, tenantId, customerId, userId, userName, actionType, cntTime, additionalInfo); + testNotificationMsgToEdgeServiceTime(entityId, entityTenantId, actionTypeEdge, cntTime); + testLogEntityActionEntityEqClass(entity, originatorId, authTenantId, customerId, userId, userName, actionType, cntTime, additionalInfo); ArgumentMatcher matcherOriginatorId = argument -> argument.equals(originatorId); - testPushMsgToRuleEngineTime(matcherOriginatorId, tenantId, entity, cntTime); + testPushMsgToRuleEngineTime(matcherOriginatorId, authTenantId, entity, cntTime); Mockito.reset(tbClusterService, auditLogService); } @@ -159,17 +165,26 @@ public abstract class AbstractNotifyEntityTest extends AbstractWebTest { TenantId tenantId, CustomerId customerId, UserId userId, String userName, ActionType actionType, int cntTime, int cntTimeEdge, int cntTimeRuleEngine, Object... additionalInfo) { + testNotifyManyEntityManyTimeMsgToEdgeServiceEntityEqAny(tenantId, entity, originator, tenantId, customerId, userId, userName, actionType, + cntTime, cntTimeEdge, cntTimeRuleEngine, additionalInfo); + } + + protected void testNotifyManyEntityManyTimeMsgToEdgeServiceEntityEqAny(TenantId entityTenantId, HasName entity, HasName originator, + TenantId authTenantId, CustomerId customerId, UserId userId, String userName, + ActionType actionType, + int cntTime, int cntTimeEdge, int cntTimeRuleEngine, Object... additionalInfo) { EntityId originatorId = createEntityId_NULL_UUID(originator); - testSendNotificationMsgToEdgeServiceTimeEntityEqAny(tenantId, actionType, cntTimeEdge); + testSendNotificationMsgToEdgeServiceTimeEntityEqAny(entityTenantId, actionType, cntTimeEdge); ArgumentMatcher matcherEntityClassEquals = argument -> argument.getClass().equals(entity.getClass()); ArgumentMatcher matcherOriginatorId = argument -> argument.getClass().equals(originatorId.getClass()); ArgumentMatcher matcherCustomerId = customerId == null ? argument -> argument.getClass().equals(CustomerId.class) : argument -> argument.equals(customerId); ArgumentMatcher matcherUserId = userId == null ? argument -> argument.getClass().equals(UserId.class) : argument -> argument.equals(userId); - testLogEntityActionAdditionalInfo(matcherEntityClassEquals, matcherOriginatorId, tenantId, matcherCustomerId, matcherUserId, userName, actionType, cntTime, + testLogEntityActionAdditionalInfo(matcherEntityClassEquals, matcherOriginatorId, authTenantId, matcherCustomerId, matcherUserId, userName, actionType, cntTime, extractMatcherAdditionalInfoClass(additionalInfo)); - testPushMsgToRuleEngineTime(matcherOriginatorId, tenantId, entity, cntTimeRuleEngine); + testPushMsgToRuleEngineTime(matcherOriginatorId, authTenantId, entity, cntTimeRuleEngine); + } protected void testNotifyManyEntityManyTimeMsgToEdgeServiceEntityEqAnyAdditionalInfoAny(HasName entity, HasName originator, diff --git a/application/src/test/java/org/thingsboard/server/controller/TbResourceControllerTest.java b/application/src/test/java/org/thingsboard/server/controller/TbResourceControllerTest.java index 352e1fcfeb..d88aaa2756 100644 --- a/application/src/test/java/org/thingsboard/server/controller/TbResourceControllerTest.java +++ b/application/src/test/java/org/thingsboard/server/controller/TbResourceControllerTest.java @@ -943,7 +943,7 @@ public class TbResourceControllerTest extends AbstractControllerTest { private List loadLwm2mResources() throws Exception { - var models = List.of("1", "2", "3", "5", "6", "9", "19", "3303"); + var models = List.of("1", "2", "3-1_2", "5", "6", "9", "19", "3303"); List resources = new ArrayList<>(models.size()); diff --git a/application/src/test/java/org/thingsboard/server/controller/UserControllerTest.java b/application/src/test/java/org/thingsboard/server/controller/UserControllerTest.java index a08440ca4f..28d43699fd 100644 --- a/application/src/test/java/org/thingsboard/server/controller/UserControllerTest.java +++ b/application/src/test/java/org/thingsboard/server/controller/UserControllerTest.java @@ -116,7 +116,7 @@ public class UserControllerTest extends AbstractControllerTest { foundUser.setAdditionalInfo(savedUser.getAdditionalInfo()); Assert.assertEquals(foundUser, savedUser); - testNotifyManyEntityManyTimeMsgToEdgeServiceEntityEqAny(foundUser, foundUser, + testNotifyManyEntityManyTimeMsgToEdgeServiceEntityEqAny(user.getTenantId(), foundUser, foundUser, SYSTEM_TENANT, customerNUULId, null, SYS_ADMIN_EMAIL, ActionType.ADDED, 1, 1, 1); Mockito.reset(tbClusterService, auditLogService); @@ -155,7 +155,7 @@ public class UserControllerTest extends AbstractControllerTest { doDelete("/api/user/" + savedUser.getId().getId().toString()) .andExpect(status().isOk()); - testNotifyEntityAllOneTimeLogEntityActionEntityEqClass(foundUser, foundUser.getId(), foundUser.getId(), + testNotifyEntityAllOneTimeLogEntityActionEntityEqClass(user.getTenantId(), foundUser, foundUser.getId(), foundUser.getId(), SYSTEM_TENANT, customerNUULId, null, SYS_ADMIN_EMAIL, ActionType.DELETED, ActionType.DELETED, SYSTEM_TENANT.getId().toString()); } @@ -414,7 +414,7 @@ public class UserControllerTest extends AbstractControllerTest { User testManyUser = new User(); testManyUser.setTenantId(tenantId); - testNotifyManyEntityManyTimeMsgToEdgeServiceEntityEqAny(testManyUser, testManyUser, + testNotifyManyEntityManyTimeMsgToEdgeServiceEntityEqAny(tenantId, testManyUser, testManyUser, SYSTEM_TENANT, customerNUULId, null, SYS_ADMIN_EMAIL, ActionType.ADDED, cntEntity, cntEntity, cntEntity); @@ -526,7 +526,7 @@ public class UserControllerTest extends AbstractControllerTest { } User testManyUser = new User(); testManyUser.setTenantId(tenantId); - testNotifyManyEntityManyTimeMsgToEdgeServiceEntityEqAny(testManyUser, testManyUser, + testNotifyManyEntityManyTimeMsgToEdgeServiceEntityEqAny(tenantId, testManyUser, testManyUser, SYSTEM_TENANT, customerNUULId, null, SYS_ADMIN_EMAIL, ActionType.DELETED, cntEntity, NUMBER_OF_USERS, cntEntity, ""); diff --git a/application/src/test/java/org/thingsboard/server/transport/lwm2m/AbstractLwM2MIntegrationTest.java b/application/src/test/java/org/thingsboard/server/transport/lwm2m/AbstractLwM2MIntegrationTest.java index 226ef0dc57..a7d4cea8d4 100644 --- a/application/src/test/java/org/thingsboard/server/transport/lwm2m/AbstractLwM2MIntegrationTest.java +++ b/application/src/test/java/org/thingsboard/server/transport/lwm2m/AbstractLwM2MIntegrationTest.java @@ -21,6 +21,7 @@ import com.google.gson.JsonArray; import com.google.gson.JsonElement; import lombok.extern.slf4j.Slf4j; import org.apache.commons.io.IOUtils; +import org.awaitility.core.ConditionTimeoutException; import org.eclipse.leshan.client.LeshanClient; import org.eclipse.leshan.client.object.Security; import org.eclipse.leshan.client.servers.LwM2mServer; @@ -95,6 +96,7 @@ import java.util.Map; import java.util.Set; import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicLong; import static org.awaitility.Awaitility.await; import static org.eclipse.leshan.client.object.Security.noSec; @@ -119,6 +121,7 @@ import static org.thingsboard.server.transport.lwm2m.Lwm2mTestHelper.LwM2MClient import static org.thingsboard.server.transport.lwm2m.Lwm2mTestHelper.LwM2MClientState.ON_UPDATE_SUCCESS; import static org.thingsboard.server.transport.lwm2m.Lwm2mTestHelper.LwM2MProfileBootstrapConfigType; import static org.thingsboard.server.transport.lwm2m.Lwm2mTestHelper.LwM2MProfileBootstrapConfigType.NONE; +import static org.thingsboard.server.transport.lwm2m.Lwm2mTestHelper.lwm2mClientResources; import static org.thingsboard.server.transport.lwm2m.ota.AbstractOtaLwM2MIntegrationTest.CLIENT_LWM2M_SETTINGS_19; @Slf4j @@ -304,7 +307,7 @@ public abstract class AbstractLwM2MIntegrationTest extends AbstractTransportInte protected final Set expectedStatusesRegistrationBsSuccess = new HashSet<>(Arrays.asList(ON_BOOTSTRAP_STARTED, ON_BOOTSTRAP_SUCCESS, ON_REGISTRATION_STARTED, ON_REGISTRATION_SUCCESS)); protected ScheduledExecutorService executor; protected LwM2MTestClient lwM2MTestClient; - private String[] resources; + private String[] resources = lwm2mClientResources; protected String deviceId; protected boolean supportFormatOnly_SenMLJSON_SenMLCBOR = false; @@ -546,7 +549,9 @@ public abstract class AbstractLwM2MIntegrationTest extends AbstractTransportInte } public void setResources(String[] resources) { - this.resources = resources; + if (this.resources == null || !Arrays.equals(this.resources, resources)) { + this.resources = resources; + } } public void createNewClient(Security security, Security securityBs, boolean isRpc, @@ -741,11 +746,19 @@ public abstract class AbstractLwM2MIntegrationTest extends AbstractTransportInte return credentials; } - protected void awaitObserveReadAll(int cntObserve, String deviceIdStr) throws Exception { - await("ObserveReadAll: countObserve " + cntObserve) - .atMost(40, TimeUnit.SECONDS) - .until(() -> cntObserve == getCntObserveAll(deviceIdStr)); + + protected void awaitObserveReadAll(int cntObserve, String deviceIdStr) throws Exception { + try { + await("ObserveReadAll: countObserve " + cntObserve) + .atMost(40, TimeUnit.SECONDS) + .until(() -> cntObserve == getCntObserveAll(deviceIdStr)); + } catch (ConditionTimeoutException e) { + int current = getCntObserveAll(deviceIdStr); + log.error("Condition or device {} with alias 'ObserveReadAll: countObserve {}, but received {}", deviceIdStr, cntObserve, current); + throw e; + } } + protected void awaitDeleteDevice(String deviceIdStr) throws Exception { await("Delete device with id: " + deviceIdStr) .atMost(40, TimeUnit.SECONDS) @@ -756,6 +769,19 @@ public abstract class AbstractLwM2MIntegrationTest extends AbstractTransportInte }); } + protected void updateRegAtLeastOnceAfterAction() { + long initialInvocationCount = countUpdateReg(); + AtomicLong newInvocationCount = new AtomicLong(initialInvocationCount); + log.trace("updateRegAtLeastOnceAfterAction: initialInvocationCount [{}]", initialInvocationCount); + await("Update Registration at-least-once after action") + .atMost(50, TimeUnit.SECONDS) + .until(() -> { + newInvocationCount.set(countUpdateReg()); + return newInvocationCount.get() > initialInvocationCount; + }); + log.trace("updateRegAtLeastOnceAfterAction: newInvocationCount [{}]", newInvocationCount.get()); + } + protected Integer getCntObserveAll(String deviceIdStr) throws Exception { String actualResult = sendObserveOK("ObserveReadAll", null, deviceIdStr); ObjectNode rpcActualResult = JacksonUtil.fromString(actualResult, ObjectNode.class); diff --git a/application/src/test/java/org/thingsboard/server/transport/lwm2m/Lwm2mTestHelper.java b/application/src/test/java/org/thingsboard/server/transport/lwm2m/Lwm2mTestHelper.java index ce073d137d..c5233b378b 100644 --- a/application/src/test/java/org/thingsboard/server/transport/lwm2m/Lwm2mTestHelper.java +++ b/application/src/test/java/org/thingsboard/server/transport/lwm2m/Lwm2mTestHelper.java @@ -17,7 +17,7 @@ package org.thingsboard.server.transport.lwm2m; public class Lwm2mTestHelper { - public static final String[] lwm2mClientResources = new String[]{"3.xml", "5.xml", "6.xml", "9.xml", "19.xml", "3303.xml"}; + public static final String[] lwm2mClientResources = new String[]{"3-1_2.xml", "5.xml", "6.xml", "9.xml", "19.xml", "3303.xml"}; // Models public static final int BINARY_APP_DATA_CONTAINER = 19; diff --git a/application/src/test/java/org/thingsboard/server/transport/lwm2m/client/FwLwM2MDevice.java b/application/src/test/java/org/thingsboard/server/transport/lwm2m/client/FwLwM2MDevice.java index 051a68c9ab..8e247a8814 100644 --- a/application/src/test/java/org/thingsboard/server/transport/lwm2m/client/FwLwM2MDevice.java +++ b/application/src/test/java/org/thingsboard/server/transport/lwm2m/client/FwLwM2MDevice.java @@ -16,6 +16,7 @@ package org.thingsboard.server.transport.lwm2m.client; import lombok.extern.slf4j.Slf4j; +import org.eclipse.leshan.client.LeshanClient; import org.eclipse.leshan.client.resource.BaseInstanceEnabler; import org.eclipse.leshan.client.servers.LwM2mServer; import org.eclipse.leshan.core.model.ObjectModel; @@ -32,6 +33,9 @@ import java.util.List; import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicInteger; +import static org.thingsboard.server.dao.service.OtaPackageServiceTest.TARGET_FW_VERSION; +import static org.thingsboard.server.dao.service.OtaPackageServiceTest.TITLE; @Slf4j public class FwLwM2MDevice extends BaseInstanceEnabler implements Destroyable { @@ -44,6 +48,12 @@ public class FwLwM2MDevice extends BaseInstanceEnabler implements Destroyable { private final AtomicInteger updateResult = new AtomicInteger(0); + private LeshanClient leshanClient; + private String pkgNameDef = "firmware"; + private String pkgName; + private String pkgVersionDef = "1.0.0"; + private String pkgVersion; + @Override public ReadResponse read(LwM2mServer identity, int resourceId) { if (!identity.isSystem()) @@ -74,7 +84,7 @@ public class FwLwM2MDevice extends BaseInstanceEnabler implements Destroyable { switch (resourceId) { case 2: - startUpdating(); + startUpdating(identity); return ExecuteResponse.success(); default: return super.execute(identity, resourceId, arguments); @@ -106,11 +116,13 @@ public class FwLwM2MDevice extends BaseInstanceEnabler implements Destroyable { } private String getPkgName() { - return "firmware"; + this.pkgName = this.pkgName == null ? this.pkgNameDef : this.pkgName; + return this.pkgName; } private String getPkgVersion() { - return "1.0.0"; + this.pkgVersion = this.pkgVersion == null ? this.pkgVersionDef : this.pkgVersion; + return this.pkgVersion; } private int getFirmwareUpdateDeliveryMethod() { @@ -140,7 +152,7 @@ public class FwLwM2MDevice extends BaseInstanceEnabler implements Destroyable { }, 100, TimeUnit.MILLISECONDS); } - private void startUpdating() { + private void startUpdating(LwM2mServer identity) { scheduler.schedule(() -> { try { state.set(3); @@ -148,9 +160,25 @@ public class FwLwM2MDevice extends BaseInstanceEnabler implements Destroyable { Thread.sleep(100); updateResult.set(1); fireResourceChange(5); + this.pkgName = TITLE; + fireResourceChange(6); + this.pkgVersion = TARGET_FW_VERSION; + fireResourceChange(7); + if (this.leshanClient != null) { + log.info("Stop/reboot LwM2M client {}", this.leshanClient.getEndpoint(identity)); + this.leshanClient.stop(false); + log.info("Start after update fw LwM2M client {}", this.leshanClient.getEndpoint(identity)); + this.leshanClient.start(); + this.pkgName = this.pkgNameDef; + this.pkgVersion = this.pkgVersionDef; + } } catch (Exception e) { } }, 100, TimeUnit.MILLISECONDS); } + protected void setLeshanClient(LeshanClient leshanClient) { + this.leshanClient = leshanClient; + } + } diff --git a/application/src/test/java/org/thingsboard/server/transport/lwm2m/client/LwM2MTestClient.java b/application/src/test/java/org/thingsboard/server/transport/lwm2m/client/LwM2MTestClient.java index 993e416ded..d2543f84d7 100644 --- a/application/src/test/java/org/thingsboard/server/transport/lwm2m/client/LwM2MTestClient.java +++ b/application/src/test/java/org/thingsboard/server/transport/lwm2m/client/LwM2MTestClient.java @@ -467,6 +467,7 @@ public class LwM2MTestClient { this.awaitClientAfterStartConnectLw(); } lwM2mTemperatureSensor12.setLeshanClient(leshanClient); + fwLwM2MDevice.setLeshanClient(leshanClient); } } diff --git a/application/src/test/java/org/thingsboard/server/transport/lwm2m/ota/AbstractOtaLwM2MIntegrationTest.java b/application/src/test/java/org/thingsboard/server/transport/lwm2m/ota/AbstractOtaLwM2MIntegrationTest.java index a67f3cf00f..cff56694e5 100644 --- a/application/src/test/java/org/thingsboard/server/transport/lwm2m/ota/AbstractOtaLwM2MIntegrationTest.java +++ b/application/src/test/java/org/thingsboard/server/transport/lwm2m/ota/AbstractOtaLwM2MIntegrationTest.java @@ -54,7 +54,6 @@ import static org.thingsboard.server.transport.lwm2m.server.ota.DefaultLwM2MOtaU @DaoSqlTest public abstract class AbstractOtaLwM2MIntegrationTest extends AbstractLwM2MIntegrationTest { - private final String[] RESOURCES_OTA = new String[]{"3.xml", "5.xml", "9.xml", "19.xml"}; protected static final String CLIENT_ENDPOINT_WITHOUT_FW_INFO = "WithoutFirmwareInfoDevice"; protected static final String CLIENT_ENDPOINT_OTA5 = "Ota5_Device"; protected static final String CLIENT_ENDPOINT_OTA9 = "Ota9_Device"; @@ -186,10 +185,6 @@ public abstract class AbstractOtaLwM2MIntegrationTest extends AbstractLwM2MInteg " \"attributeLwm2m\": {}\n" + " }"; - public AbstractOtaLwM2MIntegrationTest() { - setResources(this.RESOURCES_OTA); - } - protected OtaPackageInfo createFirmware(String version, DeviceProfileId deviceProfileId) throws Exception { String CHECKSUM = "4bf5122f344554c53bde2ebb8cd2b7e3d1600ad631c385a5d7cce23c7785459a"; diff --git a/application/src/test/java/org/thingsboard/server/transport/lwm2m/ota/sql/Ota5LwM2MIntegrationTest.java b/application/src/test/java/org/thingsboard/server/transport/lwm2m/ota/sql/Ota5LwM2MIntegrationTest.java index 3065255687..e4db6f70bf 100644 --- a/application/src/test/java/org/thingsboard/server/transport/lwm2m/ota/sql/Ota5LwM2MIntegrationTest.java +++ b/application/src/test/java/org/thingsboard/server/transport/lwm2m/ota/sql/Ota5LwM2MIntegrationTest.java @@ -45,6 +45,7 @@ import static org.thingsboard.server.common.data.ota.OtaPackageUpdateStatus.INIT import static org.thingsboard.server.common.data.ota.OtaPackageUpdateStatus.QUEUED; import static org.thingsboard.server.common.data.ota.OtaPackageUpdateStatus.UPDATED; import static org.thingsboard.server.common.data.ota.OtaPackageUpdateStatus.UPDATING; +import static org.thingsboard.server.dao.service.OtaPackageServiceTest.TARGET_FW_VERSION; import static org.thingsboard.server.transport.lwm2m.Lwm2mTestHelper.BINARY_APP_DATA_CONTAINER; import static org.thingsboard.server.transport.lwm2m.Lwm2mTestHelper.LwM2MProfileBootstrapConfigType.NONE; import static org.thingsboard.server.transport.lwm2m.Lwm2mTestHelper.RESOURCE_ID_0; @@ -91,13 +92,14 @@ public class Ota5LwM2MIntegrationTest extends AbstractOtaLwM2MIntegrationTest { @Test public void testFirmwareUpdateByObject5_Ok() throws Exception { Lwm2mDeviceProfileTransportConfiguration transportConfiguration = getTransportConfiguration(OBSERVE_ATTRIBUTES_WITH_PARAMS_OTA5, getBootstrapServerCredentialsNoSec(NONE)); - DeviceProfile deviceProfile = createLwm2mDeviceProfile("profileFor" + this.CLIENT_ENDPOINT_OTA5, transportConfiguration); - LwM2MDeviceCredentials deviceCredentials = getDeviceCredentialsNoSec(createNoSecClientCredentials(this.CLIENT_ENDPOINT_OTA5)); - final Device device = createLwm2mDevice(deviceCredentials, this.CLIENT_ENDPOINT_OTA5, deviceProfile.getId()); - createNewClient(SECURITY_NO_SEC, null, false, this.CLIENT_ENDPOINT_OTA5, device.getId().getId().toString()); + DeviceProfile deviceProfile = createLwm2mDeviceProfile("profileFor" + this.CLIENT_ENDPOINT_OTA5 + "Ok", transportConfiguration); + String endpoint = this.CLIENT_ENDPOINT_OTA5 + "Ok"; + LwM2MDeviceCredentials deviceCredentials = getDeviceCredentialsNoSec(createNoSecClientCredentials(endpoint)); + final Device device = createLwm2mDevice(deviceCredentials, endpoint, deviceProfile.getId()); + createNewClient(SECURITY_NO_SEC, null, false, endpoint, device.getId().getId().toString()); awaitObserveReadAll(5, device.getId().getId().toString()); - device.setFirmwareId(createFirmware("fw.v.1.5.0-update", deviceProfile.getId()).getId()); + device.setFirmwareId(createFirmware(TARGET_FW_VERSION, deviceProfile.getId()).getId()); final Device savedDevice = doPost("/api/device", device, Device.class); assertThat(savedDevice).as("saved device").isNotNull(); @@ -110,7 +112,6 @@ public class Ota5LwM2MIntegrationTest extends AbstractOtaLwM2MIntegrationTest { log.warn("Object5: Got the ts: {}", ts); } - /** * ObjectId = 19/65533/0 * { @@ -133,13 +134,14 @@ public class Ota5LwM2MIntegrationTest extends AbstractOtaLwM2MIntegrationTest { @Test public void testFirmwareUpdateByObject5WithObject19_Ok() throws Exception { Lwm2mDeviceProfileTransportConfiguration transportConfiguration = getTransportConfiguration19(OBSERVE_ATTRIBUTES_WITH_PARAMS_OTA5_19, getBootstrapServerCredentialsNoSec(NONE)); - DeviceProfile deviceProfile = createLwm2mDeviceProfile("profileFor" + this.CLIENT_ENDPOINT_OTA5, transportConfiguration); - LwM2MDeviceCredentials deviceCredentials = getDeviceCredentialsNoSec(createNoSecClientCredentials(this.CLIENT_ENDPOINT_OTA5)); - final Device device = createLwm2mDevice(deviceCredentials, this.CLIENT_ENDPOINT_OTA5, deviceProfile.getId()); - createNewClient(SECURITY_NO_SEC, null, false, this.CLIENT_ENDPOINT_OTA5, device.getId().getId().toString()); + DeviceProfile deviceProfile = createLwm2mDeviceProfile("profileFor" + this.CLIENT_ENDPOINT_OTA5 + "19_Ok", transportConfiguration); + String endpoint = this.CLIENT_ENDPOINT_OTA5 + "19_Ok"; + LwM2MDeviceCredentials deviceCredentials = getDeviceCredentialsNoSec(createNoSecClientCredentials(endpoint)); + final Device device = createLwm2mDevice(deviceCredentials, endpoint, deviceProfile.getId()); + createNewClient(SECURITY_NO_SEC, null, false, endpoint, device.getId().getId().toString()); awaitObserveReadAll(6, device.getId().getId().toString()); - OtaPackageInfo otaPackageInfo = createFirmware("fw.v.1.5.0-update", deviceProfile.getId()); + OtaPackageInfo otaPackageInfo = createFirmware(TARGET_FW_VERSION, deviceProfile.getId()); device.setFirmwareId(otaPackageInfo.getId()); final Device savedDevice = doPost("/api/device", device, Device.class); @@ -154,6 +156,6 @@ public class Ota5LwM2MIntegrationTest extends AbstractOtaLwM2MIntegrationTest { String ver_Id_19 = lwM2MTestClient.getLeshanClient().getObjectTree().getModel().getObjectModel(BINARY_APP_DATA_CONTAINER).version; String resourceIdVer = "/" + BINARY_APP_DATA_CONTAINER + "_" + ver_Id_19 + "/" + FW_INSTANCE_ID + "/" + RESOURCE_ID_0; resultReadOtaParams_19(resourceIdVer, otaPackageInfo); - log.warn("Object5: Got the ts: {}", ts); + log.warn("Object5 with Object19: Got the ts: {}", ts); } } diff --git a/application/src/test/java/org/thingsboard/server/transport/lwm2m/rpc/AbstractRpcLwM2MIntegrationObserve_Ver_1_0_Test.java b/application/src/test/java/org/thingsboard/server/transport/lwm2m/rpc/AbstractRpcLwM2MIntegrationObserve_Ver_1_0_Test.java index 7e9efbbf75..f333fd012c 100644 --- a/application/src/test/java/org/thingsboard/server/transport/lwm2m/rpc/AbstractRpcLwM2MIntegrationObserve_Ver_1_0_Test.java +++ b/application/src/test/java/org/thingsboard/server/transport/lwm2m/rpc/AbstractRpcLwM2MIntegrationObserve_Ver_1_0_Test.java @@ -20,9 +20,8 @@ import org.thingsboard.server.dao.service.DaoSqlTest; @DaoSqlTest public abstract class AbstractRpcLwM2MIntegrationObserve_Ver_1_0_Test extends AbstractRpcLwM2MIntegrationTest{ - public AbstractRpcLwM2MIntegrationObserve_Ver_1_0_Test() { - String[] RESOURCES_RPC_VER_1_1 = new String[]{"3-1_0.xml", "5.xml", "6.xml", "9.xml", "19.xml"}; - setResources(RESOURCES_RPC_VER_1_1); + public AbstractRpcLwM2MIntegrationObserve_Ver_1_0_Test() throws Exception { + String[] RESOURCES_RPC_VER_1_0 = new String[]{"3-1_0.xml", "5.xml", "6.xml", "9.xml", "19.xml"}; + setResources(RESOURCES_RPC_VER_1_0); } } - diff --git a/application/src/test/java/org/thingsboard/server/transport/lwm2m/rpc/AbstractRpcLwM2MIntegrationObserve_Ver_1_1_Test.java b/application/src/test/java/org/thingsboard/server/transport/lwm2m/rpc/AbstractRpcLwM2MIntegrationObserve_Ver_1_1_Test.java index f21c857e8a..b39b3d1130 100644 --- a/application/src/test/java/org/thingsboard/server/transport/lwm2m/rpc/AbstractRpcLwM2MIntegrationObserve_Ver_1_1_Test.java +++ b/application/src/test/java/org/thingsboard/server/transport/lwm2m/rpc/AbstractRpcLwM2MIntegrationObserve_Ver_1_1_Test.java @@ -20,7 +20,7 @@ import org.thingsboard.server.dao.service.DaoSqlTest; @DaoSqlTest public abstract class AbstractRpcLwM2MIntegrationObserve_Ver_1_1_Test extends AbstractRpcLwM2MIntegrationTest{ - public AbstractRpcLwM2MIntegrationObserve_Ver_1_1_Test() { + public AbstractRpcLwM2MIntegrationObserve_Ver_1_1_Test() throws Exception { String[] RESOURCES_RPC_VER_1_1 = new String[]{"3-1_1.xml", "5.xml", "6.xml", "9.xml", "19.xml"}; setResources(RESOURCES_RPC_VER_1_1); } diff --git a/application/src/test/java/org/thingsboard/server/transport/lwm2m/rpc/AbstractRpcLwM2MIntegrationObserve_Ver_1_2_Test.java b/application/src/test/java/org/thingsboard/server/transport/lwm2m/rpc/AbstractRpcLwM2MIntegrationObserve_Ver_1_2_Test.java index 347d04a900..8c07379004 100644 --- a/application/src/test/java/org/thingsboard/server/transport/lwm2m/rpc/AbstractRpcLwM2MIntegrationObserve_Ver_1_2_Test.java +++ b/application/src/test/java/org/thingsboard/server/transport/lwm2m/rpc/AbstractRpcLwM2MIntegrationObserve_Ver_1_2_Test.java @@ -20,9 +20,9 @@ import org.thingsboard.server.dao.service.DaoSqlTest; @DaoSqlTest public abstract class AbstractRpcLwM2MIntegrationObserve_Ver_1_2_Test extends AbstractRpcLwM2MIntegrationTest{ - public AbstractRpcLwM2MIntegrationObserve_Ver_1_2_Test() { - String[] RESOURCES_RPC_VER_1_1 = new String[]{"3.xml", "5.xml", "6.xml", "9.xml", "19.xml"}; - setResources(RESOURCES_RPC_VER_1_1); + public AbstractRpcLwM2MIntegrationObserve_Ver_1_2_Test() throws Exception { + String[] RESOURCES_RPC_VER_1_2 = new String[]{"3-1_2.xml", "5.xml", "6.xml", "9.xml", "19.xml"}; + setResources(RESOURCES_RPC_VER_1_2); } } diff --git a/application/src/test/java/org/thingsboard/server/transport/lwm2m/rpc/AbstractRpcLwM2MIntegrationTest.java b/application/src/test/java/org/thingsboard/server/transport/lwm2m/rpc/AbstractRpcLwM2MIntegrationTest.java index 5ef2c1ad93..6c75f7161b 100644 --- a/application/src/test/java/org/thingsboard/server/transport/lwm2m/rpc/AbstractRpcLwM2MIntegrationTest.java +++ b/application/src/test/java/org/thingsboard/server/transport/lwm2m/rpc/AbstractRpcLwM2MIntegrationTest.java @@ -15,7 +15,9 @@ */ package org.thingsboard.server.transport.lwm2m.rpc; +import com.fasterxml.jackson.databind.node.ObjectNode; import lombok.extern.slf4j.Slf4j; +import org.eclipse.leshan.core.ResponseCode; import org.eclipse.leshan.core.link.LinkParser; import org.eclipse.leshan.core.link.lwm2m.DefaultLwM2mLinkParser; import org.junit.Before; @@ -40,6 +42,9 @@ import java.util.concurrent.atomic.AtomicLong; import java.util.function.Predicate; import static org.awaitility.Awaitility.await; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertTrue; +import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.status; import static org.eclipse.leshan.core.LwM2mId.ACCESS_CONTROL; import static org.eclipse.leshan.core.LwM2mId.DEVICE; import static org.eclipse.leshan.core.LwM2mId.FIRMWARE; @@ -101,10 +106,6 @@ public abstract class AbstractRpcLwM2MIntegrationTest extends AbstractLwM2MInteg @SpyBean protected LwM2mTransportServerHelper lwM2mTransportServerHelperTest; - public AbstractRpcLwM2MIntegrationTest() { - setResources(lwm2mClientResources); - } - @Before public void startInitRPC() throws Exception { if (this.getClass().getSimpleName().equals("RpcLwm2mIntegrationWriteCborTest")) { @@ -324,19 +325,6 @@ public abstract class AbstractRpcLwM2MIntegrationTest extends AbstractLwM2MInteg .count(); } - protected void updateRegAtLeastOnceAfterAction() { - long initialInvocationCount = countUpdateReg(); - AtomicLong newInvocationCount = new AtomicLong(initialInvocationCount); - log.trace("updateRegAtLeastOnceAfterAction: initialInvocationCount [{}]", initialInvocationCount); - await("Update Registration at-least-once after action") - .atMost(50, TimeUnit.SECONDS) - .until(() -> { - newInvocationCount.set(countUpdateReg()); - return newInvocationCount.get() > initialInvocationCount; - }); - log.trace("updateRegAtLeastOnceAfterAction: newInvocationCount [{}]", newInvocationCount.get()); - } - protected long countSendParametersOnThingsboardTelemetryResource(String rezName) { return Mockito.mockingDetails(lwM2mTransportServerHelperTest) .getInvocations().stream() @@ -350,4 +338,36 @@ public abstract class AbstractRpcLwM2MIntegrationTest extends AbstractLwM2MInteg ) .count(); } + + protected String sendDiscover(String path) throws Exception { + String setRpcRequest = "{\"method\": \"Discover\", \"params\": {\"id\": \"" + path + "\"}}"; + return doPostAsync("/api/plugins/rpc/twoway/" + lwM2MTestClient.getDeviceIdStr(), setRpcRequest, String.class, status().isOk()); + } + + protected String sendRpcObserveReadAllWithResult() throws Exception { + ObjectNode rpcActualResult = sendRpcObserveWithResult("ObserveReadAll", null); + assertEquals(ResponseCode.CONTENT.getName(), rpcActualResult.get("result").asText()); + return rpcActualResult.get("value").asText(); + } + + protected String sendRpcObserveReadAllWithResult(String params) throws Exception { + sendRpcObserveOk("Observe", params); + ObjectNode rpcActualResult = sendRpcObserveWithResult("ObserveReadAll", null); + assertEquals(ResponseCode.CONTENT.getName(), rpcActualResult.get("result").asText()); + return rpcActualResult.get("value").asText(); + } + + protected void testObserveOneResourceValue_Count_4_CancelAll_Reboot_After_Observe_Count_4(String expectedIdVer) throws Exception { + String expectedIdObserve = "SingleObservation:/3/0/9"; + sendObserveCancelAllWithAwait(lwM2MTestClient.getDeviceIdStr()); + updateRegAtLeastOnceAfterAction(); + lwM2MTestClient.getLeshanClient().stop(false); + lwM2MTestClient.getLeshanClient().start(); + updateRegAtLeastOnceAfterAction(); + awaitObserveReadAll(4,lwM2MTestClient.getDeviceIdStr()); + String actualIdVer = sendDiscover(objectIdVer_3); + assertTrue(actualIdVer.contains(expectedIdVer)); + String actualAllObserve = sendRpcObserveReadAllWithResult(); + assertTrue(actualAllObserve.contains(expectedIdObserve)); + } } diff --git a/application/src/test/java/org/thingsboard/server/transport/lwm2m/rpc/sql/RpcLwm2mIntegrationDiscoverTest.java b/application/src/test/java/org/thingsboard/server/transport/lwm2m/rpc/sql/RpcLwm2mIntegrationDiscoverTest.java index dceadfb7b5..77960b1c54 100644 --- a/application/src/test/java/org/thingsboard/server/transport/lwm2m/rpc/sql/RpcLwm2mIntegrationDiscoverTest.java +++ b/application/src/test/java/org/thingsboard/server/transport/lwm2m/rpc/sql/RpcLwm2mIntegrationDiscoverTest.java @@ -192,11 +192,6 @@ public class RpcLwm2mIntegrationDiscoverTest extends AbstractRpcLwM2MIntegration assertTrue(rpcActualResult.get("error").asText().contains(expected)); } - private String sendDiscover(String path) throws Exception { - String setRpcRequest = "{\"method\": \"Discover\", \"params\": {\"id\": \"" + path + "\"}}"; - return doPostAsync("/api/plugins/rpc/twoway/" + lwM2MTestClient.getDeviceIdStr(), setRpcRequest, String.class, status().isOk()); - } - private String convertObjectIdToVerId(String path, String ver) { ver = ver != null ? ver : TbLwM2mVersion.VERSION_1_0.getVersion().toString(); try { diff --git a/application/src/test/java/org/thingsboard/server/transport/lwm2m/rpc/sql/RpcLwm2mIntegrationDiscoverWriteAttributesTest.java b/application/src/test/java/org/thingsboard/server/transport/lwm2m/rpc/sql/RpcLwm2mIntegrationDiscoverWriteAttributesTest.java index bd1216eb06..2a4a401107 100644 --- a/application/src/test/java/org/thingsboard/server/transport/lwm2m/rpc/sql/RpcLwm2mIntegrationDiscoverWriteAttributesTest.java +++ b/application/src/test/java/org/thingsboard/server/transport/lwm2m/rpc/sql/RpcLwm2mIntegrationDiscoverWriteAttributesTest.java @@ -166,9 +166,4 @@ public class RpcLwm2mIntegrationDiscoverWriteAttributesTest extends AbstractRpcL String setRpcRequest = "{\"method\": \"WriteAttributes\", \"params\": {\"id\": \"" + path + "\", \"attributes\": " + value + " }}"; return doPostAsync("/api/plugins/rpc/twoway/" + lwM2MTestClient.getDeviceIdStr(), setRpcRequest, String.class, status().isOk()); } - - private String sendDiscover(String path) throws Exception { - String setRpcRequest = "{\"method\": \"Discover\", \"params\": {\"id\": \"" + path + "\"}}"; - return doPostAsync("/api/plugins/rpc/twoway/" + lwM2MTestClient.getDeviceIdStr(), setRpcRequest, String.class, status().isOk()); - } } diff --git a/application/src/test/java/org/thingsboard/server/transport/lwm2m/rpc/sql/RpcLwm2mIntegrationObserveTest.java b/application/src/test/java/org/thingsboard/server/transport/lwm2m/rpc/sql/RpcLwm2mIntegrationObserveTest.java index 483442382d..505cebff69 100644 --- a/application/src/test/java/org/thingsboard/server/transport/lwm2m/rpc/sql/RpcLwm2mIntegrationObserveTest.java +++ b/application/src/test/java/org/thingsboard/server/transport/lwm2m/rpc/sql/RpcLwm2mIntegrationObserveTest.java @@ -335,12 +335,5 @@ public class RpcLwm2mIntegrationObserveTest extends AbstractRpcLwM2MIntegrationT sendRpcObserveOk("Observe", expectedId_1); sendRpcObserveOk("Observe", expectedId_2); } - - private String sendRpcObserveReadAllWithResult(String params) throws Exception { - sendRpcObserveOk("Observe", params); - ObjectNode rpcActualResult = sendRpcObserveWithResult("ObserveReadAll", null); - assertEquals(ResponseCode.CONTENT.getName(), rpcActualResult.get("result").asText()); - return rpcActualResult.get("value").asText(); - } } diff --git a/application/src/test/java/org/thingsboard/server/transport/lwm2m/rpc/sql/RpcLwm2mIntegrationObserve_Ver_1_0_Test.java b/application/src/test/java/org/thingsboard/server/transport/lwm2m/rpc/sql/RpcLwm2mIntegrationObserveVer10Test.java similarity index 71% rename from application/src/test/java/org/thingsboard/server/transport/lwm2m/rpc/sql/RpcLwm2mIntegrationObserve_Ver_1_0_Test.java rename to application/src/test/java/org/thingsboard/server/transport/lwm2m/rpc/sql/RpcLwm2mIntegrationObserveVer10Test.java index 3d3640cb2e..e6f035e5cc 100644 --- a/application/src/test/java/org/thingsboard/server/transport/lwm2m/rpc/sql/RpcLwm2mIntegrationObserve_Ver_1_0_Test.java +++ b/application/src/test/java/org/thingsboard/server/transport/lwm2m/rpc/sql/RpcLwm2mIntegrationObserveVer10Test.java @@ -19,13 +19,14 @@ import lombok.extern.slf4j.Slf4j; import org.junit.Before; import org.junit.Test; import org.thingsboard.server.transport.lwm2m.rpc.AbstractRpcLwM2MIntegrationObserve_Ver_1_0_Test; -import org.thingsboard.server.transport.lwm2m.rpc.AbstractRpcLwM2MIntegrationObserve_Ver_1_1_Test; - import static org.junit.Assert.assertTrue; import static org.thingsboard.server.transport.lwm2m.Lwm2mTestHelper.RESOURCE_ID_NAME_3_9; @Slf4j -public class RpcLwm2mIntegrationObserve_Ver_1_0_Test extends AbstractRpcLwM2MIntegrationObserve_Ver_1_0_Test { +public class RpcLwm2mIntegrationObserveVer10Test extends AbstractRpcLwM2MIntegrationObserve_Ver_1_0_Test { + + public RpcLwm2mIntegrationObserveVer10Test() throws Exception { + } @Before public void setupObserveTest() throws Exception { @@ -44,5 +45,21 @@ public class RpcLwm2mIntegrationObserve_Ver_1_0_Test extends AbstractRpcLwM2MInt updateRegAtLeastOnceAfterAction(); long lastSendTelemetryAtCount = countSendParametersOnThingsboardTelemetryResource(RESOURCE_ID_NAME_3_9); assertTrue(lastSendTelemetryAtCount > initSendTelemetryAtCount); + awaitObserveReadAll(1,lwM2MTestClient.getDeviceIdStr()); + } + + /** + * "3_1.0/0/9" + * Observe count 4 + * CancelAll Observe + * Reboot + * Observe count 4 contains + * "/3_1.0" - Discover Object - find ver + * @throws Exception + */ + @Test + public void testObserveOneResourceValue_Count_4_CancelAll_Reboot_After_Observe_Count_4_ObjectVer_1_0() throws Exception { + String expectedIdVer = ";ver=1.0"; + testObserveOneResourceValue_Count_4_CancelAll_Reboot_After_Observe_Count_4(expectedIdVer); } } diff --git a/application/src/test/java/org/thingsboard/server/transport/lwm2m/rpc/sql/RpcLwm2mIntegrationObserve_Ver_1_1_Test.java b/application/src/test/java/org/thingsboard/server/transport/lwm2m/rpc/sql/RpcLwm2mIntegrationObserveVer11Test.java similarity index 72% rename from application/src/test/java/org/thingsboard/server/transport/lwm2m/rpc/sql/RpcLwm2mIntegrationObserve_Ver_1_1_Test.java rename to application/src/test/java/org/thingsboard/server/transport/lwm2m/rpc/sql/RpcLwm2mIntegrationObserveVer11Test.java index a4f7727773..a64ff43b92 100644 --- a/application/src/test/java/org/thingsboard/server/transport/lwm2m/rpc/sql/RpcLwm2mIntegrationObserve_Ver_1_1_Test.java +++ b/application/src/test/java/org/thingsboard/server/transport/lwm2m/rpc/sql/RpcLwm2mIntegrationObserveVer11Test.java @@ -23,7 +23,10 @@ import static org.junit.Assert.assertTrue; import static org.thingsboard.server.transport.lwm2m.Lwm2mTestHelper.RESOURCE_ID_NAME_3_9; @Slf4j -public class RpcLwm2mIntegrationObserve_Ver_1_1_Test extends AbstractRpcLwM2MIntegrationObserve_Ver_1_1_Test { +public class RpcLwm2mIntegrationObserveVer11Test extends AbstractRpcLwM2MIntegrationObserve_Ver_1_1_Test { + + public RpcLwm2mIntegrationObserveVer11Test() throws Exception { + } @Before public void setupObserveTest() throws Exception { @@ -43,4 +46,19 @@ public class RpcLwm2mIntegrationObserve_Ver_1_1_Test extends AbstractRpcLwM2MInt long lastSendTelemetryAtCount = countSendParametersOnThingsboardTelemetryResource(RESOURCE_ID_NAME_3_9); assertTrue(lastSendTelemetryAtCount > initSendTelemetryAtCount); } + + /** + * "3_1.1/0/9" + * Observe count 4 + * CancelAll Observe + * Reboot + * Observe count 4 contains + * "/3" - Discover Object - find ver (lwm2mVersion == 1.1) + * @throws Exception + */ + @Test + public void testObserveOneResourceValue_Count_4_CancelAll_Reboot_After_Observe_Count_4_ObjectVer_1_1() throws Exception { + String expectedIdVer = ""; + testObserveOneResourceValue_Count_4_CancelAll_Reboot_After_Observe_Count_4(expectedIdVer); + } } diff --git a/application/src/test/java/org/thingsboard/server/transport/lwm2m/rpc/sql/RpcLwm2mIntegrationObserve_Ver_1_2_Test.java b/application/src/test/java/org/thingsboard/server/transport/lwm2m/rpc/sql/RpcLwm2mIntegrationObserveVer12Test.java similarity index 73% rename from application/src/test/java/org/thingsboard/server/transport/lwm2m/rpc/sql/RpcLwm2mIntegrationObserve_Ver_1_2_Test.java rename to application/src/test/java/org/thingsboard/server/transport/lwm2m/rpc/sql/RpcLwm2mIntegrationObserveVer12Test.java index a166e0139b..89062dd303 100644 --- a/application/src/test/java/org/thingsboard/server/transport/lwm2m/rpc/sql/RpcLwm2mIntegrationObserve_Ver_1_2_Test.java +++ b/application/src/test/java/org/thingsboard/server/transport/lwm2m/rpc/sql/RpcLwm2mIntegrationObserveVer12Test.java @@ -18,14 +18,16 @@ package org.thingsboard.server.transport.lwm2m.rpc.sql; import lombok.extern.slf4j.Slf4j; import org.junit.Before; import org.junit.Test; -import org.thingsboard.server.transport.lwm2m.rpc.AbstractRpcLwM2MIntegrationObserve_Ver_1_0_Test; import org.thingsboard.server.transport.lwm2m.rpc.AbstractRpcLwM2MIntegrationObserve_Ver_1_2_Test; import static org.junit.Assert.assertTrue; import static org.thingsboard.server.transport.lwm2m.Lwm2mTestHelper.RESOURCE_ID_NAME_3_9; @Slf4j -public class RpcLwm2mIntegrationObserve_Ver_1_2_Test extends AbstractRpcLwM2MIntegrationObserve_Ver_1_2_Test { +public class RpcLwm2mIntegrationObserveVer12Test extends AbstractRpcLwM2MIntegrationObserve_Ver_1_2_Test { + + public RpcLwm2mIntegrationObserveVer12Test() throws Exception { + } @Before public void setupObserveTest() throws Exception { @@ -45,4 +47,19 @@ public class RpcLwm2mIntegrationObserve_Ver_1_2_Test extends AbstractRpcLwM2MInt long lastSendTelemetryAtCount = countSendParametersOnThingsboardTelemetryResource(RESOURCE_ID_NAME_3_9); assertTrue(lastSendTelemetryAtCount > initSendTelemetryAtCount); } + + /** + * "3_1.2/0/9" + * Observe count 4 + * CancelAll Observe + * Reboot + * Observe count 4 contains + * "/3_1.2" - Discover Object - find ver + * @throws Exception + */ + @Test + public void testObserveOneResourceValue_Count_4_CancelAll_Reboot_After_Observe_Count_4_ObjectVer_1_2() throws Exception { + String expectedIdVer = ";ver=1.2"; + testObserveOneResourceValue_Count_4_CancelAll_Reboot_After_Observe_Count_4(expectedIdVer); + } } diff --git a/application/src/test/java/org/thingsboard/server/transport/lwm2m/security/AbstractSecurityLwM2MIntegrationTest.java b/application/src/test/java/org/thingsboard/server/transport/lwm2m/security/AbstractSecurityLwM2MIntegrationTest.java index a16b782acf..e642035b32 100644 --- a/application/src/test/java/org/thingsboard/server/transport/lwm2m/security/AbstractSecurityLwM2MIntegrationTest.java +++ b/application/src/test/java/org/thingsboard/server/transport/lwm2m/security/AbstractSecurityLwM2MIntegrationTest.java @@ -22,6 +22,7 @@ import org.eclipse.leshan.client.object.Security; import org.eclipse.leshan.core.ResponseCode; import org.eclipse.leshan.core.util.Hex; import org.junit.Assert; +import org.junit.Before; import org.springframework.test.web.servlet.MvcResult; import org.thingsboard.common.util.JacksonUtil; import org.thingsboard.server.common.data.Device; @@ -119,7 +120,6 @@ public abstract class AbstractSecurityLwM2MIntegrationTest extends AbstractLwM2M protected final PrivateKey clientPrivateKeyFromCertTrust; // client private key used for X509 and RPK protected final X509Certificate clientX509CertTrustNo; // client certificate signed by intermediate, rootCA with a good CN ("host name") protected final PrivateKey clientPrivateKeyFromCertTrustNo; // client private key used for X509 and RPK - private final String[] RESOURCES_SECURITY = new String[]{"1.xml", "2.xml", "3.xml", "5.xml", "9.xml", "19.xml"}; private final LwM2MBootstrapClientCredentials defaultBootstrapCredentials; @@ -134,7 +134,6 @@ public abstract class AbstractSecurityLwM2MIntegrationTest extends AbstractLwM2M public AbstractSecurityLwM2MIntegrationTest() { // create client credentials - setResources(this.RESOURCES_SECURITY); try { // Get certificates from key store char[] clientKeyStorePwd = CLIENT_STORE_PWD.toCharArray(); @@ -178,6 +177,12 @@ public abstract class AbstractSecurityLwM2MIntegrationTest extends AbstractLwM2M defaultBootstrapCredentials.setLwm2mServer(serverCredentials); } + @Before + public void init() throws Exception { + String[] RESOURCES_SECURITY = new String[]{"3-1_2.xml", "5.xml", "6.xml", "9.xml", "19.xml"}; + setResources(RESOURCES_SECURITY); + } + public void basicTestConnectionStartBS(String clientEndpoint, String awaitAlias, LwM2MProfileBootstrapConfigType type, diff --git a/application/src/test/java/org/thingsboard/server/transport/lwm2m/security/sql/PskLwm2mIntegrationTest.java b/application/src/test/java/org/thingsboard/server/transport/lwm2m/security/sql/PskLwm2mIntegrationTest.java index 275383104e..a35b663052 100644 --- a/application/src/test/java/org/thingsboard/server/transport/lwm2m/security/sql/PskLwm2mIntegrationTest.java +++ b/application/src/test/java/org/thingsboard/server/transport/lwm2m/security/sql/PskLwm2mIntegrationTest.java @@ -68,10 +68,12 @@ public class PskLwm2mIntegrationTest extends AbstractSecurityLwM2MIntegrationTes ON_REGISTRATION_SUCCESS, true); } + @Test public void testWithPskConnectLwm2mOneObserveSuccessUpdateProfileManyObserveUpdateRegistrationSuccess() throws Exception { - String clientEndpoint = CLIENT_ENDPOINT_PSK; - String identity = CLIENT_PSK_IDENTITY; + String suf = "UpdateReg"; + String clientEndpoint = CLIENT_ENDPOINT_PSK + "_" + suf; + String identity = CLIENT_PSK_IDENTITY + "_" + suf; String keyPsk = CLIENT_PSK_KEY; PSKClientCredential clientCredentials = new PSKClientCredential(); clientCredentials.setEndpoint(clientEndpoint); @@ -103,10 +105,12 @@ public class PskLwm2mIntegrationTest extends AbstractSecurityLwM2MIntegrationTes awaitObserveReadAll(2, lwm2mDevice.getId().getId().toString()); awaitUpdateReg(3); } + @Test public void testWithPskConnectLwm2mSuccessObserveSuccessUnRegClientUpdateProfileObserveConnectLwm2mSuccessOWithNewObserve() throws Exception { - String clientEndpoint = CLIENT_ENDPOINT_PSK; - String identity = CLIENT_PSK_IDENTITY; + String suf = "UnReg"; + String clientEndpoint = CLIENT_ENDPOINT_PSK + "_" + suf; + String identity = CLIENT_PSK_IDENTITY + "_" + suf; String keyPsk = CLIENT_PSK_KEY; PSKClientCredential clientCredentials = new PSKClientCredential(); clientCredentials.setEndpoint(clientEndpoint); @@ -139,7 +143,7 @@ public class PskLwm2mIntegrationTest extends AbstractSecurityLwM2MIntegrationTes Assert.assertNotNull(lwm2mDeviceProfileManyParams); lwM2MTestClient.start(true); - awaitObserveReadAll(2, lwm2mDevice.getId().getId().toString()); + awaitObserveReadAll(1, lwm2mDevice.getId().getId().toString()); awaitUpdateReg(3); } diff --git a/application/src/test/resources/lwm2m/3.xml b/application/src/test/resources/lwm2m/3-1_2.xml similarity index 100% rename from application/src/test/resources/lwm2m/3.xml rename to application/src/test/resources/lwm2m/3-1_2.xml diff --git a/common/dao-api/src/main/java/org/thingsboard/server/dao/relation/RelationService.java b/common/dao-api/src/main/java/org/thingsboard/server/dao/relation/RelationService.java index 214dc5247e..aaa52f99a7 100644 --- a/common/dao-api/src/main/java/org/thingsboard/server/dao/relation/RelationService.java +++ b/common/dao-api/src/main/java/org/thingsboard/server/dao/relation/RelationService.java @@ -26,6 +26,7 @@ import org.thingsboard.server.common.data.relation.RelationTypeGroup; import org.thingsboard.server.common.data.rule.RuleChainType; import java.util.List; +import java.util.function.Predicate; /** * Created by ashvayka on 27.04.17. @@ -86,6 +87,8 @@ public interface RelationService { ListenableFuture> findByRelationPathQueryAsync(TenantId tenantId, EntityRelationPathQuery relationPathQuery); + ListenableFuture> findFilteredRelationsByPathQueryAsync(TenantId tenantId, EntityRelationPathQuery relationPathQuery, Predicate relationFilter); + List findByRelationPathQuery(TenantId tenantId, EntityRelationPathQuery relationPathQuery); // TODO: This method may be useful for some validations in the future diff --git a/common/data/src/main/java/org/thingsboard/server/common/data/cf/CalculatedField.java b/common/data/src/main/java/org/thingsboard/server/common/data/cf/CalculatedField.java index 0d2543d4c2..8c5adefcf8 100644 --- a/common/data/src/main/java/org/thingsboard/server/common/data/cf/CalculatedField.java +++ b/common/data/src/main/java/org/thingsboard/server/common/data/cf/CalculatedField.java @@ -65,6 +65,10 @@ public class CalculatedField extends BaseData implements HasN EntityType.DEVICE, EntityType.ASSET, EntityType.CUSTOMER, EntityType.TENANT )); + public static boolean isSupportedRefEntity(EntityId entity) { + return SUPPORTED_REFERENCED_ENTITIES.contains(entity.getEntityType()); + } + private TenantId tenantId; private EntityId entityId; diff --git a/common/data/src/main/java/org/thingsboard/server/common/data/cf/configuration/aggregation/RelatedEntitiesAggregationCalculatedFieldConfiguration.java b/common/data/src/main/java/org/thingsboard/server/common/data/cf/configuration/aggregation/RelatedEntitiesAggregationCalculatedFieldConfiguration.java index 9d4c7bdaf6..cf7040c4bb 100644 --- a/common/data/src/main/java/org/thingsboard/server/common/data/cf/configuration/aggregation/RelatedEntitiesAggregationCalculatedFieldConfiguration.java +++ b/common/data/src/main/java/org/thingsboard/server/common/data/cf/configuration/aggregation/RelatedEntitiesAggregationCalculatedFieldConfiguration.java @@ -23,12 +23,13 @@ import org.thingsboard.server.common.data.cf.CalculatedFieldType; import org.thingsboard.server.common.data.cf.configuration.Argument; import org.thingsboard.server.common.data.cf.configuration.ArgumentsBasedCalculatedFieldConfiguration; import org.thingsboard.server.common.data.cf.configuration.Output; +import org.thingsboard.server.common.data.cf.configuration.ScheduledUpdateSupportedCalculatedFieldConfiguration; import org.thingsboard.server.common.data.relation.RelationPathLevel; import java.util.Map; @Data -public class RelatedEntitiesAggregationCalculatedFieldConfiguration implements ArgumentsBasedCalculatedFieldConfiguration { +public class RelatedEntitiesAggregationCalculatedFieldConfiguration implements ArgumentsBasedCalculatedFieldConfiguration, ScheduledUpdateSupportedCalculatedFieldConfiguration { @NotNull private RelationPathLevel relation; @@ -40,11 +41,18 @@ public class RelatedEntitiesAggregationCalculatedFieldConfiguration implements A private Output output; private boolean useLatestTs; + private int scheduledUpdateInterval; + @Override public CalculatedFieldType getType() { return CalculatedFieldType.RELATED_ENTITIES_AGGREGATION; } + @Override + public boolean isScheduledUpdateEnabled() { + return true; + } + @Override public void validate() { relation.validate(); diff --git a/common/data/src/main/java/org/thingsboard/server/common/data/tenant/profile/DefaultTenantProfileConfiguration.java b/common/data/src/main/java/org/thingsboard/server/common/data/tenant/profile/DefaultTenantProfileConfiguration.java index 9457b688f2..62af00300e 100644 --- a/common/data/src/main/java/org/thingsboard/server/common/data/tenant/profile/DefaultTenantProfileConfiguration.java +++ b/common/data/src/main/java/org/thingsboard/server/common/data/tenant/profile/DefaultTenantProfileConfiguration.java @@ -186,8 +186,8 @@ public class DefaultTenantProfileConfiguration implements TenantProfileConfigura private long maxStateSizeInKBytes = 32; @Schema(example = "2") private long maxSingleValueArgumentSizeInKBytes = 2; - @Schema(example = "3600") - private long minAllowedDeduplicationIntervalInSecForCF = 3600; + @Schema(example = "60") + private long minAllowedDeduplicationIntervalInSecForCF = 60; @Schema(example = "60") private long minAggregationIntervalInSecForCF = 60; diff --git a/common/transport/lwm2m/src/main/java/org/thingsboard/server/transport/lwm2m/server/uplink/DefaultLwM2mUplinkMsgHandler.java b/common/transport/lwm2m/src/main/java/org/thingsboard/server/transport/lwm2m/server/uplink/DefaultLwM2mUplinkMsgHandler.java index 201376dbb7..b7e56c139a 100644 --- a/common/transport/lwm2m/src/main/java/org/thingsboard/server/transport/lwm2m/server/uplink/DefaultLwM2mUplinkMsgHandler.java +++ b/common/transport/lwm2m/src/main/java/org/thingsboard/server/transport/lwm2m/server/uplink/DefaultLwM2mUplinkMsgHandler.java @@ -224,8 +224,11 @@ public class DefaultLwM2mUplinkMsgHandler extends LwM2MExecutorAwareService impl log.info("[{}] Closing old session: {}", registration.getEndpoint(), new UUID(oldSessionInfo.get().getSessionIdMSB(), oldSessionInfo.get().getSessionIdLSB())); sessionManager.deregister(oldSessionInfo.get()); } - logService.log(lwM2MClient, LOG_LWM2M_INFO + ": Client registered with registration id: " + registration.getId() + " version: " - + registration.getLwM2mVersion() + " and modes: " + registration.getQueueMode() + ", " + registration.getBindingMode()); + String msgLogService = String.format(""" + %s: Endpoint [%s] Client registered with registration id: [%s] LwM2mVersion: [%s], SupportedObjectIdVer [%s] QueueMode [%s], BindingMode %s + """, LOG_LWM2M_INFO, registration.getEndpoint(), registration.getId(), registration.getLwM2mVersion(), registration.getSupportedObject(), registration.getQueueMode(), registration.getBindingMode()); + logService.log(lwM2MClient, msgLogService); + log.debug(msgLogService); sessionManager.register(lwM2MClient.getSession()); this.initClientTelemetry(lwM2MClient); this.initAttributes(lwM2MClient, true); @@ -244,7 +247,7 @@ public class DefaultLwM2mUplinkMsgHandler extends LwM2MExecutorAwareService impl logService.log(lwM2MClient, LOG_LWM2M_WARN + ": Client registration failed due to invalid state: " + stateException.getState()); } } catch (Throwable t) { - log.error("[{}] endpoint [{}] error Unable registration.", registration.getEndpoint(), t); + log.error("Endpoint [{}], Error Unable registration: [{}].", registration.getEndpoint(), t.getMessage(), t); logService.log(lwM2MClient, LOG_LWM2M_WARN + ": Client registration failed due to: " + t.getMessage()); } }); @@ -290,7 +293,6 @@ public class DefaultLwM2mUplinkMsgHandler extends LwM2MExecutorAwareService impl clientContext.unregister(client, registration); SessionInfoProto sessionInfo = client.getSession(); if (sessionInfo != null) { - securityStore.remove(client.getEndpoint(), client.getRegistration().getId()); sessionManager.deregister(sessionInfo); sessionStore.remove(registration.getEndpoint()); log.info("Client close session: [{}] unReg [{}] name [{}] profile ", registration.getId(), registration.getEndpoint(), sessionInfo.getDeviceType()); diff --git a/dao/src/main/java/org/thingsboard/server/dao/relation/BaseRelationService.java b/dao/src/main/java/org/thingsboard/server/dao/relation/BaseRelationService.java index ff0d09ef2c..ffb800c7f7 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/relation/BaseRelationService.java +++ b/dao/src/main/java/org/thingsboard/server/dao/relation/BaseRelationService.java @@ -71,6 +71,7 @@ import java.util.concurrent.ConcurrentLinkedQueue; import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.TimeUnit; import java.util.function.BiConsumer; +import java.util.function.Predicate; import static org.thingsboard.server.dao.service.Validator.validateId; import static org.thingsboard.server.dao.service.Validator.validatePositiveNumber; @@ -507,6 +508,11 @@ public class BaseRelationService implements RelationService { @Override public ListenableFuture> findByRelationPathQueryAsync(TenantId tenantId, EntityRelationPathQuery relationPathQuery) { + return findFilteredRelationsByPathQueryAsync(tenantId, relationPathQuery, null); + } + + @Override + public ListenableFuture> findFilteredRelationsByPathQueryAsync(TenantId tenantId, EntityRelationPathQuery relationPathQuery, Predicate relationFilter) { log.trace("Executing findByRelationPathQuery, tenantId [{}], relationPathQuery {}", tenantId, relationPathQuery); validateId(tenantId, id -> "Invalid tenant id: " + id); validate(relationPathQuery); @@ -518,10 +524,24 @@ public class BaseRelationService implements RelationService { case FROM -> findByFromAndTypeAsync(tenantId, relationPathQuery.rootEntityId(), relationPathLevel.relationType(), RelationTypeGroup.COMMON); case TO -> findByToAndTypeAsync(tenantId, relationPathQuery.rootEntityId(), relationPathLevel.relationType(), RelationTypeGroup.COMMON); }; - return Futures.transform(relationsFuture, entityRelations -> entityRelations.size() > limit ? - entityRelations.subList(0, limit) : entityRelations, MoreExecutors.directExecutor()); + return Futures.transform(relationsFuture, entityRelations -> { + if (entityRelations == null || entityRelations.isEmpty()) { + return Collections.emptyList(); + } + List relations = relationFilter != null ? filterRelations(entityRelations, relationFilter) : entityRelations; + return relations.size() > limit ? relations.subList(0, limit) : relations; + }, MoreExecutors.directExecutor()); } - return executor.submit(() -> relationDao.findByRelationPathQuery(tenantId, relationPathQuery, limit)); + return executor.submit(() -> { + List entityRelations = relationDao.findByRelationPathQuery(tenantId, relationPathQuery, limit); + return relationFilter != null ? filterRelations(entityRelations, relationFilter) : entityRelations; + }); + } + + private List filterRelations(List entityRelations, Predicate relationFilter) { + return entityRelations.stream() + .filter(relationFilter) + .toList(); } @Override diff --git a/dao/src/main/java/org/thingsboard/server/dao/user/UserServiceImpl.java b/dao/src/main/java/org/thingsboard/server/dao/user/UserServiceImpl.java index 3336da44a0..9209b69ccf 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/user/UserServiceImpl.java +++ b/dao/src/main/java/org/thingsboard/server/dao/user/UserServiceImpl.java @@ -190,7 +190,7 @@ public class UserServiceImpl extends AbstractCachedEntityService { this.columns.push(new EntityTableColumn('createRule', 'alarm-rule.severities', '67%', entity => Object.keys(entity.configuration.createRules).map((severity) => this.translate.instant(alarmSeverityTranslations.get(severity as AlarmSeverity))).join(', '), () => ({}), false)); - this.columns.push(new EntityTableColumn('clearRule', 'alarm-rule.cleared', '60px', - entity => checkBoxCell(!!entity.configuration.clearRule), ()=> { return {padding: '0 14px'}}, false)); + this.columns.push(new EntityTableColumn('clearRule', 'alarm-rule.cleared', '70px', + entity => checkBoxCell(!!entity.configuration.clearRule), ()=> { return {padding: 0, textAlign: 'center'}}, false)); this.cellActionDescriptors.push( { diff --git a/ui-ngx/src/app/modules/home/components/alarm-rules/cf-alarm-rule.component.html b/ui-ngx/src/app/modules/home/components/alarm-rules/cf-alarm-rule.component.html index e1f2f41bfd..cd6b2c3796 100644 --- a/ui-ngx/src/app/modules/home/components/alarm-rules/cf-alarm-rule.component.html +++ b/ui-ngx/src/app/modules/home/components/alarm-rules/cf-alarm-rule.component.html @@ -18,7 +18,7 @@
- @if (!disabled || alarmRuleFormGroup.get('dashboardId').value) { + @if (!disabled || alarmRuleFormGroup.get('alarmDetails').value) {
alarm-rule.alarm-rule-additional-info diff --git a/ui-ngx/src/app/modules/home/components/alarm-rules/filter/alarm-rule-complex-filter-predicate-dialog.component.ts b/ui-ngx/src/app/modules/home/components/alarm-rules/filter/alarm-rule-complex-filter-predicate-dialog.component.ts index dab366f960..c8766b0b17 100644 --- a/ui-ngx/src/app/modules/home/components/alarm-rules/filter/alarm-rule-complex-filter-predicate-dialog.component.ts +++ b/ui-ngx/src/app/modules/home/components/alarm-rules/filter/alarm-rule-complex-filter-predicate-dialog.component.ts @@ -15,7 +15,6 @@ /// import { Component, Inject } from '@angular/core'; -import { ErrorStateMatcher } from '@angular/material/core'; import { MAT_DIALOG_DATA, MatDialogRef } from '@angular/material/dialog'; import { Store } from '@ngrx/store'; import { AppState } from '@core/core.state'; @@ -41,7 +40,7 @@ export interface AlarmRuleComplexFilterPredicateDialogData { @Component({ selector: 'tb-alarm-rule-complex-filter-predicate-dialog', templateUrl: './alarm-rule-complex-filter-predicate-dialog.component.html', - providers: [{provide: ErrorStateMatcher, useExisting: AlarmRuleComplexFilterPredicateDialogComponent}], + providers: [], styleUrls: [] }) diff --git a/ui-ngx/src/app/modules/home/components/alarm-rules/filter/alarm-rule-filter-list.component.ts b/ui-ngx/src/app/modules/home/components/alarm-rules/filter/alarm-rule-filter-list.component.ts index 86b7520910..11e2728659 100644 --- a/ui-ngx/src/app/modules/home/components/alarm-rules/filter/alarm-rule-filter-list.component.ts +++ b/ui-ngx/src/app/modules/home/components/alarm-rules/filter/alarm-rule-filter-list.component.ts @@ -102,6 +102,14 @@ export class AlarmRuleFilterListComponent implements ControlValueAccessor, Valid }; } + setDisabledState(isDisabled: boolean): void { + if (isDisabled) { + this.filterListFormGroup.disable({emitEvent: false}); + } else { + this.filterListFormGroup.enable({emitEvent: false}); + } + } + writeValue(filters: Array): void { const keyFilterControls: Array = []; if (filters) { diff --git a/ui-ngx/src/app/modules/home/components/alarm-rules/filter/alarm-rule-filter-predicate-list.component.html b/ui-ngx/src/app/modules/home/components/alarm-rules/filter/alarm-rule-filter-predicate-list.component.html index 210f3f9424..4eaa1eafb6 100644 --- a/ui-ngx/src/app/modules/home/components/alarm-rules/filter/alarm-rule-filter-predicate-list.component.html +++ b/ui-ngx/src/app/modules/home/components/alarm-rules/filter/alarm-rule-filter-predicate-list.component.html @@ -46,7 +46,7 @@ [formControl]="predicateControl">