Showing
5 changed files
with
69 additions
and
37 deletions
| @@ -401,9 +401,14 @@ public class DefaultTbEntityDataSubscriptionService implements TbEntityDataSubsc | @@ -401,9 +401,14 @@ public class DefaultTbEntityDataSubscriptionService implements TbEntityDataSubsc | ||
| 401 | new EntityDataUpdate(ctx.getCmdId(), SubscriptionErrorCode.INTERNAL_ERROR.getCode(), "Failed to fetch historical data!")); | 401 | new EntityDataUpdate(ctx.getCmdId(), SubscriptionErrorCode.INTERNAL_ERROR.getCode(), "Failed to fetch historical data!")); |
| 402 | } | 402 | } |
| 403 | }); | 403 | }); |
| 404 | - EntityDataUpdate update = new EntityDataUpdate(ctx.getCmdId(), ctx.getData(), null); | 404 | + EntityDataUpdate update; |
| 405 | + if (!ctx.isInitialDataSent()) { | ||
| 406 | + update = new EntityDataUpdate(ctx.getCmdId(), ctx.getData(), null); | ||
| 407 | + ctx.setInitialDataSent(true); | ||
| 408 | + } else { | ||
| 409 | + update = new EntityDataUpdate(ctx.getCmdId(), null, ctx.getData().getData()); | ||
| 410 | + } | ||
| 405 | wsService.sendWsMsg(ctx.getSessionId(), update); | 411 | wsService.sendWsMsg(ctx.getSessionId(), update); |
| 406 | - ctx.setInitialDataSent(true); | ||
| 407 | return ctx; | 412 | return ctx; |
| 408 | }, wsCallBackExecutor); | 413 | }, wsCallBackExecutor); |
| 409 | } | 414 | } |
| @@ -191,12 +191,14 @@ public class TbEntityDataSubCtx { | @@ -191,12 +191,14 @@ public class TbEntityDataSubCtx { | ||
| 191 | if (latestCtxValues != null) { | 191 | if (latestCtxValues != null) { |
| 192 | latestCtxValues.forEach((k, v) -> { | 192 | latestCtxValues.forEach((k, v) -> { |
| 193 | TsValue update = latestUpdate.get(k); | 193 | TsValue update = latestUpdate.get(k); |
| 194 | - if (update.getTs() < v.getTs()) { | ||
| 195 | - log.trace("[{}][{}][{}] Removed stale update for key: {} and ts: {}", sessionId, cmdId, subscriptionUpdate.getSubscriptionId(), k, update.getTs()); | ||
| 196 | - latestUpdate.remove(k); | ||
| 197 | - } else if ((update.getTs() == v.getTs() && update.getValue().equals(v.getValue()))) { | ||
| 198 | - log.trace("[{}][{}][{}] Removed duplicate update for key: {} and ts: {}", sessionId, cmdId, subscriptionUpdate.getSubscriptionId(), k, update.getTs()); | ||
| 199 | - latestUpdate.remove(k); | 194 | + if (update != null) { |
| 195 | + if (update.getTs() < v.getTs()) { | ||
| 196 | + log.trace("[{}][{}][{}] Removed stale update for key: {} and ts: {}", sessionId, cmdId, subscriptionUpdate.getSubscriptionId(), k, update.getTs()); | ||
| 197 | + latestUpdate.remove(k); | ||
| 198 | + } else if ((update.getTs() == v.getTs() && update.getValue().equals(v.getValue()))) { | ||
| 199 | + log.trace("[{}][{}][{}] Removed duplicate update for key: {} and ts: {}", sessionId, cmdId, subscriptionUpdate.getSubscriptionId(), k, update.getTs()); | ||
| 200 | + latestUpdate.remove(k); | ||
| 201 | + } | ||
| 200 | } | 202 | } |
| 201 | }); | 203 | }); |
| 202 | //Setting new values | 204 | //Setting new values |
| @@ -204,8 +206,16 @@ public class TbEntityDataSubCtx { | @@ -204,8 +206,16 @@ public class TbEntityDataSubCtx { | ||
| 204 | } | 206 | } |
| 205 | } | 207 | } |
| 206 | if (!latestUpdate.isEmpty()) { | 208 | if (!latestUpdate.isEmpty()) { |
| 207 | - Map<EntityKeyType, Map<String, TsValue>> latestMap = Collections.singletonMap(keyType, latestUpdate); | ||
| 208 | - entityData = new EntityData(entityId, latestMap, null); | 209 | + if (resultToLatestValues) { |
| 210 | + Map<EntityKeyType, Map<String, TsValue>> latestMap = Collections.singletonMap(keyType, latestUpdate); | ||
| 211 | + entityData = new EntityData(entityId, latestMap, null); | ||
| 212 | + } else { | ||
| 213 | + Map<String, TsValue[]> tsMap = new HashMap<>(); | ||
| 214 | + latestUpdate.forEach((key, tsValue) -> { | ||
| 215 | + tsMap.put(key, new TsValue[]{tsValue}); | ||
| 216 | + }); | ||
| 217 | + entityData = new EntityData(entityId, null, tsMap); | ||
| 218 | + } | ||
| 209 | wsService.sendWsMsg(sessionId, new EntityDataUpdate(cmdId, null, Collections.singletonList(entityData))); | 219 | wsService.sendWsMsg(sessionId, new EntityDataUpdate(cmdId, null, Collections.singletonList(entityData))); |
| 210 | } | 220 | } |
| 211 | } else { | 221 | } else { |
| @@ -206,25 +206,31 @@ public class DefaultEntityQueryRepository implements EntityQueryRepository { | @@ -206,25 +206,31 @@ public class DefaultEntityQueryRepository implements EntityQueryRepository { | ||
| 206 | case RELATIONS_QUERY: | 206 | case RELATIONS_QUERY: |
| 207 | case DEVICE_SEARCH_QUERY: | 207 | case DEVICE_SEARCH_QUERY: |
| 208 | case ASSET_SEARCH_QUERY: | 208 | case ASSET_SEARCH_QUERY: |
| 209 | - ctx.addUuidParameter("permissions_tenant_id", tenantId.getId()); | ||
| 210 | - ctx.addUuidParameter("permissions_customer_id", customerId.getId()); | ||
| 211 | - return "e.tenant_id=:permissions_tenant_id and e.customer_id=:permissions_customer_id"; | 209 | + return this.defaultPermissionQuery(ctx, tenantId, customerId, entityType); |
| 212 | default: | 210 | default: |
| 213 | if (entityType == EntityType.TENANT) { | 211 | if (entityType == EntityType.TENANT) { |
| 214 | ctx.addUuidParameter("permissions_tenant_id", tenantId.getId()); | 212 | ctx.addUuidParameter("permissions_tenant_id", tenantId.getId()); |
| 215 | return "e.id=:permissions_tenant_id"; | 213 | return "e.id=:permissions_tenant_id"; |
| 216 | - } else if (entityType == EntityType.CUSTOMER) { | ||
| 217 | - ctx.addUuidParameter("permissions_tenant_id", tenantId.getId()); | ||
| 218 | - ctx.addUuidParameter("permissions_customer_id", customerId.getId()); | ||
| 219 | - return "e.tenant_id=:permissions_tenant_id and e.id=:permissions_customer_id"; | ||
| 220 | } else { | 214 | } else { |
| 221 | - ctx.addUuidParameter("permissions_tenant_id", tenantId.getId()); | ||
| 222 | - ctx.addUuidParameter("permissions_customer_id", customerId.getId()); | ||
| 223 | - return "e.tenant_id=:permissions_tenant_id and e.customer_id=:permissions_customer_id"; | 215 | + return this.defaultPermissionQuery(ctx, tenantId, customerId, entityType); |
| 224 | } | 216 | } |
| 225 | } | 217 | } |
| 226 | } | 218 | } |
| 227 | 219 | ||
| 220 | + private String defaultPermissionQuery(EntityQueryContext ctx, TenantId tenantId, CustomerId customerId, EntityType entityType) { | ||
| 221 | + ctx.addUuidParameter("permissions_tenant_id", tenantId.getId()); | ||
| 222 | + if (customerId != null && !customerId.isNullUid()) { | ||
| 223 | + ctx.addUuidParameter("permissions_customer_id", customerId.getId()); | ||
| 224 | + if (entityType == EntityType.CUSTOMER) { | ||
| 225 | + return "e.tenant_id=:permissions_tenant_id and e.id=:permissions_customer_id"; | ||
| 226 | + } else { | ||
| 227 | + return "e.tenant_id=:permissions_tenant_id and e.customer_id=:permissions_customer_id"; | ||
| 228 | + } | ||
| 229 | + } else { | ||
| 230 | + return "e.tenant_id=:permissions_tenant_id"; | ||
| 231 | + } | ||
| 232 | + } | ||
| 233 | + | ||
| 228 | private String buildEntityFilterQuery(EntityQueryContext ctx, EntityFilter entityFilter) { | 234 | private String buildEntityFilterQuery(EntityQueryContext ctx, EntityFilter entityFilter) { |
| 229 | switch (entityFilter.getType()) { | 235 | switch (entityFilter.getType()) { |
| 230 | case SINGLE_ENTITY: | 236 | case SINGLE_ENTITY: |
| @@ -222,21 +222,26 @@ export class EntityDataSubscription { | @@ -222,21 +222,26 @@ export class EntityDataSubscription { | ||
| 222 | ); | 222 | ); |
| 223 | 223 | ||
| 224 | this.subscriber.reconnect$.subscribe(() => { | 224 | this.subscriber.reconnect$.subscribe(() => { |
| 225 | - const newSubsTw: SubscriptionTimewindow = this.listener.updateRealtimeSubscription(); | ||
| 226 | - this.listener.setRealtimeSubscription(newSubsTw); | ||
| 227 | - this.subsTw = newSubsTw; | ||
| 228 | if (this.started && !this.entityDataSubscriptionOptions.isLatestDataSubscription) { | 225 | if (this.started && !this.entityDataSubscriptionOptions.isLatestDataSubscription) { |
| 229 | - this.subsCommand.tsCmd.startTs = this.subsTw.startTs; | ||
| 230 | - this.subsCommand.tsCmd.timeWindow = this.subsTw.aggregation.timeWindow; | ||
| 231 | - this.subsCommand.tsCmd.interval = this.subsTw.aggregation.interval; | ||
| 232 | - this.subsCommand.tsCmd.limit = this.subsTw.aggregation.limit; | ||
| 233 | - this.subsCommand.tsCmd.agg = this.subsTw.aggregation.type; | ||
| 234 | - if (this.subsTw.aggregation.stateData) { | ||
| 235 | - this.subsCommand.historyCmd.startTs = this.subsTw.startTs - YEAR; | ||
| 236 | - this.subsCommand.historyCmd.endTs = this.subsTw.startTs; | ||
| 237 | - this.subsCommand.historyCmd.interval = this.subsTw.aggregation.interval; | ||
| 238 | - this.subsCommand.historyCmd.limit = this.subsTw.aggregation.limit; | ||
| 239 | - this.subsCommand.historyCmd.agg = this.subsTw.aggregation.type; | 226 | + if (this.entityDataSubscriptionOptions.type === widgetType.timeseries && |
| 227 | + !this.history && this.tsFields.length) { | ||
| 228 | + const newSubsTw: SubscriptionTimewindow = this.listener.updateRealtimeSubscription(); | ||
| 229 | + this.subsTw = newSubsTw; | ||
| 230 | + this.subsCommand.tsCmd.startTs = this.subsTw.startTs; | ||
| 231 | + this.subsCommand.tsCmd.timeWindow = this.subsTw.aggregation.timeWindow; | ||
| 232 | + this.subsCommand.tsCmd.interval = this.subsTw.aggregation.interval; | ||
| 233 | + this.subsCommand.tsCmd.limit = this.subsTw.aggregation.limit; | ||
| 234 | + this.subsCommand.tsCmd.agg = this.subsTw.aggregation.type; | ||
| 235 | + if (this.subsTw.aggregation.stateData) { | ||
| 236 | + this.subsCommand.historyCmd.startTs = this.subsTw.startTs - YEAR; | ||
| 237 | + this.subsCommand.historyCmd.endTs = this.subsTw.startTs; | ||
| 238 | + this.subsCommand.historyCmd.interval = this.subsTw.aggregation.interval; | ||
| 239 | + this.subsCommand.historyCmd.limit = this.subsTw.aggregation.limit; | ||
| 240 | + this.subsCommand.historyCmd.agg = this.subsTw.aggregation.type; | ||
| 241 | + } | ||
| 242 | + this.dataAggregators.forEach((dataAggregator) => { | ||
| 243 | + dataAggregator.reset(newSubsTw.startTs, newSubsTw.aggregation.timeWindow, newSubsTw.aggregation.interval); | ||
| 244 | + }); | ||
| 240 | } | 245 | } |
| 241 | this.subsCommand.query = this.dataCommand.query; | 246 | this.subsCommand.query = this.dataCommand.query; |
| 242 | this.subscriber.subscriptionCommands = [this.subsCommand]; | 247 | this.subscriber.subscriptionCommands = [this.subsCommand]; |
| @@ -370,8 +375,10 @@ export class EntityDataSubscription { | @@ -370,8 +375,10 @@ export class EntityDataSubscription { | ||
| 370 | }; | 375 | }; |
| 371 | } | 376 | } |
| 372 | } | 377 | } |
| 373 | - this.subscriber.subscriptionCommands = [this.subsCommand]; | ||
| 374 | - this.subscriber.update(); | 378 | + if (!this.subsCommand.isEmpty()) { |
| 379 | + this.subscriber.subscriptionCommands = [this.subsCommand]; | ||
| 380 | + this.subscriber.update(); | ||
| 381 | + } | ||
| 375 | } else if (this.datasourceType === DatasourceType.function) { | 382 | } else if (this.datasourceType === DatasourceType.function) { |
| 376 | this.frequency = 1000; | 383 | this.frequency = 1000; |
| 377 | if (this.entityDataSubscriptionOptions.type === widgetType.timeseries) { | 384 | if (this.entityDataSubscriptionOptions.type === widgetType.timeseries) { |
| @@ -480,7 +487,7 @@ export class EntityDataSubscription { | @@ -480,7 +487,7 @@ export class EntityDataSubscription { | ||
| 480 | } | 487 | } |
| 481 | if (this.entityDataSubscriptionOptions.type === widgetType.timeseries && entityData.timeseries) { | 488 | if (this.entityDataSubscriptionOptions.type === widgetType.timeseries && entityData.timeseries) { |
| 482 | const subscriptionData = this.toSubscriptionData(entityData.timeseries, true); | 489 | const subscriptionData = this.toSubscriptionData(entityData.timeseries, true); |
| 483 | - if (aggregate) { | 490 | + if (!this.history && aggregate) { |
| 484 | this.dataAggregators[dataIndex].onData({data: subscriptionData}, false, false, true); | 491 | this.dataAggregators[dataIndex].onData({data: subscriptionData}, false, false, true); |
| 485 | } else { | 492 | } else { |
| 486 | this.onData(subscriptionData, DataKeyType.timeseries, dataIndex, true, dataUpdatedCb); | 493 | this.onData(subscriptionData, DataKeyType.timeseries, dataIndex, true, dataUpdatedCb); |
| @@ -157,6 +157,10 @@ export class EntityDataCmd implements WebsocketCmd { | @@ -157,6 +157,10 @@ export class EntityDataCmd implements WebsocketCmd { | ||
| 157 | historyCmd?: EntityHistoryCmd; | 157 | historyCmd?: EntityHistoryCmd; |
| 158 | latestCmd?: LatestValueCmd; | 158 | latestCmd?: LatestValueCmd; |
| 159 | tsCmd?: TimeSeriesCmd; | 159 | tsCmd?: TimeSeriesCmd; |
| 160 | + | ||
| 161 | + public isEmpty(): boolean { | ||
| 162 | + return !this.query && !this.historyCmd && !this.latestCmd && !this.tsCmd; | ||
| 163 | + } | ||
| 160 | } | 164 | } |
| 161 | 165 | ||
| 162 | export class EntityDataUnsubscribeCmd implements WebsocketCmd { | 166 | export class EntityDataUnsubscribeCmd implements WebsocketCmd { |