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

Merge pull request #14501 from thingsboard/bugfix/cfs-arguments-last-update-ts 

Fixed last update ts handling for CFs arguments + minor improvements to calculated field procesisng logic
pull/14503/head
Viacheslav Klimov 9 months ago
committed by GitHub
parent
65d52bef93 071c254f67
commit
56c42a380a
No known key found for this signature in database GPG Key ID: B5690EEEBB952194
8 changed files with 78 additions and 82 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Unified View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 59
      application/src/main/java/org/thingsboard/server/service/cf/AbstractCalculatedFieldProcessingService.java
  2. 14
      application/src/main/java/org/thingsboard/server/service/cf/ctx/state/BaseCalculatedFieldState.java
  3. 18
      application/src/main/java/org/thingsboard/server/service/cf/ctx/state/CalculatedFieldCtx.java
  4. 11
      application/src/main/java/org/thingsboard/server/service/cf/ctx/state/SingleValueArgumentEntry.java
  5. 19
      application/src/main/java/org/thingsboard/server/service/cf/ctx/state/aggregation/RelatedEntitiesAggregationCalculatedFieldState.java
  6. 6
      application/src/main/java/org/thingsboard/server/service/cf/ctx/state/aggregation/single/EntityAggregationCalculatedFieldState.java
  7. 4
      application/src/main/java/org/thingsboard/server/service/cf/ctx/state/geofencing/GeofencingCalculatedFieldState.java
  8. 29
      application/src/main/java/org/thingsboard/server/utils/CalculatedFieldArgumentUtils.java

59
application/src/main/java/org/thingsboard/server/service/cf/AbstractCalculatedFieldProcessingService.java
View File

