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

removed unnecessary fetch arguments methods

pull/14141/head
IrynaMatveieva 11 months ago
parent
2778f79e5e
commit
7bf7e0f994
4 changed files with 27 additions and 70 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Split View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 77
      application/src/main/java/org/thingsboard/server/service/cf/AbstractCalculatedFieldProcessingService.java
  2. 3
      application/src/main/java/org/thingsboard/server/service/cf/DefaultCalculatedFieldProcessingService.java
  3. 8
      application/src/main/java/org/thingsboard/server/service/cf/ctx/state/ArgumentEntry.java
  4. 9
      application/src/main/java/org/thingsboard/server/utils/CalculatedFieldArgumentUtils.java

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

@ -64,7 +64,6 @@ import static org.thingsboard.server.common.data.cf.configuration.geofencing.Ent
import static org.thingsboard.server.common.data.cf.configuration.geofencing.EntityCoordinates.ENTITY_ID_LONGITUDE_ARGUMENT_KEY;
import static org.thingsboard.server.utils.CalculatedFieldArgumentUtils.createDefaultAttributeEntry;
import static org.thingsboard.server.utils.CalculatedFieldArgumentUtils.createDefaultKvEntry;
import static org.thingsboard.server.utils.CalculatedFieldArgumentUtils.transformAggSingleArgument;
import static org.thingsboard.server.utils.CalculatedFieldArgumentUtils.transformSingleValueArgument;
@Data
@ -98,7 +97,7 @@ public abstract class AbstractCalculatedFieldProcessingService {
Map<String, ListenableFuture<ArgumentEntry>> argFutures = switch (ctx.getCfType()) {
case GEOFENCING -> fetchGeofencingCalculatedFieldArguments(ctx, entityId, false, ts);
case SIMPLE, SCRIPT, ALARM, PROPAGATION -> getBaseCalculatedFieldArguments(ctx, entityId, ts);
case RELATED_ENTITIES_AGGREGATION -> fetchAggArguments(ctx, entityId, ts);
case RELATED_ENTITIES_AGGREGATION -> fetchRelatedEntitiesAggArguments(ctx, entityId, ts);
};
if (ctx.getCfType() == PROPAGATION) {
argFutures.put(PROPAGATION_CONFIG_ARGUMENT, fetchPropagationCalculatedFieldArgument(ctx, entityId));
@ -128,23 +127,6 @@ public abstract class AbstractCalculatedFieldProcessingService {
return resolveOwnerArgument(tenantId, entityId);
}
private ListenableFuture<List<EntityId>> resolveRelatedEntities(TenantId tenantId, EntityId entityId, RelationPathLevel relation) {
ListenableFuture<List<EntityRelation>> relationsFut = relationService.findByRelationPathQueryAsync(tenantId, new EntityRelationPathQuery(entityId, List.of(relation)));
return Futures.transform(relationsFut, relations -> {
if (relations == null) {
return new ArrayList<>();
}
return switch (relation.direction()) {
case FROM -> relations.stream()
.map(EntityRelation::getTo)
.toList();
case TO -> relations.isEmpty() ? List.of() : List.of(relations.get(0).getFrom());
};
}, calculatedFieldCallbackExecutor);
}
protected Map<String, ArgumentEntry> resolveArgumentFutures(Map<String, ListenableFuture<ArgumentEntry>> argFutures) {
return argFutures.entrySet().stream()
.collect(Collectors.toMap(
@ -189,7 +171,7 @@ public abstract class AbstractCalculatedFieldProcessingService {
return argFutures;
}
protected Map<String, ListenableFuture<ArgumentEntry>> fetchAggArguments(CalculatedFieldCtx ctx, EntityId entityId, long ts) {
protected Map<String, ListenableFuture<ArgumentEntry>> fetchRelatedEntitiesAggArguments(CalculatedFieldCtx ctx, EntityId entityId, long ts) {
RelatedEntitiesAggregationCalculatedFieldConfiguration aggConfig = (RelatedEntitiesAggregationCalculatedFieldConfiguration) ctx.getCalculatedField().getConfiguration();
ListenableFuture<List<EntityId>> relatedEntitiesFut = resolveRelatedEntities(ctx.getTenantId(), entityId, aggConfig.getRelation());
@ -197,10 +179,27 @@ public abstract class AbstractCalculatedFieldProcessingService {
return aggConfig.getArguments().entrySet().stream()
.collect(Collectors.toMap(
Map.Entry::getKey,
entry -> Futures.transformAsync(relatedEntitiesFut, relatedEntities -> fetchAggArgumentEntry(ctx.getTenantId(), relatedEntities, entry.getValue(), ts), MoreExecutors.directExecutor())
entry -> Futures.transformAsync(relatedEntitiesFut, relatedEntities -> fetchRelatedEntitiesArgumentEntry(ctx.getTenantId(), relatedEntities, entry.getValue(), ts), MoreExecutors.directExecutor())
));
}
private ListenableFuture<List<EntityId>> resolveRelatedEntities(TenantId tenantId, EntityId entityId, RelationPathLevel relation) {
ListenableFuture<List<EntityRelation>> relationsFut = relationService.findByRelationPathQueryAsync(tenantId, new EntityRelationPathQuery(entityId, List.of(relation)));
return Futures.transform(relationsFut, relations -> {
if (relations == null) {
return new ArrayList<>();
}
return switch (relation.direction()) {
case FROM -> relations.stream()
.map(EntityRelation::getTo)
.toList();
case TO -> relations.isEmpty() ? List.of() : List.of(relations.get(0).getFrom());
};
}, calculatedFieldCallbackExecutor);
}
private ListenableFuture<List<EntityId>> resolveGeofencingEntityIds(TenantId tenantId, EntityId entityId, Map.Entry<String, Argument> entry) {
Argument value = entry.getValue();
if (value.getRefEntityId() != null) {
@ -252,11 +251,11 @@ public abstract class AbstractCalculatedFieldProcessingService {
.collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue))), MoreExecutors.directExecutor());
}
public ListenableFuture<ArgumentEntry> fetchAggArgumentEntry(TenantId tenantId, List<EntityId> aggEntities, Argument argument, long startTs) {
public ListenableFuture<ArgumentEntry> fetchRelatedEntitiesArgumentEntry(TenantId tenantId, List<EntityId> aggEntities, Argument argument, long startTs) {
List<ListenableFuture<Map.Entry<EntityId, ArgumentEntry>>> futures = aggEntities.stream()
.map(entityId -> {
ListenableFuture<ArgumentEntry> singleAggEntryFut = fetchSingleAggArgumentEntry(tenantId, entityId, argument, startTs);
return Futures.transform(singleAggEntryFut, singleAggEntry -> Map.entry(entityId, singleAggEntry), MoreExecutors.directExecutor());
ListenableFuture<ArgumentEntry> argumentEntryFut = fetchArgumentValue(tenantId, entityId, argument, startTs);
return Futures.transform(argumentEntryFut, argumentEntry -> Map.entry(entityId, ArgumentEntry.createAggSingleArgument(entityId, argumentEntry)), MoreExecutors.directExecutor());
})
.toList();
@ -321,34 +320,4 @@ public abstract class AbstractCalculatedFieldProcessingService {
return new BaseReadTsKvQuery(argument.getRefEntityKey().getKey(), startTs, endTs, 0, limit, Aggregation.NONE);
}
private ListenableFuture<ArgumentEntry> fetchSingleAggArgumentEntry(TenantId tenantId, EntityId entityId, Argument argument, long startTs) {
return switch (argument.getRefEntityKey().getType()) {
case TS_ROLLING -> throw new IllegalStateException("TS_ROLLING is not supported for aggregation");
case ATTRIBUTE -> fetchAttributeAggEntry(tenantId, entityId, argument, startTs);
case TS_LATEST -> fetchTsLatestAggEntry(tenantId, entityId, argument, startTs);
};
}
private ListenableFuture<ArgumentEntry> fetchAttributeAggEntry(TenantId tenantId, EntityId entityId, Argument argument, long defaultLastUpdateTs) {
log.trace("[{}][{}] Fetching attribute for key {}", tenantId, entityId, argument.getRefEntityKey());
var attributeOptFuture = attributesService.find(tenantId, entityId, argument.getRefEntityKey().getScope(), argument.getRefEntityKey().getKey());
return Futures.transform(attributeOptFuture, 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 transformAggSingleArgument(entityId, Optional.of(attributeKvEntry));
}, calculatedFieldCallbackExecutor);
}
private ListenableFuture<ArgumentEntry> fetchTsLatestAggEntry(TenantId tenantId, EntityId entityId, Argument argument, long defaultTs) {
String key = argument.getRefEntityKey().getKey();
log.trace("[{}][{}] Fetching latest timeseries {}", tenantId, entityId, key);
return Futures.transform(
timeseriesService.findLatest(tenantId, entityId, key),
result -> {
log.debug("[{}][{}] Fetched latest timeseries {}: {}", tenantId, entityId, key, result);
Optional<TsKvEntry> tsKvEntry = result.or(() -> Optional.of(new BasicTsKvEntry(defaultTs, createDefaultKvEntry(argument), SingleValueArgumentEntry.DEFAULT_VERSION)));
return transformAggSingleArgument(entityId, tsKvEntry);
}, calculatedFieldCallbackExecutor);
}
}

3
application/src/main/java/org/thingsboard/server/service/cf/DefaultCalculatedFieldProcessingService.java
View File

@ -15,9 +15,7 @@
*/
package org.thingsboard.server.service.cf;
import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.ListenableFuture;
import com.google.common.util.concurrent.MoreExecutors;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;
import org.thingsboard.server.actors.calculatedField.CalculatedFieldTelemetryMsg;
@ -56,7 +54,6 @@ import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.UUID;
import java.util.stream.Collectors;
import static org.thingsboard.server.common.data.cf.configuration.PropagationCalculatedFieldConfiguration.PROPAGATION_CONFIG_ARGUMENT;
import static org.thingsboard.server.utils.CalculatedFieldUtils.toProto;

8
application/src/main/java/org/thingsboard/server/service/cf/ctx/state/ArgumentEntry.java
View File

@ -22,8 +22,8 @@ import org.thingsboard.script.api.tbel.TbelCfArg;
import org.thingsboard.server.common.data.id.EntityId;
import org.thingsboard.server.common.data.kv.KvEntry;
import org.thingsboard.server.common.data.kv.TsKvEntry;
import org.thingsboard.server.service.cf.ctx.state.aggregation.RelatedEntitiesArgumentEntry;
import org.thingsboard.server.service.cf.ctx.state.aggregation.AggSingleEntityArgumentEntry;
import org.thingsboard.server.service.cf.ctx.state.aggregation.RelatedEntitiesArgumentEntry;
import org.thingsboard.server.service.cf.ctx.state.geofencing.GeofencingArgumentEntry;
import org.thingsboard.server.service.cf.ctx.state.propagation.PropagationArgumentEntry;
@ -40,7 +40,7 @@ import java.util.Map;
@JsonSubTypes.Type(value = TsRollingArgumentEntry.class, name = "TS_ROLLING"),
@JsonSubTypes.Type(value = GeofencingArgumentEntry.class, name = "GEOFENCING"),
@JsonSubTypes.Type(value = PropagationArgumentEntry.class, name = "PROPAGATION"),
@JsonSubTypes.Type(value = RelatedEntitiesArgumentEntry.class, name = "AGGREGATE_LATEST"),
@JsonSubTypes.Type(value = RelatedEntitiesArgumentEntry.class, name = "RELATED_ENTITIES"),
@JsonSubTypes.Type(value = AggSingleEntityArgumentEntry.class, name = "AGGREGATE_LATEST_SINGLE")
})
public interface ArgumentEntry {
@ -80,8 +80,8 @@ public interface ArgumentEntry {
return new RelatedEntitiesArgumentEntry(entityIdkvEntryMap, false);
}
static ArgumentEntry createAggSingleArgument(EntityId entityId, KvEntry kvEntry) {
return new AggSingleEntityArgumentEntry(entityId, kvEntry);
static ArgumentEntry createAggSingleArgument(EntityId entityId, ArgumentEntry argumentEntry) {
return new AggSingleEntityArgumentEntry(entityId, argumentEntry);
}
}

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

@ -34,7 +34,6 @@ import org.thingsboard.server.service.cf.ctx.state.CalculatedFieldState;
import org.thingsboard.server.service.cf.ctx.state.ScriptCalculatedFieldState;
import org.thingsboard.server.service.cf.ctx.state.SimpleCalculatedFieldState;
import org.thingsboard.server.service.cf.ctx.state.SingleValueArgumentEntry;
import org.thingsboard.server.service.cf.ctx.state.aggregation.AggSingleEntityArgumentEntry;
import org.thingsboard.server.service.cf.ctx.state.aggregation.RelatedEntitiesAggregationCalculatedFieldState;
import org.thingsboard.server.service.cf.ctx.state.alarm.AlarmCalculatedFieldState;
import org.thingsboard.server.service.cf.ctx.state.geofencing.GeofencingCalculatedFieldState;
@ -56,14 +55,6 @@ public class CalculatedFieldArgumentUtils {
}
}
public static ArgumentEntry transformAggSingleArgument(EntityId entityId, Optional<? extends KvEntry> kvEntry) {
if (kvEntry.isPresent() && kvEntry.get().getValue() != null) {
return ArgumentEntry.createAggSingleArgument(entityId, kvEntry.get());
} else {
return new AggSingleEntityArgumentEntry();
}
}
public static KvEntry createDefaultKvEntry(Argument argument) {
String key = argument.getRefEntityKey().getKey();
String defaultValue = argument.getDefaultValue();

Write Preview
Loading…
Cancel
Save