This is an automated email from the ASF dual-hosted git repository. tenthe pushed a commit to branch fix-chart-request-strategy in repository https://gitbox.apache.org/repos/asf/streampipes.git
commit 504f9cfff022478b72e04f1b575d49dfb1f02a36 Author: Philipp Zehnder <[email protected]> AuthorDate: Wed Jun 24 12:31:02 2026 +0000 fix: Split chart request refresh strategies --- .../base/base-data-explorer-widget.directive.ts | 122 +++++++++++++++------ 1 file changed, 87 insertions(+), 35 deletions(-) diff --git a/ui/src/app/chart-shared/components/charts/base/base-data-explorer-widget.directive.ts b/ui/src/app/chart-shared/components/charts/base/base-data-explorer-widget.directive.ts index 5cf0f3e886..e5bd69023e 100644 --- a/ui/src/app/chart-shared/components/charts/base/base-data-explorer-widget.directive.ts +++ b/ui/src/app/chart-shared/components/charts/base/base-data-explorer-widget.directive.ts @@ -44,7 +44,7 @@ import { } from '../../../models/dataview-dashboard.model'; import { EMPTY, Observable, Subject, Subscription, zip } from 'rxjs'; import { ChartFieldProviderService } from '../../../services/chart-field-provider.service'; -import { catchError, exhaustMap, finalize } from 'rxjs/operators'; +import { catchError, exhaustMap, finalize, switchMap } from 'rxjs/operators'; import { ChartRegistry } from '../../../registry/chart-registry.service'; import { SpFieldUpdateService } from '../../../services/field-update.service'; import { @@ -120,10 +120,14 @@ export abstract class BaseDataExplorerWidgetDirective< timeSelection$: Subscription; nameChange$: Subscription; - requestQueue$: Subject<Observable<SpQueryResult>[]> = new Subject< + viewModeRequestQueue$: Subject<Observable<SpQueryResult>[]> = new Subject< Observable<SpQueryResult>[] >(); - requestInProgress = false; + latestRequestQueue$: Subject<Observable<SpQueryResult>[]> = new Subject< + Observable<SpQueryResult>[] + >(); + requestQueueSubscriptions = new Subscription(); + viewModeRequestInProgress = false; protected widgetConfigurationService = inject(ChartConfigurationService); protected resizeService = inject(ResizeService); @@ -141,33 +145,7 @@ export abstract class BaseDataExplorerWidgetDirective< this.fieldProvider = this.fieldService.generateFieldLists(sourceConfigs); - this.requestQueue$ - .pipe( - exhaustMap(observables => { - this.requestInProgress = true; - this.errorCallback.emit(undefined); - return zip(...observables).pipe( - catchError(err => { - this.timerCallback.emit(false); - this.errorCallback.emit(err.error); - this.dataReceivedCallback.emit([]); - return EMPTY; - }), - finalize(() => { - this.requestInProgress = false; - }), - ); - }), - ) - .subscribe(results => { - results.forEach( - (result, index) => (result.sourceIndex = index), - ); - this.timerCallback.emit(false); - setTimeout(() => { - this.validateReceivedData(results); - }); - }); + this.initializeRequestQueues(); this.widgetConfiguration$ = this.widgetConfigurationService.configurationChangedSubject.subscribe( @@ -243,7 +221,9 @@ export abstract class BaseDataExplorerWidgetDirective< this.resize$?.unsubscribe(); this.timeSelection$.unsubscribe(); this.nameChange$.unsubscribe(); - this.requestQueue$?.unsubscribe(); + this.requestQueueSubscriptions.unsubscribe(); + this.viewModeRequestQueue$?.unsubscribe(); + this.latestRequestQueue$?.unsubscribe(); } public setShownComponents( @@ -259,7 +239,10 @@ export abstract class BaseDataExplorerWidgetDirective< } public updateData(includeTooMuchEventsParameter: boolean = true) { - if (this.requestInProgress) { + if ( + !this.shouldUseLatestRequestStrategy() && + this.viewModeRequestInProgress + ) { return; } this.beforeDataFetched(); @@ -267,10 +250,18 @@ export abstract class BaseDataExplorerWidgetDirective< } private loadData(includeTooMuchEventsParameter: boolean) { + const observables = this.makeDataRequest(includeTooMuchEventsParameter); + this.timerCallback.emit(true); + this.getRequestQueue().next(observables); + } + + private makeDataRequest( + includeTooMuchEventsParameter: boolean, + ): Observable<SpQueryResult>[] { const returnCompleteResult = includeTooMuchEventsParameter && !this.dataExplorerWidget.dataConfig.ignoreTooMuchDataWarning; - const observables = this.observableGenerator.generateObservables( + return this.observableGenerator.generateObservables( this.timeSettings.startTime, this.timeSettings.endTime, this.dataExplorerWidget.dataConfig as DataExplorerDataConfig, @@ -279,8 +270,69 @@ export abstract class BaseDataExplorerWidgetDirective< ? BaseDataExplorerWidgetDirective.TOO_MUCH_DATA_PARAMETER : undefined, ); - this.timerCallback.emit(true); - this.requestQueue$.next(observables); + } + + private getRequestQueue(): Subject<Observable<SpQueryResult>[]> { + return this.shouldUseLatestRequestStrategy() + ? this.latestRequestQueue$ + : this.viewModeRequestQueue$; + } + + private shouldUseLatestRequestStrategy(): boolean { + return this.dataViewMode && this.editMode; + } + + private initializeRequestQueues(): void { + this.requestQueueSubscriptions.add( + this.viewModeRequestQueue$ + .pipe( + exhaustMap(observables => + this.executeDataRequest(observables, true), + ), + ) + .subscribe(results => this.handleDataRequestResult(results)), + ); + + this.requestQueueSubscriptions.add( + this.latestRequestQueue$ + .pipe( + switchMap(observables => + this.executeDataRequest(observables, false), + ), + ) + .subscribe(results => this.handleDataRequestResult(results)), + ); + } + + private executeDataRequest( + observables: Observable<SpQueryResult>[], + trackViewModeRequest: boolean, + ): Observable<SpQueryResult[]> { + if (trackViewModeRequest) { + this.viewModeRequestInProgress = true; + } + this.errorCallback.emit(undefined); + return zip(...observables).pipe( + catchError(err => { + this.timerCallback.emit(false); + this.errorCallback.emit(err.error); + this.dataReceivedCallback.emit([]); + return EMPTY; + }), + finalize(() => { + if (trackViewModeRequest) { + this.viewModeRequestInProgress = false; + } + }), + ); + } + + private handleDataRequestResult(results: SpQueryResult[]): void { + results.forEach((result, index) => (result.sourceIndex = index)); + this.timerCallback.emit(false); + setTimeout(() => { + this.validateReceivedData(results); + }); } validateReceivedData(spQueryResults: SpQueryResult[]) {