@ -54,9 +54,7 @@ import org.thingsboard.server.common.data.id.EntityId;
import org.thingsboard.server.common.data.id.TenantId; import org.thingsboard.server.common.data.id.TenantId;
import org.thingsboard.server.common.data.kv.Aggregation; import org.thingsboard.server.common.data.kv.Aggregation;
import org.thingsboard.server.common.data.kv.AttributeKvEntry; import org.thingsboard.server.common.data.kv.AttributeKvEntry;
import org.thingsboard.server.common.data.kv.BaseAttributeKvEntry;
import org.thingsboard.server.common.data.kv.BaseReadTsKvQuery; import org.thingsboard.server.common.data.kv.BaseReadTsKvQuery;
import org.thingsboard.server.common.data.kv.BasicTsKvEntry;
import org.thingsboard.server.common.data.kv.KvEntry; import org.thingsboard.server.common.data.kv.KvEntry;
import org.thingsboard.server.common.data.kv.ReadTsKvQuery; import org.thingsboard.server.common.data.kv.ReadTsKvQuery;
import org.thingsboard.server.common.data.kv.TsKvEntry; import org.thingsboard.server.common.data.kv.TsKvEntry;
@ -75,7 +73,6 @@ import org.thingsboard.server.queue.TbQueueCallback;
import org.thingsboard.server.queue.TbQueueMsgMetadata; import org.thingsboard.server.queue.TbQueueMsgMetadata;
import org.thingsboard.server.service.cf.ctx.state.ArgumentEntry; import org.thingsboard.server.service.cf.ctx.state.ArgumentEntry;
import org.thingsboard.server.service.cf.ctx.state.CalculatedFieldCtx; import org.thingsboard.server.service.cf.ctx.state.CalculatedFieldCtx;
import org.thingsboard.server.service.cf.ctx.state.SingleValueArgumentEntry;
import org.thingsboard.server.service.cf.ctx.state.aggregation.single.AggIntervalEntry; import org.thingsboard.server.service.cf.ctx.state.aggregation.single.AggIntervalEntry;
import org.thingsboard.server.service.telemetry.TelemetrySubscriptionService; import org.thingsboard.server.service.telemetry.TelemetrySubscriptionService;
@ -84,7 +81,6 @@ import java.util.HashMap;
import java.util.List; import java.util.List;
import java.util.Map; import java.util.Map;
import java.util.Objects; import java.util.Objects;
import java.util.Optional;
import java.util.Set; import java.util.Set;
import java.util.concurrent.ExecutionException; import java.util.concurrent.ExecutionException;
import java.util.function.Function; import java.util.function.Function;
@ -101,7 +97,7 @@ import static org.thingsboard.server.common.data.msg.TbMsgType.ATTRIBUTES_UPDATE
import static org.thingsboard.server.dao.util.KvUtils.filterChangedAttr; import static org.thingsboard.server.dao.util.KvUtils.filterChangedAttr;
import static org.thingsboard.server.dao.util.KvUtils.toTsKvEntryList; import static org.thingsboard.server.dao.util.KvUtils.toTsKvEntryList;
import static org.thingsboard.server.utils.CalculatedFieldArgumentUtils.createDefaultAttributeEntry; import static org.thingsboard.server.utils.CalculatedFieldArgumentUtils.createDefaultAttributeEntry;
import static org.thingsboard.server.utils.CalculatedFieldArgumentUtils.createDefaultKvEntry; import static org.thingsboard.server.utils.CalculatedFieldArgumentUtils.createDefaultTsKvEntry;
import static org.thingsboard.server.utils.CalculatedFieldArgumentUtils.transformAggMetricArgument; import static org.thingsboard.server.utils.CalculatedFieldArgumentUtils.transformAggMetricArgument;
import static org.thingsboard.server.utils.CalculatedFieldArgumentUtils.transformAggregationArgument; import static org.thingsboard.server.utils.CalculatedFieldArgumentUtils.transformAggregationArgument;
import static org.thingsboard.server.utils.CalculatedFieldArgumentUtils.transformSingleValueArgument; import static org.thingsboard.server.utils.CalculatedFieldArgumentUtils.transformSingleValueArgument;
@ -210,14 +206,14 @@ public abstract class AbstractCalculatedFieldProcessingService {
default -> { default -> {
var resolvedEntityIdsFuture = resolveGeofencingEntityIds(ctx.getTenantId(), entityId, entry); var resolvedEntityIdsFuture = resolveGeofencingEntityIds(ctx.getTenantId(), entityId, entry);
argFutures.put(entry.getKey(), Futures.transformAsync(resolvedEntityIdsFuture, resolvedEntityIds -> argFutures.put(entry.getKey(), Futures.transformAsync(resolvedEntityIdsFuture, resolvedEntityIds ->
fetchGeofencingKvEntry(ctx.getTenantId(), resolvedEntityIds, entry.getValue()), MoreExecutors.directExecutor())); fetchGeofencingArgumentValue(ctx.getTenantId(), resolvedEntityIds, entry.getValue(), startTs), MoreExecutors.directExecutor()));
} }
} }
} }
return argFutures; return argFutures;
} }
protected Map<String, ListenableFuture<ArgumentEntry>> fetchRelatedEntitiesAggArguments(CalculatedFieldCtx ctx, EntityId entityId, long ts) { private Map<String, ListenableFuture<ArgumentEntry>> fetchRelatedEntitiesAggArguments(CalculatedFieldCtx ctx, EntityId entityId, long ts) {
if (!(ctx.getCalculatedField().getConfiguration() instanceof RelatedEntitiesAggregationCalculatedFieldConfiguration config)) { if (!(ctx.getCalculatedField().getConfiguration() instanceof RelatedEntitiesAggregationCalculatedFieldConfiguration config)) {
return Collections.emptyMap(); return Collections.emptyMap();
} }
@ -230,7 +226,7 @@ public abstract class AbstractCalculatedFieldProcessingService {
)); ));
} }
protected Map<String, ListenableFuture<ArgumentEntry>> fetchEntityAggArguments(CalculatedFieldCtx ctx, EntityId entityId, long ts) { private Map<String, ListenableFuture<ArgumentEntry>> fetchEntityAggArguments(CalculatedFieldCtx ctx, EntityId entityId, long ts) {
if (!(ctx.getCalculatedField().getConfiguration() instanceof EntityAggregationCalculatedFieldConfiguration config)) { if (!(ctx.getCalculatedField().getConfiguration() instanceof EntityAggregationCalculatedFieldConfiguration config)) {
return Collections.emptyMap(); return Collections.emptyMap();
} }
@ -291,31 +287,28 @@ public abstract class AbstractCalculatedFieldProcessingService {
return ownerService.getOwner(tenantId, entityId); return ownerService.getOwner(tenantId, entityId);
} }
private ListenableFuture<ArgumentEntry> fetchGeofencingKvEntry(TenantId tenantId, List<EntityId> geofencingEntities, Argument argument) { private ListenableFuture<ArgumentEntry> fetchGeofencingArgumentValue(TenantId tenantId, List<EntityId> geofencingEntities, Argument argument, long startTs) {
if (argument.getRefEntityKey().getType() != ArgumentType.ATTRIBUTE) { if (argument.getRefEntityKey().getType() != ArgumentType.ATTRIBUTE) {
throw new IllegalStateException("Unsupported argument key type: " + argument.getRefEntityKey().getType()); throw new IllegalStateException("Unsupported argument key type: " + argument.getRefEntityKey().getType());
} }
List<ListenableFuture<Map.Entry<EntityId, AttributeKvEntry>>> kvFutures = geofencingEntities.stream() var geofencingEntityIdToKvEntryMapFutures = Futures.allAsList(fetchGeofencingEntityIdToKvEntriesFutures(tenantId, geofencingEntities, argument, startTs));
return Futures.transform(geofencingEntityIdToKvEntryMapFutures, entries -> ArgumentEntry.createGeofencingValueArgument(entries.stream()
.collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue))), MoreExecutors.directExecutor());
}
private List<ListenableFuture<Map.Entry<EntityId, AttributeKvEntry>>> fetchGeofencingEntityIdToKvEntriesFutures(TenantId tenantId, List<EntityId> geofencingEntities, Argument argument, long startTs) {
return geofencingEntities.stream()
.map(entityId -> { .map(entityId -> {
var attributesFuture = attributesService.find( AttributeScope scope = argument.getRefEntityKey().getScope();
tenantId, String key = argument.getRefEntityKey().getKey();
entityId, var attributesFuture = attributesService.find(tenantId, entityId, scope, key);
argument.getRefEntityKey().getScope(),
argument.getRefEntityKey().getKey()
);
return Futures.transform(attributesFuture, resultOpt -> return Futures.transform(attributesFuture, resultOpt ->
Map.entry(entityId, resultOpt.orElseGet(() -> createDefaultAttributeEntry(argument, System.currentTimeMillis()))), Map.entry(entityId, resultOpt.orElseGet(() -> createDefaultAttributeEntry(argument, startTs))),
calculatedFieldCallbackExecutor calculatedFieldCallbackExecutor);
);
}).collect(Collectors.toList()); }).collect(Collectors.toList());
ListenableFuture<List<Map.Entry<EntityId, AttributeKvEntry>>> allFutures = Futures.allAsList(kvFutures);
return Futures.transform(allFutures, entries -> ArgumentEntry.createGeofencingValueArgument(entries.stream()
.collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue))), MoreExecutors.directExecutor());
} }
public ListenableFuture<ArgumentEntry> fetchRelatedEntitiesArgumentEntry(TenantId tenantId, List<EntityId> aggEntities, Argument argument, long startTs) { private ListenableFuture<ArgumentEntry> fetchRelatedEntitiesArgumentEntry(TenantId tenantId, List<EntityId> aggEntities, Argument argument, long startTs) {
List<ListenableFuture<Map.Entry<EntityId, ArgumentEntry>>> futures = aggEntities.stream() List<ListenableFuture<Map.Entry<EntityId, ArgumentEntry>>> futures = aggEntities.stream()
.map(entityId -> { .map(entityId -> {
ListenableFuture<ArgumentEntry> argumentEntryFut = fetchArgumentValue(tenantId, entityId, argument, startTs); ListenableFuture<ArgumentEntry> argumentEntryFut = fetchArgumentValue(tenantId, entityId, argument, startTs);
@ -368,21 +361,17 @@ public abstract class AbstractCalculatedFieldProcessingService {
return Futures.transform(attributeOptFuture, attrOpt -> { return Futures.transform(attributeOptFuture, attrOpt -> {
log.debug("[{}][{}] Fetched attribute for key {}: {}", tenantId, entityId, argument.getRefEntityKey(), attrOpt); log.debug("[{}][{}] Fetched attribute for key {}: {}", tenantId, entityId, argument.getRefEntityKey(), attrOpt);
AttributeKvEntry attributeKvEntry = attrOpt.orElseGet(() -> new BaseAttributeKvEntry(createDefaultKvEntry(argument), defaultLastUpdateTs, SingleValueArgumentEntry.DEFAULT_VERSION)); return transformSingleValueArgument(attrOpt.orElseGet(() -> createDefaultAttributeEntry(argument, defaultLastUpdateTs)));
return transformSingleValueArgument(Optional.of(attributeKvEntry));
}, calculatedFieldCallbackExecutor); }, calculatedFieldCallbackExecutor);
} }
protected ListenableFuture<ArgumentEntry> fetchTsLatest(TenantId tenantId, EntityId entityId, Argument argument, long defaultTs) { private ListenableFuture<ArgumentEntry> fetchTsLatest(TenantId tenantId, EntityId entityId, Argument argument, long defaultTs) {
String timeseriesKey = argument.getRefEntityKey().getKey(); String timeseriesKey = argument.getRefEntityKey().getKey();
log.trace("[{}][{}] Fetching latest timeseries {}", tenantId, entityId, timeseriesKey); log.trace("[{}][{}] Fetching latest timeseries {}", tenantId, entityId, timeseriesKey);
return transformSingleValueArgument( return Futures.transform(timeseriesService.findLatest(tenantId, entityId, timeseriesKey), result -> {
Futures.transform( log.debug("[{}][{}] Fetched latest timeseries {}: {}", tenantId, entityId, timeseriesKey, result);
timeseriesService.findLatest(tenantId, entityId, timeseriesKey), return transformSingleValueArgument(result.orElseGet(() -> createDefaultTsKvEntry(argument, defaultTs)));
result -> { }, calculatedFieldCallbackExecutor);
log.debug("[{}][{}] Fetched latest timeseries {}: {}", tenantId, entityId, timeseriesKey, result);
return result.or(() -> Optional.of(new BasicTsKvEntry(defaultTs, createDefaultKvEntry(argument), SingleValueArgumentEntry.DEFAULT_VERSION)));
}, calculatedFieldCallbackExecutor));
} }
private ListenableFuture<ArgumentEntry> fetchTimeSeriesInternal(TenantId tenantId, EntityId entityId, ReadTsKvQuery query, Function<List<TsKvEntry>, ArgumentEntry> transformArgument) { private ListenableFuture<ArgumentEntry> fetchTimeSeriesInternal(TenantId tenantId, EntityId entityId, ReadTsKvQuery query, Function<List<TsKvEntry>, ArgumentEntry> transformArgument) {

14
application/src/main/java/org/thingsboard/server/service/cf/ctx/state/BaseCalculatedFieldState.java
View File

@ -41,6 +41,8 @@ import java.util.stream.Collectors;
@Getter @Getter
public abstract class BaseCalculatedFieldState implements CalculatedFieldState, Closeable { public abstract class BaseCalculatedFieldState implements CalculatedFieldState, Closeable {
protected static final long DEFAULT_LAST_UPDATE_TS = -1L;
protected final EntityId entityId; protected final EntityId entityId;
protected CalculatedFieldCtx ctx; protected CalculatedFieldCtx ctx;
protected TbActorRef actorCtx; protected TbActorRef actorCtx;
@ -48,7 +50,7 @@ public abstract class BaseCalculatedFieldState implements CalculatedFieldState,
protected Map<String, ArgumentEntry> arguments = new HashMap<>(); protected Map<String, ArgumentEntry> arguments = new HashMap<>();
protected boolean sizeExceedsLimit; protected boolean sizeExceedsLimit;
protected long latestTimestamp = -1; protected long latestTimestamp = DEFAULT_LAST_UPDATE_TS;
protected ReadinessStatus readinessStatus; protected ReadinessStatus readinessStatus;
@Setter @Setter
@ -119,7 +121,7 @@ public abstract class BaseCalculatedFieldState implements CalculatedFieldState,
requiredArguments = null; requiredArguments = null;
arguments.clear(); arguments.clear();
sizeExceedsLimit = false; sizeExceedsLimit = false;
latestTimestamp = -1; latestTimestamp = DEFAULT_LAST_UPDATE_TS;
} }
@Override @Override
@ -147,7 +149,7 @@ public abstract class BaseCalculatedFieldState implements CalculatedFieldState,
return valuesNode; return valuesNode;
} }
long latestTs = getLatestTimestamp(); long latestTs = getLatestTimestamp();
if (latestTs == -1) { if (latestTs == DEFAULT_LAST_UPDATE_TS) {
return valuesNode; return valuesNode;
} }
ObjectNode resultNode = JacksonUtil.newObjectNode(); ObjectNode resultNode = JacksonUtil.newObjectNode();
@ -165,12 +167,12 @@ public abstract class BaseCalculatedFieldState implements CalculatedFieldState,
newTs = (lastEntry != null) ? lastEntry.getKey() : System.currentTimeMillis(); newTs = (lastEntry != null) ? lastEntry.getKey() : System.currentTimeMillis();
} else if (entry instanceof RelatedEntitiesArgumentEntry relatedEntitiesArgumentEntry) { } else if (entry instanceof RelatedEntitiesArgumentEntry relatedEntitiesArgumentEntry) {
newTs = relatedEntitiesArgumentEntry.getEntityInputs().values().stream() newTs = relatedEntitiesArgumentEntry.getEntityInputs().values().stream()
.mapToLong(e -> (e instanceof SingleValueArgumentEntry s) ? s.getTs() : 0L) .mapToLong(e -> (e instanceof SingleValueArgumentEntry s) ? s.getTs() : DEFAULT_LAST_UPDATE_TS)
.max() .max()
.orElse(0L); .orElse(DEFAULT_LAST_UPDATE_TS);
} else if (entry instanceof GeofencingArgumentEntry geofencingArgumentEntry) { } else if (entry instanceof GeofencingArgumentEntry geofencingArgumentEntry) {
newTs = geofencingArgumentEntry.getZoneStates().values().stream() newTs = geofencingArgumentEntry.getZoneStates().values().stream()
.mapToLong(GeofencingZoneState::getTs).max().orElse(0L); .mapToLong(GeofencingZoneState::getTs).max().orElse(DEFAULT_LAST_UPDATE_TS);
} }
this.latestTimestamp = Math.max(this.latestTimestamp, newTs); this.latestTimestamp = Math.max(this.latestTimestamp, newTs);
} }

18
application/src/main/java/org/thingsboard/server/service/cf/ctx/state/CalculatedFieldCtx.java
View File

@ -79,10 +79,14 @@ import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors; import java.util.stream.Collectors;
import static org.thingsboard.server.service.cf.ctx.state.BaseCalculatedFieldState.DEFAULT_LAST_UPDATE_TS;
@Data @Data
@Slf4j @Slf4j
public class CalculatedFieldCtx implements Closeable { public class CalculatedFieldCtx implements Closeable {
public static final long DISABLED_INTERVAL_VALUE = -1L;
private CalculatedField calculatedField; private CalculatedField calculatedField;
private CalculatedFieldId cfId; private CalculatedFieldId cfId;
@ -197,7 +201,7 @@ public class CalculatedFieldCtx implements Closeable {
} }
} }
if (calculatedField.getConfiguration() instanceof ScheduledUpdateSupportedCalculatedFieldConfiguration scheduledConfig) { if (calculatedField.getConfiguration() instanceof ScheduledUpdateSupportedCalculatedFieldConfiguration scheduledConfig) {
this.scheduledUpdateIntervalMillis = scheduledConfig.isScheduledUpdateEnabled() ? TimeUnit.SECONDS.toMillis(scheduledConfig.getScheduledUpdateInterval()) : -1L; this.scheduledUpdateIntervalMillis = scheduledConfig.isScheduledUpdateEnabled() ? TimeUnit.SECONDS.toMillis(scheduledConfig.getScheduledUpdateInterval()) : DISABLED_INTERVAL_VALUE;
} }
if (calculatedField.getConfiguration() instanceof RelatedEntitiesAggregationCalculatedFieldConfiguration aggConfig) { if (calculatedField.getConfiguration() instanceof RelatedEntitiesAggregationCalculatedFieldConfiguration aggConfig) {
this.useLatestTs = aggConfig.isUseLatestTs(); this.useLatestTs = aggConfig.isUseLatestTs();
@ -715,8 +719,8 @@ public class CalculatedFieldCtx implements Closeable {
return false; return false;
} }
private boolean isScheduledUpdateEnabled() { private boolean isScheduledUpdateDisabled() {
return scheduledUpdateIntervalMillis != -1; return scheduledUpdateIntervalMillis == DISABLED_INTERVAL_VALUE;
} }
public boolean shouldFetchRelationQueryDynamicArgumentsFromDb(CalculatedFieldState state) { public boolean shouldFetchRelationQueryDynamicArgumentsFromDb(CalculatedFieldState state) {
@ -726,11 +730,11 @@ public class CalculatedFieldCtx implements Closeable {
return switch (cfType) { return switch (cfType) {
case PROPAGATION -> true; case PROPAGATION -> true;
case GEOFENCING -> { case GEOFENCING -> {
if (!isScheduledUpdateEnabled()) { if (isScheduledUpdateDisabled()) {
yield false; yield false;
} }
var geofencingState = (GeofencingCalculatedFieldState) state; var geofencingState = (GeofencingCalculatedFieldState) state;
if (geofencingState.getLastDynamicArgumentsRefreshTs() == -1L) { if (geofencingState.getLastDynamicArgumentsRefreshTs() == DEFAULT_LAST_UPDATE_TS) {
yield true; yield true;
} }
yield geofencingState.getLastDynamicArgumentsRefreshTs() < yield geofencingState.getLastDynamicArgumentsRefreshTs() <
@ -744,10 +748,10 @@ public class CalculatedFieldCtx implements Closeable {
if (!(state instanceof RelatedEntitiesAggregationCalculatedFieldState relatedEntitiesAggState)) { if (!(state instanceof RelatedEntitiesAggregationCalculatedFieldState relatedEntitiesAggState)) {
return false; return false;
} }
if (!isScheduledUpdateEnabled()) { if (isScheduledUpdateDisabled()) {
return false; return false;
} }
if (relatedEntitiesAggState.getLastRelatedEntitiesRefreshTs() == -1L) { if (relatedEntitiesAggState.getLastRelatedEntitiesRefreshTs() == DEFAULT_LAST_UPDATE_TS) {
return true; return true;
} }
return relatedEntitiesAggState.getLastRelatedEntitiesRefreshTs() < System.currentTimeMillis() - scheduledUpdateIntervalMillis; return relatedEntitiesAggState.getLastRelatedEntitiesRefreshTs() < System.currentTimeMillis() - scheduledUpdateIntervalMillis;

11
application/src/main/java/org/thingsboard/server/service/cf/ctx/state/SingleValueArgumentEntry.java
View File

@ -19,7 +19,6 @@ import com.fasterxml.jackson.annotation.JsonIgnore;
import com.fasterxml.jackson.core.type.TypeReference; import com.fasterxml.jackson.core.type.TypeReference;
import lombok.AllArgsConstructor; import lombok.AllArgsConstructor;
import lombok.Data; import lombok.Data;
import lombok.NoArgsConstructor;
import org.springframework.lang.Nullable; import org.springframework.lang.Nullable;
import org.thingsboard.common.util.JacksonUtil; import org.thingsboard.common.util.JacksonUtil;
import org.thingsboard.script.api.tbel.TbelCfArg; import org.thingsboard.script.api.tbel.TbelCfArg;
@ -34,11 +33,14 @@ import org.thingsboard.server.common.util.ProtoUtils;
import org.thingsboard.server.gen.transport.TransportProtos.AttributeValueProto; import org.thingsboard.server.gen.transport.TransportProtos.AttributeValueProto;
import org.thingsboard.server.gen.transport.TransportProtos.TsKvProto; import org.thingsboard.server.gen.transport.TransportProtos.TsKvProto;
import static org.thingsboard.server.service.cf.ctx.state.BaseCalculatedFieldState.DEFAULT_LAST_UPDATE_TS;
@Data @Data
@NoArgsConstructor
@AllArgsConstructor @AllArgsConstructor
public class SingleValueArgumentEntry implements ArgumentEntry { public class SingleValueArgumentEntry implements ArgumentEntry {
public static final Long DEFAULT_VERSION = -1L;
@Nullable @Nullable
protected EntityId entityId; protected EntityId entityId;
@ -48,7 +50,10 @@ public class SingleValueArgumentEntry implements ArgumentEntry {
protected boolean forceResetPrevious; protected boolean forceResetPrevious;
public static final Long DEFAULT_VERSION = -1L; public SingleValueArgumentEntry() {
this.ts = DEFAULT_LAST_UPDATE_TS;
this.version = DEFAULT_VERSION;
}
public SingleValueArgumentEntry(EntityId entityId, ArgumentEntry entry) { public SingleValueArgumentEntry(EntityId entityId, ArgumentEntry entry) {
this(entry); this(entry);

19
application/src/main/java/org/thingsboard/server/service/cf/ctx/state/aggregation/RelatedEntitiesAggregationCalculatedFieldState.java
View File

@ -51,18 +51,19 @@ import java.util.concurrent.ScheduledFuture;
import java.util.stream.Collectors; import java.util.stream.Collectors;
import static java.util.concurrent.TimeUnit.SECONDS; import static java.util.concurrent.TimeUnit.SECONDS;
import static org.thingsboard.server.service.cf.ctx.state.CalculatedFieldCtx.DISABLED_INTERVAL_VALUE;
@Slf4j @Slf4j
@Getter @Getter
public class RelatedEntitiesAggregationCalculatedFieldState extends BaseCalculatedFieldState { public class RelatedEntitiesAggregationCalculatedFieldState extends BaseCalculatedFieldState {
@Setter @Setter
private long lastArgsRefreshTs = -1; private long lastArgsRefreshTs = DEFAULT_LAST_UPDATE_TS;
@Setter @Setter
private long lastMetricsEvalTs = -1; private long lastMetricsEvalTs = DEFAULT_LAST_UPDATE_TS;
@Setter @Setter
private long lastRelatedEntitiesRefreshTs = -1; private long lastRelatedEntitiesRefreshTs = DEFAULT_LAST_UPDATE_TS;
private long deduplicationIntervalMs = -1; private long deduplicationIntervalMs = DISABLED_INTERVAL_VALUE;
private Map<String, AggMetric> metrics; private Map<String, AggMetric> metrics;
private ScheduledFuture<?> reevaluationFuture; private ScheduledFuture<?> reevaluationFuture;
@ -102,9 +103,9 @@ public class RelatedEntitiesAggregationCalculatedFieldState extends BaseCalculat
@Override @Override
public void reset() { // must reset everything dependent on arguments public void reset() { // must reset everything dependent on arguments
super.reset(); super.reset();
lastArgsRefreshTs = -1; lastArgsRefreshTs = DEFAULT_LAST_UPDATE_TS;
lastMetricsEvalTs = -1; lastMetricsEvalTs = DEFAULT_LAST_UPDATE_TS;
lastRelatedEntitiesRefreshTs = -1; lastRelatedEntitiesRefreshTs = DEFAULT_LAST_UPDATE_TS;
metrics = null; metrics = null;
} }
@ -153,7 +154,7 @@ public class RelatedEntitiesAggregationCalculatedFieldState extends BaseCalculat
} }
public Map<String, ArgumentEntry> updateEntityData(Map<String, ArgumentEntry> fetchedArgs) { public Map<String, ArgumentEntry> updateEntityData(Map<String, ArgumentEntry> fetchedArgs) {
lastMetricsEvalTs = -1; lastMetricsEvalTs = DEFAULT_LAST_UPDATE_TS;
return update(fetchedArgs, ctx); return update(fetchedArgs, ctx);
} }
@ -162,7 +163,7 @@ public class RelatedEntitiesAggregationCalculatedFieldState extends BaseCalculat
RelatedEntitiesArgumentEntry aggEntry = (RelatedEntitiesArgumentEntry) argEntry; RelatedEntitiesArgumentEntry aggEntry = (RelatedEntitiesArgumentEntry) argEntry;
aggEntry.getEntityInputs().remove(relatedEntityId); aggEntry.getEntityInputs().remove(relatedEntityId);
}); });
lastMetricsEvalTs = -1; lastMetricsEvalTs = DEFAULT_LAST_UPDATE_TS;
lastArgsRefreshTs = System.currentTimeMillis(); lastArgsRefreshTs = System.currentTimeMillis();
} }

6
application/src/main/java/org/thingsboard/server/service/cf/ctx/state/aggregation/single/EntityAggregationCalculatedFieldState.java
View File

@ -218,7 +218,7 @@ public class EntityAggregationCalculatedFieldState extends BaseCalculatedFieldSt
if (argEntryIntervalStatus.getLastArgsRefreshTs() > argEntryIntervalStatus.getLastMetricsEvalTs()) { if (argEntryIntervalStatus.getLastArgsRefreshTs() > argEntryIntervalStatus.getLastMetricsEvalTs()) {
argEntryIntervalStatus.setLastMetricsEvalTs(System.currentTimeMillis()); argEntryIntervalStatus.setLastMetricsEvalTs(System.currentTimeMillis());
processArgument(intervalEntry, argName, false, results); processArgument(intervalEntry, argName, false, results);
} else if (argEntryIntervalStatus.getLastMetricsEvalTs() == -1) { } else if (argEntryIntervalStatus.getLastMetricsEvalTs() == DEFAULT_LAST_UPDATE_TS) {
argEntryIntervalStatus.setLastMetricsEvalTs(System.currentTimeMillis()); argEntryIntervalStatus.setLastMetricsEvalTs(System.currentTimeMillis());
processArgument(intervalEntry, argName, true, results); processArgument(intervalEntry, argName, true, results);
} }
@ -232,9 +232,9 @@ public class EntityAggregationCalculatedFieldState extends BaseCalculatedFieldSt
if (argEntryIntervalStatus.intervalPassed(checkInterval)) { if (argEntryIntervalStatus.intervalPassed(checkInterval)) {
if (argEntryIntervalStatus.argsUpdated()) { if (argEntryIntervalStatus.argsUpdated()) {
argEntryIntervalStatus.setLastMetricsEvalTs(System.currentTimeMillis()); argEntryIntervalStatus.setLastMetricsEvalTs(System.currentTimeMillis());
argEntryIntervalStatus.setLastArgsRefreshTs(-1); argEntryIntervalStatus.setLastArgsRefreshTs(DEFAULT_LAST_UPDATE_TS);
processArgument(intervalEntry, argName, false, results); processArgument(intervalEntry, argName, false, results);
} else if (argEntryIntervalStatus.getLastMetricsEvalTs() == -1) { } else if (argEntryIntervalStatus.getLastMetricsEvalTs() == DEFAULT_LAST_UPDATE_TS) {
argEntryIntervalStatus.setLastMetricsEvalTs(System.currentTimeMillis()); argEntryIntervalStatus.setLastMetricsEvalTs(System.currentTimeMillis());
processArgument(intervalEntry, argName, true, results); processArgument(intervalEntry, argName, true, results);
} }

4
application/src/main/java/org/thingsboard/server/service/cf/ctx/state/geofencing/GeofencingCalculatedFieldState.java
View File

@ -58,7 +58,7 @@ import static org.thingsboard.server.common.data.cf.configuration.geofencing.Geo
@EqualsAndHashCode(callSuper = true) @EqualsAndHashCode(callSuper = true)
public class GeofencingCalculatedFieldState extends BaseCalculatedFieldState { public class GeofencingCalculatedFieldState extends BaseCalculatedFieldState {
private long lastDynamicArgumentsRefreshTs = -1; private long lastDynamicArgumentsRefreshTs = DEFAULT_LAST_UPDATE_TS;
public GeofencingCalculatedFieldState(EntityId entityId) { public GeofencingCalculatedFieldState(EntityId entityId) {
super(entityId); super(entityId);
@ -147,7 +147,7 @@ public class GeofencingCalculatedFieldState extends BaseCalculatedFieldState {
@Override @Override
public void reset() { public void reset() {
super.reset(); super.reset();
lastDynamicArgumentsRefreshTs = -1; lastDynamicArgumentsRefreshTs = DEFAULT_LAST_UPDATE_TS;
} }
public void updateLastDynamicArgumentsRefreshTs() { public void updateLastDynamicArgumentsRefreshTs() {

29
application/src/main/java/org/thingsboard/server/utils/CalculatedFieldArgumentUtils.java
View File

@ -15,9 +15,7 @@
*/ */
package org.thingsboard.server.utils; package org.thingsboard.server.utils;
import com.google.common.util.concurrent.Futures; import lombok.NonNull;
import com.google.common.util.concurrent.ListenableFuture;
import com.google.common.util.concurrent.MoreExecutors;
import org.apache.commons.lang3.math.NumberUtils; import org.apache.commons.lang3.math.NumberUtils;
import org.thingsboard.server.common.data.StringUtils; import org.thingsboard.server.common.data.StringUtils;
import org.thingsboard.server.common.data.cf.configuration.Argument; import org.thingsboard.server.common.data.cf.configuration.Argument;
@ -25,6 +23,7 @@ import org.thingsboard.server.common.data.cf.configuration.aggregation.AggMetric
import org.thingsboard.server.common.data.id.EntityId; import org.thingsboard.server.common.data.id.EntityId;
import org.thingsboard.server.common.data.kv.AttributeKvEntry; import org.thingsboard.server.common.data.kv.AttributeKvEntry;
import org.thingsboard.server.common.data.kv.BaseAttributeKvEntry; import org.thingsboard.server.common.data.kv.BaseAttributeKvEntry;
import org.thingsboard.server.common.data.kv.BasicTsKvEntry;
import org.thingsboard.server.common.data.kv.BooleanDataEntry; import org.thingsboard.server.common.data.kv.BooleanDataEntry;
import org.thingsboard.server.common.data.kv.DoubleDataEntry; import org.thingsboard.server.common.data.kv.DoubleDataEntry;
import org.thingsboard.server.common.data.kv.KvEntry; import org.thingsboard.server.common.data.kv.KvEntry;
@ -48,20 +47,13 @@ import org.thingsboard.server.service.cf.ctx.state.propagation.PropagationCalcul
import java.util.HashMap; import java.util.HashMap;
import java.util.List; import java.util.List;
import java.util.Map; import java.util.Map;
import java.util.Optional;
import static org.thingsboard.server.service.cf.ctx.state.SingleValueArgumentEntry.DEFAULT_VERSION;
public class CalculatedFieldArgumentUtils { public class CalculatedFieldArgumentUtils {
public static ListenableFuture<ArgumentEntry> transformSingleValueArgument(ListenableFuture<Optional<? extends KvEntry>> kvEntryFuture) { public static ArgumentEntry transformSingleValueArgument(@NonNull KvEntry kvEntry) {
return Futures.transform(kvEntryFuture, CalculatedFieldArgumentUtils::transformSingleValueArgument, MoreExecutors.directExecutor()); return kvEntry.getValue() != null ? ArgumentEntry.createSingleValueArgument(kvEntry) : new SingleValueArgumentEntry();
}
public static ArgumentEntry transformSingleValueArgument(Optional<? extends KvEntry> kvEntry) {
if (kvEntry.isPresent() && kvEntry.get().getValue() != null) {
return ArgumentEntry.createSingleValueArgument(kvEntry.get());
} else {
return new SingleValueArgumentEntry();
}
} }
public static ArgumentEntry transformTsRollingArgument(List<TsKvEntry> tsRolling, int limit, long argTimeWindow) { public static ArgumentEntry transformTsRollingArgument(List<TsKvEntry> tsRolling, int limit, long argTimeWindow) {
@ -94,7 +86,7 @@ public class CalculatedFieldArgumentUtils {
return new EntityAggregationArgumentEntry(aggIntervals); return new EntityAggregationArgumentEntry(aggIntervals);
} }
public static KvEntry createDefaultKvEntry(Argument argument) { private static KvEntry createDefaultKvEntry(Argument argument) {
String key = argument.getRefEntityKey().getKey(); String key = argument.getRefEntityKey().getKey();
String defaultValue = argument.getDefaultValue(); String defaultValue = argument.getDefaultValue();
if (StringUtils.isBlank(defaultValue)) { if (StringUtils.isBlank(defaultValue)) {
@ -109,9 +101,12 @@ public class CalculatedFieldArgumentUtils {
return new StringDataEntry(key, defaultValue); return new StringDataEntry(key, defaultValue);
} }
public static TsKvEntry createDefaultTsKvEntry(Argument argument, long ts) {
return new BasicTsKvEntry(ts, createDefaultKvEntry(argument), DEFAULT_VERSION);
}
public static AttributeKvEntry createDefaultAttributeEntry(Argument argument, long ts) { public static AttributeKvEntry createDefaultAttributeEntry(Argument argument, long ts) {
KvEntry kvEntry = createDefaultKvEntry(argument); return new BaseAttributeKvEntry(createDefaultKvEntry(argument), ts, DEFAULT_VERSION);
return new BaseAttributeKvEntry(kvEntry, ts, 0L);
} }
public static CalculatedFieldState createStateByType(CalculatedFieldCtx ctx, EntityId entityId) { public static CalculatedFieldState createStateByType(CalculatedFieldCtx ctx, EntityId entityId) {

Write Preview
Loading…
Cancel
Save