UI: Improve data subscription reconnect flow

Igor Kulikov
1 parent e6e9be18
Showing 6 changed files with 62 additions and 27 deletions
ui-ngx/src/app/core/api/entity-data-subscription.ts
ui-ngx/src/app/core/api/entity-data.service.ts
ui-ngx/src/app/core/api/widget-api.models.ts
ui-ngx/src/app/core/api/widget-subscription.ts
ui-ngx/src/app/modules/home/components/widget/lib/entities-table-widget.component.ts
ui-ngx/src/app/modules/home/components/widget/widget.component.ts
--- a/ui-ngx/src/app/core/api/entity-data-subscription.ts
View file @391c88e
+++ b/ui-ngx/src/app/core/api/entity-data-subscription.ts
View file @391c88e
@@ -37,7 +37,7 @@ import {
 } from '@shared/models/telemetry/telemetry.models';
 import { UtilsService } from '@core/services/utils.service';
 import { EntityDataListener, EntityDataLoadResult } from '@core/api/entity-data.service';
-import { deepClone, isDefinedAndNotNull, isObject, objectHashCode } from '@core/utils';
+import { deepClone, isDefinedAndNotNull, isEqual, isObject, objectHashCode } from '@core/utils';
 import { PageData } from '@shared/models/page/page-data';
 import { DataAggregator } from '@core/api/data-aggregator';
 import { NULL_UUID } from '@shared/models/id/has-uuid';
