Cancel subscriptions of overwritten data explorer queries
@@ -31,10 +31,11 @@ import {
} from '@streampipes/platform-services';
import { ResizeService } from '../../../services/resize.service';
import { FieldProvider } from '../../../models/dataview-dashboard.model';
-import { Observable, Subscription, zip } from 'rxjs';
+import { Observable, Subject, Subscription, zip } from 'rxjs';
import { DataExplorerFieldProviderService } from '../../../services/data-explorer-field-provider-service';
import { BaseWidgetData } from './data-explorer-widget-data';
import { TimeSelectionService } from '../../../services/time-selection.service';
+import { switchMap } from 'rxjs/operators';
@Directive()
export abstract class BaseDataExplorerWidgetDirective<T extends DataExplorerWidgetModel> implements BaseWidgetData<T>, OnInit, OnDestroy {
@@ -78,6 +79,8 @@ export abstract class BaseDataExplorerWidgetDirective<T extends DataExplorerWidg
resizeSub: Subscription;
timeSelectionSub: Subscription;
+ requestQueue$: Subject<Observable<SpQueryResult>[]> = new Subject<Observable<SpQueryResult>[]>();
+
constructor(protected dataLakeRestService: DatalakeRestService,
protected widgetConfigurationService: WidgetConfigurationService,
protected resizeService: ResizeService,
@@ -92,6 +95,16 @@ export abstract class BaseDataExplorerWidgetDirective<T extends DataExplorerWidg
this.showData = true;
const sourceConfigs = this.dataExplorerWidget.dataConfig.sourceConfigs;
this.fieldProvider = this.fieldService.generateFieldLists(sourceConfigs);
+
+ this.requestQueue$.pipe(switchMap((observables) => {
+ return zip(...observables);
+ })).subscribe(results => {
+ results.forEach((result, index) => result.sourceIndex = index);
+ this.validateReceivedData(results);
+ this.refreshView();
+ this.timerCallback.emit(false);
+ });
+
this.widgetConfigurationSub = this.widgetConfigurationService.configurationChangedSubject.subscribe(refreshMessage => {
if (refreshMessage.widgetId === this.dataExplorerWidget._id) {
if (refreshMessage.refreshData) {
@@ -129,6 +142,7 @@ export abstract class BaseDataExplorerWidgetDirective<T extends DataExplorerWidg
this.resizeSub.unsubscribe();
}
this.timeSelectionSub.unsubscribe();
+ this.requestQueue$.unsubscribe();
}
public removeWidget() {
@@ -175,13 +189,7 @@ export abstract class BaseDataExplorerWidgetDirective<T extends DataExplorerWidg
}
this.timerCallback.emit(true);
- zip(...observables).subscribe(results => {
- results.forEach((result, index) => result.sourceIndex = index);
- this.validateReceivedData(results);
- // this.onDataReceived(results);
- this.refreshView();
- this.timerCallback.emit(false);
- });
+ this.requestQueue$.next(observables);
}
validateReceivedData(spQueryResults: SpQueryResult[]) {
