This is an automated email from the ASF dual-hosted git repository.
tenthe pushed a commit to branch dev
in repository https://gitbox.apache.org/repos/asf/streampipes.git
The following commit(s) were added to refs/heads/dev by this push:
new 067704522d fix: Split chart request refresh strategies (#4634)
067704522d is described below
commit 067704522d1c15d04f298e0b09c83ba312044f69
Author: Philipp Zehnder <[email protected]>
AuthorDate: Thu Jun 25 08:08:48 2026 +0200
fix: Split chart request refresh strategies (#4634)
---
.../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[]) {
