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

Fix bugs. Remove floating window support

pull/7288/head
Andrii Shvaika 4 years ago
parent
c6c40f50a8
commit
9689e9c041
5 changed files with 45 additions and 34 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. 71
      application/src/main/java/org/thingsboard/server/service/subscription/DefaultTbEntityDataSubscriptionService.java
  2. 1
      application/src/main/java/org/thingsboard/server/service/telemetry/cmd/v2/AggTimeSeriesCmd.java
  3. 3
      common/data/src/main/java/org/thingsboard/server/common/data/query/EntityData.java
  4. 2
      common/data/src/main/java/org/thingsboard/server/common/data/query/EntityKey.java
  5. 2
      dao/src/main/java/org/thingsboard/server/dao/sql/query/EntityDataAdapter.java

71
application/src/main/java/org/thingsboard/server/service/subscription/DefaultTbEntityDataSubscriptionService.java
View File

@ -70,6 +70,7 @@ import javax.annotation.PreDestroy;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashMap;
import java.util.HashSet;
import java.util.LinkedHashSet;
import java.util.List;
import java.util.Map;
@ -209,20 +210,20 @@ public class DefaultTbEntityDataSubscriptionService implements TbEntityDataSubsc
}
}
if (cmd.getAggHistoryCmd() != null) {
handleAggHistoryCmd(session, ctx, cmd.getAggHistoryCmd());
handleAggHistoryCmd(ctx, cmd.getAggHistoryCmd(), cmd.getLatestCmd());
} else if (cmd.getAggTsCmd() != null) {
handleAggTsCmd(session, ctx, cmd.getAggTsCmd());
handleAggTsCmd(ctx, cmd.getAggTsCmd(), cmd.getLatestCmd());
} else if (cmd.hasRegularCmds()) {
handleRegularCommands(session, ctx, cmd);
handleRegularCommands(ctx, cmd);
} else {
checkAndSendInitialData(ctx);
}
}
private void handleRegularCommands(TelemetryWebSocketSessionRef session, TbEntityDataSubCtx ctx, EntityDataCmd cmd) {
private void handleRegularCommands(TbEntityDataSubCtx ctx, EntityDataCmd cmd) {
ListenableFuture<TbEntityDataSubCtx> historyFuture;
if (cmd.getHistoryCmd() != null) {
log.trace("[{}][{}] Going to process history command: {}", session.getSessionId(), cmd.getCmdId(), cmd.getHistoryCmd());
log.trace("[{}][{}] Going to process history command: {}", ctx.getSessionId(), cmd.getCmdId(), cmd.getHistoryCmd());
try {
historyFuture = handleHistoryCmd(ctx, cmd.getHistoryCmd());
} catch (RuntimeException e) {
@ -253,7 +254,7 @@ public class DefaultTbEntityDataSubscriptionService implements TbEntityDataSubsc
@Override
public void onFailure(Throwable t) {
log.warn("[{}][{}] Failed to process command", session.getSessionId(), cmd.getCmdId());
log.warn("[{}][{}] Failed to process command", ctx.getSessionId(), cmd.getCmdId());
}
}, wsCallBackExecutor);
}
@ -266,34 +267,30 @@ public class DefaultTbEntityDataSubscriptionService implements TbEntityDataSubsc
}
}
private void handleAggHistoryCmd(TelemetryWebSocketSessionRef session, TbEntityDataSubCtx ctx, AggHistoryCmd cmd) {
private void handleAggHistoryCmd(TbEntityDataSubCtx ctx, AggHistoryCmd cmd, LatestValueCmd latestCmd) {
var keys = cmd.getKeys();
long interval = cmd.getEndTs() - cmd.getStartTs();
List<ReadTsKvQuery> queries = keys.stream().map(key -> new BaseReadTsKvQuery(
key.getKey(), cmd.getStartTs(), cmd.getEndTs(), interval, 1, key.getAgg()
)).distinct().collect(Collectors.toList());
handleAggCmd(session, ctx, cmd.getKeys(), queries, cmd.getStartTs(), cmd.getEndTs(), false, false);
handleAggCmd(ctx, cmd.getKeys(), queries, cmd.getStartTs(), cmd.getEndTs(), latestCmd, false);
}
private void handleAggTsCmd(TelemetryWebSocketSessionRef session, TbEntityDataSubCtx ctx, AggTimeSeriesCmd cmd) {
private void handleAggTsCmd(TbEntityDataSubCtx ctx, AggTimeSeriesCmd cmd, LatestValueCmd latestCmd) {
long endTs = cmd.getStartTs() + cmd.getTimeWindow();
List<ReadTsKvQuery> queries = cmd.getKeys().stream().map(key -> {
if (cmd.isFloating()) {
return new BaseReadTsKvQuery(key.getKey(), cmd.getStartTs(), endTs, cmd.getTimeWindow(), getLimit(maxDatapointLimit), Aggregation.NONE);
} else {
return new BaseReadTsKvQuery(key.getKey(), cmd.getStartTs(), endTs, cmd.getTimeWindow(), 1, key.getAgg());
}
}).distinct().collect(Collectors.toList());
handleAggCmd(session, ctx, cmd.getKeys(), queries, cmd.getStartTs(), endTs, cmd.isFloating(), true);
List<ReadTsKvQuery> queries = cmd.getKeys().stream()
.map(key -> new BaseReadTsKvQuery(key.getKey(), cmd.getStartTs(), endTs, cmd.getTimeWindow(), 1, key.getAgg()))
.distinct().collect(Collectors.toList());
handleAggCmd(ctx, cmd.getKeys(), queries, cmd.getStartTs(), endTs, latestCmd, true);
}
private void handleAggCmd(TelemetryWebSocketSessionRef session, TbEntityDataSubCtx ctx, List<AggKey> keys, List<ReadTsKvQuery> queries,
long startTs, long endTs, boolean floating, boolean subscribe) {
private void handleAggCmd(TbEntityDataSubCtx ctx, List<AggKey> keys, List<ReadTsKvQuery> queries,
long startTs, long endTs, LatestValueCmd latestCmd, boolean subscribe) {
Map<EntityData, ListenableFuture<List<ReadTsKvQueryResult>>> fetchResultMap = new HashMap<>();
List<EntityData> entityDataList = ctx.getData().getData();
entityDataList.forEach(entityData -> fetchResultMap.put(entityData,
tsService.findAllByQueries(ctx.getTenantId(), entityData.getEntityId(), queries)));
Futures.transform(Futures.allAsList(fetchResultMap.values()), f -> {
var mainFuture = Futures.transform(Futures.allAsList(fetchResultMap.values()), f -> {
// Map that holds last ts for each key for each entity.
Map<EntityData, Map<String, Long>> lastTsEntityMap = new HashMap<>();
fetchResultMap.forEach((entityData, future) -> {
@ -304,21 +301,13 @@ public class DefaultTbEntityDataSubscriptionService implements TbEntityDataSubsc
List<ReadTsKvQueryResult> queryResults = future.get();
if (queryResults != null) {
for (ReadTsKvQueryResult queryResult : queryResults) {
if (floating) {
entityData.getAggFloating().put(queryResult.getKey(), queryResult.toTsValues());
} else {
entityData.getAggLatest().computeIfAbsent(queryResult.getAgg(), agg -> new HashMap<>()).put(queryResult.getKey(), queryResult.toTsValue());
}
entityData.getAggLatest().computeIfAbsent(queryResult.getAgg(), agg -> new HashMap<>()).put(queryResult.getKey(), queryResult.toTsValue());
lastTsMap.put(queryResult.getKey(), queryResult.getLastEntryTs());
}
}
// Populate with empty values if no data found.
keys.forEach(key -> {
if (floating) {
entityData.getAggFloating().putIfAbsent(key.getKey(), new TsValue[]{TsValue.EMPTY});
} else {
entityData.getAggLatest().computeIfAbsent(key.getAgg(), agg -> new HashMap<>()).putIfAbsent(key.getKey(), TsValue.EMPTY);
}
entityData.getAggLatest().computeIfAbsent(key.getAgg(), agg -> new HashMap<>()).putIfAbsent(key.getKey(), TsValue.EMPTY);
});
} catch (InterruptedException | ExecutionException e) {
log.warn("[{}][{}][{}] Failed to fetch historical data", ctx.getSessionId(), ctx.getCmdId(), entityData.getEntityId(), e);
@ -344,6 +333,28 @@ public class DefaultTbEntityDataSubscriptionService implements TbEntityDataSubsc
}
return ctx;
}, wsCallBackExecutor);
Futures.addCallback(mainFuture, new FutureCallback<>() {
@Override
public void onSuccess(@Nullable TbEntityDataSubCtx theCtx) {
if (latestCmd != null) {
if (subscribe) {
var commandEntityKeys = new HashSet<>(latestCmd.getKeys());
var alreadySubscribedKeys = keys.stream().map(key -> new EntityKey(EntityKeyType.TIME_SERIES, key.getKey())).collect(Collectors.toList());
if (commandEntityKeys.removeAll(alreadySubscribedKeys)) {
latestCmd.setKeys(new ArrayList<>(commandEntityKeys));
}
}
handleLatestCmd(ctx, latestCmd);
}
}
@Override
public void onFailure(Throwable t) {
log.warn("[{}][{}] Failed to process command", ctx.getSessionId(), ctx.getCmdId());
}
}, wsCallBackExecutor);
}
private void handleWsCmdRuntimeException(String sessionId, RuntimeException e, EntityDataCmd cmd) {

1
application/src/main/java/org/thingsboard/server/service/telemetry/cmd/v2/AggTimeSeriesCmd.java
View File

@ -25,6 +25,5 @@ public class AggTimeSeriesCmd {
private List<AggKey> keys;
private long startTs;
private long timeWindow;
private boolean floating;
}

3
common/data/src/main/java/org/thingsboard/server/common/data/query/EntityData.java
View File

@ -31,9 +31,8 @@ public class EntityData {
private final Map<EntityKeyType, Map<String, TsValue>> latest;
private final Map<String, TsValue[]> timeseries;
private final Map<Aggregation, Map<String, TsValue>> aggLatest;
private final Map<String, TsValue[]> aggFloating;
public EntityData(EntityId entityId, Map<EntityKeyType, Map<String, TsValue>> latest, Map<String, TsValue[]> timeseries) {
this(entityId, latest, timeseries, null, null);
this(entityId, latest, timeseries, null);
}
}

2
common/data/src/main/java/org/thingsboard/server/common/data/query/EntityKey.java
View File

@ -23,6 +23,8 @@ import java.io.Serializable;
@ApiModel
@Data
public class EntityKey implements Serializable {
private static final long serialVersionUID = -6421575477523085543L;
private final EntityKeyType type;
private final String key;
}

2
dao/src/main/java/org/thingsboard/server/dao/sql/query/EntityDataAdapter.java
View File

@ -55,7 +55,7 @@ public class EntityDataAdapter {
EntityId entityId = EntityIdFactory.getByTypeAndUuid(entityType, id);
Map<EntityKeyType, Map<String, TsValue>> latest = new HashMap<>();
//Maybe avoid empty hashmaps?
EntityData entityData = new EntityData(entityId, latest, new HashMap<>(), new HashMap<>(), new HashMap<>());
EntityData entityData = new EntityData(entityId, latest, new HashMap<>(), new HashMap<>());
for (EntityKeyMapping mapping : selectionMapping) {
if (!mapping.isIgnore()) {
EntityKey entityKey = mapping.getEntityKey();

Write Preview
Loading…
Cancel
Save