@@ -50,7 +50,7 @@ export interface EntityDataSubscriptionOptions {
   dataKeys: Array<SubscriptionDataKey>;
   type: widgetType;
   entityFilter?: EntityFilter;
-  isLatestDataSubscription?: boolean;
+  isPaginatedDataSubscription?: boolean;
   pageLink?: EntityDataPageLink;
   keyFilters?: Array<KeyFilter>;
   subscriptionTimewindow?: SubscriptionTimewindow;
@@ -154,7 +154,7 @@ export class EntityDataSubscription {
   }
   public subscribe(): Observable<EntityDataLoadResult> {
-    if (!this.entityDataSubscriptionOptions.isLatestDataSubscription) {
+    if (!this.entityDataSubscriptionOptions.isPaginatedDataSubscription) {
       this.entityDataResolveSubject = new ReplaySubject(1);
     } else {
       this.started = true;
@@ -199,7 +199,7 @@ export class EntityDataSubscription {
         latestValues: this.latestValues
       };
-      if (this.entityDataSubscriptionOptions.isLatestDataSubscription) {
+      if (this.entityDataSubscriptionOptions.isPaginatedDataSubscription) {
         if (this.entityDataSubscriptionOptions.type === widgetType.latest) {
           if (this.latestValues.length > 0) {
             this.dataCommand.latestCmd = {
@@ -222,7 +222,7 @@ export class EntityDataSubscription {
       );
       this.subscriber.reconnect$.subscribe(() => {
-        if (this.started && !this.entityDataSubscriptionOptions.isLatestDataSubscription) {
+        if (this.started && !this.entityDataSubscriptionOptions.isPaginatedDataSubscription) {
           if (this.entityDataSubscriptionOptions.type === widgetType.timeseries &&
               !this.history && this.tsFields.length) {
             const newSubsTw: SubscriptionTimewindow = this.listener.updateRealtimeSubscription();
@@ -271,14 +271,14 @@ export class EntityDataSubscription {
         totalPages: 1
       };
       this.onPageData(pageData);
-      if (this.entityDataSubscriptionOptions.isLatestDataSubscription) {
+      if (this.entityDataSubscriptionOptions.isPaginatedDataSubscription) {
         if (this.entityDataSubscriptionOptions.type === widgetType.latest) {
           this.frequency = 1000;
           this.timer = setTimeout(this.onTick.bind(this, true), 0);
         }
       }
     }
-    if (this.entityDataSubscriptionOptions.isLatestDataSubscription) {
+    if (this.entityDataSubscriptionOptions.isPaginatedDataSubscription) {
       return of(null);
     } else {
       return this.entityDataResolveSubject.asObservable();
@@ -286,7 +286,7 @@ export class EntityDataSubscription {
   }
   public start() {
-    if (this.entityDataSubscriptionOptions.isLatestDataSubscription) {
+    if (this.entityDataSubscriptionOptions.isPaginatedDataSubscription) {
       return;
     }
     this.subsTw = this.entityDataSubscriptionOptions.subscriptionTimewindow;
@@ -432,9 +432,26 @@ export class EntityDataSubscription {
     }
   }
+  private pageDataChanged(prevPageData: PageData<EntityData>, nextPageData: PageData<EntityData>) {
+    const prevIds = prevPageData.data.map((entityData) => entityData.entityId.id);
+    const nextIds = nextPageData.data.map((entityData) => entityData.entityId.id);
+    return !isEqual(prevIds, nextIds);
+  }
+
   private onPageData(pageData: PageData<EntityData>) {
+    const isInitialData = !this.pageData;
+    if (!isInitialData && !this.entityDataSubscriptionOptions.isPaginatedDataSubscription) {
+      if (this.pageDataChanged(this.pageData, pageData)) {
+        if (this.listener.initialPageDataChanged) {
+          this.listener.initialPageDataChanged(pageData);
+        }
+        return;
+      }
+    }
     this.pageData = pageData;
-    this.resetData();
+    if (isInitialData || this.entityDataSubscriptionOptions.isPaginatedDataSubscription) {
+      this.resetData();
+    }
     const data: Array<Array<DataSetHolder>> = [];
     for (let dataIndex = 0; dataIndex < pageData.data.length; dataIndex++) {
       const entityData = pageData.data[dataIndex];
@@ -458,8 +475,10 @@ export class EntityDataSubscription {
       );
       this.entityDataResolveSubject.complete();
     } else {
-      this.listener.dataLoaded(pageData, data,
-        this.listener.configDatasourceIndex);
+      if (isInitialData || this.entityDataSubscriptionOptions.isPaginatedDataSubscription) {
+        this.listener.dataLoaded(pageData, data,
+          this.listener.configDatasourceIndex);
+      }
     }
   }
@@ -487,8 +506,13 @@ export class EntityDataSubscription {
     }
     if (this.entityDataSubscriptionOptions.type === widgetType.timeseries && entityData.timeseries) {
       const subscriptionData = this.toSubscriptionData(entityData.timeseries, true);
-      if (!this.history && aggregate) {
-        this.dataAggregators[dataIndex].onData({data: subscriptionData}, false, false, true);
+      if (!this.history) {
+        if (this.dataAggregators && this.dataAggregators[dataIndex]) {
+          this.dataAggregators[dataIndex].onData({data: subscriptionData}, false, false, true);
+        }
+        if (!aggregate) {
+          this.onData(subscriptionData, DataKeyType.timeseries, dataIndex, true, dataUpdatedCb);
+        }
       } else {
         this.onData(subscriptionData, DataKeyType.timeseries, dataIndex, true, dataUpdatedCb);
       }
--- a/ui-ngx/src/app/core/api/entity-data.service.ts
View file @391c88e
+++ b/ui-ngx/src/app/core/api/entity-data.service.ts
View file @391c88e
@@ -33,6 +33,7 @@ export interface EntityDataListener {
   configDatasourceIndex: number;
   dataLoaded: (pageData: PageData<EntityData>, data: Array<Array<DataSetHolder>>, datasourceIndex: number) => void;
   dataUpdated: (data: DataSetHolder, datasourceIndex: number, dataIndex: number, dataKeyIndex: number, detectChanges: boolean) => void;
+  initialPageDataChanged?: (nextPageData: PageData<EntityData>) => void;
   updateRealtimeSubscription?: () => SubscriptionTimewindow;
   setRealtimeSubscription?: (subscriptionTimewindow: SubscriptionTimewindow) => void;
   subscription?: EntityDataSubscription;
@@ -70,9 +71,9 @@ export class EntityDataService {
     listener.subscription.start();
   }
-  public subscribeForLatestData(listener: EntityDataListener,
-                                pageLink: EntityDataPageLink,
-                                keyFilters: KeyFilter[]) {
+  public subscribeForPaginatedData(listener: EntityDataListener,
+                                   pageLink: EntityDataPageLink,
+                                   keyFilters: KeyFilter[]) {
     const datasource = listener.configDatasource;
     if (datasource.type === DatasourceType.entity && (!datasource.entityFilter || !pageLink)) {
       return;
@@ -89,7 +90,7 @@ export class EntityDataService {
   private createSubscription(listener: EntityDataListener,
                              pageLink: EntityDataPageLink,
                              keyFilters: KeyFilter[],
-                             isLatestDataSubscription: boolean): EntityDataSubscription {
+                             isPaginatedDataSubscription: boolean): EntityDataSubscription {
     const datasource = listener.configDatasource;
     const subscriptionDataKeys: Array<SubscriptionDataKey> = [];
     datasource.dataKeys.forEach((dataKey) => {
@@ -111,7 +112,7 @@ export class EntityDataService {
       entityDataSubscriptionOptions.pageLink = pageLink;
       entityDataSubscriptionOptions.keyFilters = keyFilters;
     }
-    entityDataSubscriptionOptions.isLatestDataSubscription = isLatestDataSubscription;
+    entityDataSubscriptionOptions.isPaginatedDataSubscription = isPaginatedDataSubscription;
     return new EntityDataSubscription(entityDataSubscriptionOptions,
       listener, this.telemetryService, this.utils);
   }
--- a/ui-ngx/src/app/core/api/widget-api.models.ts
View file @391c88e
+++ b/ui-ngx/src/app/core/api/widget-api.models.ts
View file @391c88e
@@ -41,7 +41,7 @@ import { EntityAliases } from '@shared/models/alias.models';
 import { EntityInfo } from '@app/shared/models/entity.models';
 import { IDashboardComponent } from '@home/models/dashboard-component.models';
 import * as moment_ from 'moment';
-import { EntityDataPageLink, EntityFilter, KeyFilter } from '@shared/models/query/query.models';
+import { EntityData, EntityDataPageLink, EntityFilter, KeyFilter } from '@shared/models/query/query.models';
 import { EntityDataService } from '@core/api/entity-data.service';
 import { PageData } from '@shared/models/page/page-data';
@@ -185,6 +185,7 @@ export class WidgetSubscriptionContext {
 export interface WidgetSubscriptionCallbacks {
   onDataUpdated?: (subscription: IWidgetSubscription, detectChanges: boolean) => void;
   onDataUpdateError?: (subscription: IWidgetSubscription, e: any) => void;
+  onInitialPageDataChanged?: (subscription: IWidgetSubscription, nextPageData: PageData<EntityData>) => void;
   dataLoading?: (subscription: IWidgetSubscription) => void;
   legendDataUpdated?: (subscription: IWidgetSubscription, detectChanges: boolean) => void;
   timeWindowUpdated?: (subscription: IWidgetSubscription, timeWindowConfig: Timewindow) => void;
@@ -279,9 +280,9 @@ export interface IWidgetSubscription {
   subscribe(): void;
-  subscribeForLatestData(datasourceIndex: number,
-                         pageLink: EntityDataPageLink,
-                         keyFilters: KeyFilter[]): void;
+  subscribeForPaginatedData(datasourceIndex: number,
+                            pageLink: EntityDataPageLink,
+                            keyFilters: KeyFilter[]): void;
   isDataResolved(): boolean;
--- a/ui-ngx/src/app/core/api/widget-subscription.ts
View file @391c88e
+++ b/ui-ngx/src/app/core/api/widget-subscription.ts
View file @391c88e
@@ -209,6 +209,7 @@ export class WidgetSubscription implements IWidgetSubscription {
     } else {
       this.callbacks.onDataUpdated = this.callbacks.onDataUpdated || (() => {});
       this.callbacks.onDataUpdateError = this.callbacks.onDataUpdateError || (() => {});
+      this.callbacks.onInitialPageDataChanged = this.callbacks.onInitialPageDataChanged || (() => {});
       this.callbacks.dataLoading = this.callbacks.dataLoading || (() => {});
       this.callbacks.legendDataUpdated = this.callbacks.legendDataUpdated || (() => {});
       this.callbacks.timeWindowUpdated = this.callbacks.timeWindowUpdated || (() => {});
@@ -400,6 +401,7 @@ export class WidgetSubscription implements IWidgetSubscription {
         dataLoaded: (pageData, data1, datasourceIndex) => {
           this.dataLoaded(pageData, data1, datasourceIndex, true)
         },
+        initialPageDataChanged: this.initialPageDataChanged.bind(this),
         dataUpdated: this.dataUpdated.bind(this),
         updateRealtimeSubscription: () => {
           this.subscriptionTimewindow = this.updateRealtimeSubscription();
@@ -852,9 +854,9 @@ export class WidgetSubscription implements IWidgetSubscription {
     }
   }
-  subscribeForLatestData(datasourceIndex: number,
-                         pageLink: EntityDataPageLink,
-                         keyFilters: KeyFilter[]): void {
+  subscribeForPaginatedData(datasourceIndex: number,
+                            pageLink: EntityDataPageLink,
+                            keyFilters: KeyFilter[]): void {
     let entityDataListener = this.entityDataListeners[datasourceIndex];
     if (entityDataListener) {
       this.ctx.entityDataService.stopSubscription(entityDataListener);
@@ -871,7 +873,7 @@ export class WidgetSubscription implements IWidgetSubscription {
         dataUpdated: this.dataUpdated.bind(this)
       };
       this.entityDataListeners[datasourceIndex] = entityDataListener;
-      this.ctx.entityDataService.subscribeForLatestData(entityDataListener, pageLink, keyFilters);
+      this.ctx.entityDataService.subscribeForPaginatedData(entityDataListener, pageLink, keyFilters);
     }
   }
@@ -1156,6 +1158,10 @@ export class WidgetSubscription implements IWidgetSubscription {
     return this.timewindowForComparison;
   }
+  private initialPageDataChanged(nextPageData: PageData<EntityData>) {
+    this.callbacks.onInitialPageDataChanged(this, nextPageData);
+  }
+
   private dataLoaded(pageData: PageData<EntityData>,
                      data: Array<Array<DataSetHolder>>,
                      datasourceIndex: number, isUpdate: boolean) {
--- a/ui-ngx/src/app/modules/home/components/widget/lib/entities-table-widget.component.ts
View file @391c88e
+++ b/ui-ngx/src/app/modules/home/components/widget/lib/entities-table-widget.component.ts
View file @391c88e
@@ -614,7 +614,7 @@ class EntityDatasource implements DataSource<EntityData> {
   }
   loadEntities(pageLink: EntityDataPageLink, keyFilters: KeyFilter[]) {
-    this.subscription.subscribeForLatestData(0, pageLink, keyFilters);
+    this.subscription.subscribeForPaginatedData(0, pageLink, keyFilters);
 /*    this.fetchEntities(pageLink).pipe(
       catchError(() => of(emptyPageData<EntityData>())),
     ).subscribe(
--- a/ui-ngx/src/app/modules/home/components/widget/widget.component.ts
View file @391c88e
+++ b/ui-ngx/src/app/modules/home/components/widget/widget.component.ts
View file @391c88e
@@ -818,6 +818,9 @@ export class WidgetComponent extends PageComponent implements OnInit, AfterViewI
       onDataUpdateError: (subscription, e) => {
         this.handleWidgetException(e);
       },
+      onInitialPageDataChanged: (subscription, nextPageData) => {
+        this.reInit();
+      },
       dataLoading: (subscription) => {
         if (this.loadingData !== subscription.loadingData) {
           this.loadingData = subscription.loadingData;