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

Improvements to startTime and endTime for subscriptions

pull/4257/head
Andrii Shvaika 6 years ago
parent
60c141d3ca
commit
b4184d014c
8 changed files with 80 additions and 60 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. 57
      application/src/main/java/org/thingsboard/server/service/subscription/DefaultSubscriptionManagerService.java
  2. 10
      application/src/main/java/org/thingsboard/server/service/subscription/DefaultTbEntityDataSubscriptionService.java
  3. 25
      application/src/main/java/org/thingsboard/server/service/subscription/TbAbstractDataSubCtx.java
  4. 6
      application/src/main/java/org/thingsboard/server/service/subscription/TbAlarmDataSubCtx.java
  5. 34
      application/src/main/java/org/thingsboard/server/service/subscription/TbEntityDataSubCtx.java
  6. 2
      application/src/main/java/org/thingsboard/server/service/subscription/TbSubscriptionUtils.java
  7. 5
      application/src/main/java/org/thingsboard/server/service/subscription/TbTimeseriesSubscription.java
  8. 1
      common/queue/src/main/proto/queue.proto

57
application/src/main/java/org/thingsboard/server/service/subscription/DefaultSubscriptionManagerService.java
View File

@ -213,7 +213,7 @@ public class DefaultSubscriptionManagerService extends TbApplicationEventListene
}, s -> true, s -> {
List<TsKvEntry> subscriptionUpdate = null;
for (TsKvEntry kv : ts) {
if (isInTimeRange(s, kv.getTs()) && (s.isAllKeys() || s.getKeyStates().containsKey((kv.getKey())))) {
if ((s.isAllKeys() || s.getKeyStates().containsKey((kv.getKey())))) {
if (subscriptionUpdate == null) {
subscriptionUpdate = new ArrayList<>();
}
@ -375,11 +375,6 @@ public class DefaultSubscriptionManagerService extends TbApplicationEventListene
}
}
private boolean isInTimeRange(TbTimeseriesSubscription subscription, long kvTime) {
return (subscription.getStartTime() == 0 || subscription.getStartTime() <= kvTime)
&& (subscription.getEndTime() == 0 || subscription.getEndTime() >= kvTime);
}
private void removeSubscriptionFromEntityMap(TbSubscription sub) {
Set<TbSubscription> entitySubSet = subscriptionsByEntityId.get(sub.getEntityId());
if (entitySubSet != null) {
@ -429,18 +424,9 @@ public class DefaultSubscriptionManagerService extends TbApplicationEventListene
serviceId, subscription.getSessionId(), subscription.getSubscriptionId(), subscription.getEntityId());
long curTs = System.currentTimeMillis();
List<ReadTsKvQuery> queries = new ArrayList<>();
subscription.getKeyStates().forEach((key, value) -> {
if (curTs > value) {
long startTs = subscription.getStartTime() > 0 ? Math.max(subscription.getStartTime(), value + 1L) : (value + 1L);
long endTs = subscription.getEndTime() > 0 ? Math.min(subscription.getEndTime(), curTs) : curTs;
if (startTs > 1) {
queries.add(new BaseReadTsKvQuery(key, startTs, endTs, 0, 1000, Aggregation.NONE));
}
}
});
if (!queries.isEmpty()) {
DonAsynchron.withCallback(tsService.findAll(subscription.getTenantId(), subscription.getEntityId(), queries),
if (subscription.isLatestValues()) {
DonAsynchron.withCallback(tsService.findLatest(subscription.getTenantId(), subscription.getEntityId(), subscription.getKeyStates().keySet()),
missedUpdates -> {
if (missedUpdates != null && !missedUpdates.isEmpty()) {
TopicPartitionInfo tpi = partitionService.getNotificationsTopic(ServiceType.TB_CORE, subscription.getServiceId());
@ -449,6 +435,26 @@ public class DefaultSubscriptionManagerService extends TbApplicationEventListene
},
e -> log.error("Failed to fetch missed updates.", e),
tsCallBackExecutor);
} else {
List<ReadTsKvQuery> queries = new ArrayList<>();
subscription.getKeyStates().forEach((key, value) -> {
if (curTs > value) {
long startTs = subscription.getStartTime() > 0 ? Math.max(subscription.getStartTime(), value + 1L) : (value + 1L);
long endTs = subscription.getEndTime() > 0 ? Math.min(subscription.getEndTime(), curTs) : curTs;
queries.add(new BaseReadTsKvQuery(key, startTs, endTs, 0, 1000, Aggregation.NONE));
}
});
if (!queries.isEmpty()) {
DonAsynchron.withCallback(tsService.findAll(subscription.getTenantId(), subscription.getEntityId(), queries),
missedUpdates -> {
if (missedUpdates != null && !missedUpdates.isEmpty()) {
TopicPartitionInfo tpi = partitionService.getNotificationsTopic(ServiceType.TB_CORE, subscription.getServiceId());
toCoreNotificationsProducer.send(tpi, toProto(subscription, missedUpdates), null);
}
},
e -> log.error("Failed to fetch missed updates.", e),
tsCallBackExecutor);
}
}
}
@ -470,12 +476,19 @@ public class DefaultSubscriptionManagerService extends TbApplicationEventListene
data.forEach((key, value) -> {
TbSubscriptionUpdateValueListProto.Builder dataBuilder = TbSubscriptionUpdateValueListProto.newBuilder();
dataBuilder.setKey(key);
value.forEach(v -> {
boolean hasData = false;
for (Object v : value) {
Object[] array = (Object[]) v;
dataBuilder.addTs((long) array[0]);
dataBuilder.addValue((String) array[1]);
});
builder.addData(dataBuilder.build());
String strVal = (String) array[1];
if (strVal != null) {
hasData = true;
dataBuilder.addValue(strVal);
}
}
if (hasData) {
builder.addData(dataBuilder.build());
}
});
ToCoreNotificationMsg toCoreMsg = ToCoreNotificationMsg.newBuilder().setToLocalSubscriptionServiceMsg(

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

@ -215,7 +215,7 @@ public class DefaultTbEntityDataSubscriptionService implements TbEntityDataSubsc
} else {
historyFuture = Futures.immediateFuture(ctx);
}
Futures.addCallback(historyFuture, new FutureCallback<TbEntityDataSubCtx>() {
Futures.addCallback(historyFuture, new FutureCallback<>() {
@Override
public void onSuccess(@Nullable TbEntityDataSubCtx theCtx) {
if (cmd.getLatestCmd() != null) {
@ -278,7 +278,7 @@ public class DefaultTbEntityDataSubscriptionService implements TbEntityDataSubsc
wsService.sendWsMsg(ctx.getSessionId(), update);
} else {
ctx.fetchAlarms();
ctx.createSubscriptions(cmd.getQuery().getLatestValues(), true);
ctx.createLatestValuesSubscriptions(cmd.getQuery().getLatestValues());
if (adq.getPageLink().getTimeWindow() > 0) {
TbAlarmDataSubCtx finalCtx = ctx;
ScheduledFuture<?> task = scheduler.scheduleWithFixedDelay(
@ -419,7 +419,7 @@ public class DefaultTbEntityDataSubscriptionService implements TbEntityDataSubsc
}
wsService.sendWsMsg(ctx.getSessionId(), update);
if (subscribe) {
ctx.createSubscriptions(keys.stream().map(key -> new EntityKey(EntityKeyType.TIME_SERIES, key)).collect(Collectors.toList()), false);
ctx.createTimeseriesSubscriptions(keys.stream().map(key -> new EntityKey(EntityKeyType.TIME_SERIES, key)).collect(Collectors.toList()), cmd.getStartTs(), cmd.getEndTs());
}
ctx.getData().getData().forEach(ed -> ed.getTimeseries().clear());
return ctx;
@ -468,7 +468,7 @@ public class DefaultTbEntityDataSubscriptionService implements TbEntityDataSubsc
update = new EntityDataUpdate(ctx.getCmdId(), null, ctx.getData().getData(), ctx.getMaxEntitiesPerDataSubscription());
}
wsService.sendWsMsg(ctx.getSessionId(), update);
ctx.createSubscriptions(latestCmd.getKeys(), true);
ctx.createLatestValuesSubscriptions(latestCmd.getKeys());
}
@Override
@ -484,7 +484,7 @@ public class DefaultTbEntityDataSubscriptionService implements TbEntityDataSubsc
wsService.sendWsMsg(ctx.getSessionId(), update);
ctx.setInitialDataSent(true);
}
ctx.createSubscriptions(latestCmd.getKeys(), true);
ctx.createLatestValuesSubscriptions(latestCmd.getKeys());
}
}

25
application/src/main/java/org/thingsboard/server/service/subscription/TbAbstractDataSubCtx.java
View File

@ -115,10 +115,18 @@ public abstract class TbAbstractDataSubCtx<T extends AbstractDataQuery<? extends
}
}
public void createSubscriptions(List<EntityKey> keys, boolean resultToLatestValues) {
public void createLatestValuesSubscriptions(List<EntityKey> keys) {
createSubscriptions(keys, true, 0, 0);
}
public void createTimeseriesSubscriptions(List<EntityKey> keys, long startTs, long endTs) {
createSubscriptions(keys, false, startTs, endTs);
}
private void createSubscriptions(List<EntityKey> keys, boolean latestValues, long startTs, long endTs) {
Map<EntityKeyType, List<EntityKey>> keysByType = getEntityKeyByTypeMap(keys);
for (EntityData entityData : data.getData()) {
List<TbSubscription> entitySubscriptions = addSubscriptions(entityData, keysByType, resultToLatestValues);
List<TbSubscription> entitySubscriptions = addSubscriptions(entityData, keysByType, latestValues, startTs, endTs);
entitySubscriptions.forEach(localSubscriptionService::addSubscription);
}
}
@ -129,14 +137,14 @@ public abstract class TbAbstractDataSubCtx<T extends AbstractDataQuery<? extends
return keysByType;
}
protected List<TbSubscription> addSubscriptions(EntityData entityData, Map<EntityKeyType, List<EntityKey>> keysByType, boolean resultToLatestValues) {
protected List<TbSubscription> addSubscriptions(EntityData entityData, Map<EntityKeyType, List<EntityKey>> keysByType, boolean latestValues, long startTs, long endTs) {
List<TbSubscription> subscriptionList = new ArrayList<>();
keysByType.forEach((keysType, keysList) -> {
int subIdx = sessionRef.getSessionSubIdSeq().incrementAndGet();
subToEntityIdMap.put(subIdx, entityData.getEntityId());
switch (keysType) {
case TIME_SERIES:
subscriptionList.add(createTsSub(entityData, subIdx, keysList, resultToLatestValues));
subscriptionList.add(createTsSub(entityData, subIdx, keysList, latestValues, startTs, endTs));
break;
case CLIENT_ATTRIBUTE:
subscriptionList.add(createAttrSub(entityData, subIdx, keysType, TbAttributeSubscriptionScope.CLIENT_SCOPE, keysList));
@ -171,9 +179,9 @@ public abstract class TbAbstractDataSubCtx<T extends AbstractDataQuery<? extends
.build();
}
private TbSubscription createTsSub(EntityData entityData, int subIdx, List<EntityKey> subKeys, boolean resultToLatestValues) {
private TbSubscription createTsSub(EntityData entityData, int subIdx, List<EntityKey> subKeys, boolean latestValues, long startTs, long endTs) {
Map<String, Long> keyStates = buildKeyStats(entityData, EntityKeyType.TIME_SERIES, subKeys);
if (entityData.getTimeseries() != null) {
if (!latestValues && entityData.getTimeseries() != null) {
entityData.getTimeseries().forEach((k, v) -> {
long ts = Arrays.stream(v).map(TsValue::getTs).max(Long::compareTo).orElse(0L);
log.trace("[{}][{}] Updating key: {} with ts: {}", serviceId, cmdId, k, ts);
@ -187,9 +195,12 @@ public abstract class TbAbstractDataSubCtx<T extends AbstractDataQuery<? extends
.subscriptionId(subIdx)
.tenantId(sessionRef.getSecurityCtx().getTenantId())
.entityId(entityData.getEntityId())
.updateConsumer((sessionId, subscriptionUpdate) -> sendWsMsg(sessionId, subscriptionUpdate, EntityKeyType.TIME_SERIES, resultToLatestValues))
.updateConsumer((sessionId, subscriptionUpdate) -> sendWsMsg(sessionId, subscriptionUpdate, EntityKeyType.TIME_SERIES, latestValues))
.allKeys(false)
.keyStates(keyStates)
.latestValues(latestValues)
.startTime(startTs)
.endTime(endTs)
.build();
}

6
application/src/main/java/org/thingsboard/server/service/subscription/TbAlarmDataSubCtx.java
View File

@ -132,8 +132,8 @@ public class TbAlarmDataSubCtx extends TbAbstractDataSubCtx<AlarmDataQuery> {
}
@Override
public void createSubscriptions(List<EntityKey> keys, boolean resultToLatestValues) {
super.createSubscriptions(keys, resultToLatestValues);
public void createLatestValuesSubscriptions(List<EntityKey> keys) {
super.createLatestValuesSubscriptions(keys);
createAlarmSubscriptions();
}
@ -282,7 +282,7 @@ public class TbAlarmDataSubCtx extends TbAbstractDataSubCtx<AlarmDataQuery> {
newSubsList.forEach(
entity -> {
log.trace("[{}][{}] Found new subscription for entity: {}", sessionRef.getSessionId(), cmdId, entity.getEntityId());
subsToAdd.addAll(addSubscriptions(entity, keysByType, true));
subsToAdd.addAll(addSubscriptions(entity, keysByType, true, 0, 0));
}
);
}

34
application/src/main/java/org/thingsboard/server/service/subscription/TbEntityDataSubCtx.java
View File

@ -48,9 +48,6 @@ import java.util.stream.Collectors;
@Slf4j
public class TbEntityDataSubCtx extends TbAbstractDataSubCtx<EntityDataQuery> {
@Getter
@Setter
private TimeSeriesCmd tsCmd;
@Getter
@Setter
private boolean initialDataSent;
@ -183,25 +180,18 @@ public class TbEntityDataSubCtx extends TbAbstractDataSubCtx<EntityDataQuery> {
subIdsToCancel.forEach(subToEntityIdMap::remove);
List<EntityData> newSubsList = newDataMap.entrySet().stream().filter(entry -> !currentSubs.contains(entry.getKey())).map(Map.Entry::getValue).collect(Collectors.toList());
if (!newSubsList.isEmpty()) {
boolean resultToLatestValues;
List<EntityKey> keys = null;
if (curTsCmd != null) {
resultToLatestValues = false;
keys = curTsCmd.getKeys().stream().map(key -> new EntityKey(EntityKeyType.TIME_SERIES, key)).collect(Collectors.toList());
} else if (latestValueCmd != null) {
resultToLatestValues = true;
keys = latestValueCmd.getKeys();
} else {
resultToLatestValues = true;
}
if (keys != null && !keys.isEmpty()) {
Map<EntityKeyType, List<EntityKey>> keysByType = getEntityKeyByTypeMap(keys);
newSubsList.forEach(
entity -> {
log.trace("[{}][{}] Found new subscription for entity: {}", sessionRef.getSessionId(), cmdId, entity.getEntityId());
subsToAdd.addAll(addSubscriptions(entity, keysByType, resultToLatestValues));
}
);
// NOTE: We ignore the TS subscriptions for new entities here, because widgets will re-init it's content and will create new subscriptions.
if (curTsCmd == null && latestValueCmd != null) {
List<EntityKey> keys = latestValueCmd.getKeys();
if (keys != null && !keys.isEmpty()) {
Map<EntityKeyType, List<EntityKey>> keysByType = getEntityKeyByTypeMap(keys);
newSubsList.forEach(
entity -> {
log.trace("[{}][{}] Found new subscription for entity: {}", sessionRef.getSessionId(), cmdId, entity.getEntityId());
subsToAdd.addAll(addSubscriptions(entity, keysByType, true, 0, 0));
}
);
}
}
}
wsService.sendWsMsg(sessionRef.getSessionId(), new EntityDataUpdate(cmdId, data, null, maxEntitiesPerDataSubscription));

2
application/src/main/java/org/thingsboard/server/service/subscription/TbSubscriptionUtils.java
View File

@ -83,6 +83,7 @@ public class TbSubscriptionUtils {
TbSubscriptionKetStateProto.newBuilder().setKey(key).setTs(value).build()));
tSubProto.setStartTime(tSub.getStartTime());
tSubProto.setEndTime(tSub.getEndTime());
tSubProto.setLatestValues(tSub.isLatestValues());
msgBuilder.setTelemetrySub(tSubProto.build());
break;
case ATTRIBUTES:
@ -146,6 +147,7 @@ public class TbSubscriptionUtils {
telemetrySub.getKeyStatesList().forEach(ksProto -> keyStates.put(ksProto.getKey(), ksProto.getTs()));
builder.startTime(telemetrySub.getStartTime());
builder.endTime(telemetrySub.getEndTime());
builder.latestValues(telemetrySub.getLatestValues());
builder.keyStates(keyStates);
return builder.build();
}

5
application/src/main/java/org/thingsboard/server/service/subscription/TbTimeseriesSubscription.java
View File

@ -34,16 +34,19 @@ public class TbTimeseriesSubscription extends TbSubscription<TelemetrySubscripti
private final long startTime;
@Getter
private final long endTime;
@Getter
private final boolean latestValues;
@Builder
public TbTimeseriesSubscription(String serviceId, String sessionId, int subscriptionId, TenantId tenantId, EntityId entityId,
BiConsumer<String, TelemetrySubscriptionUpdate> updateConsumer,
boolean allKeys, Map<String, Long> keyStates, long startTime, long endTime) {
boolean allKeys, Map<String, Long> keyStates, long startTime, long endTime, boolean latestValues) {
super(serviceId, sessionId, subscriptionId, tenantId, entityId, TbSubscriptionType.TIMESERIES, updateConsumer);
this.allKeys = allKeys;
this.keyStates = keyStates;
this.startTime = startTime;
this.endTime = endTime;
this.latestValues = latestValues;
}
@Override

1
common/queue/src/main/proto/queue.proto
View File

@ -337,6 +337,7 @@ message TbTimeSeriesSubscriptionProto {
repeated TbSubscriptionKetStateProto keyStates = 3;
int64 startTime = 4;
int64 endTime = 5;
bool latestValues = 6;
}
message TbAttributeSubscriptionProto {

Write Preview
Loading…
Cancel
Save