From 0953dcbea3820b8000d1f5b2715f360590494e53 Mon Sep 17 00:00:00 2001 From: sshiv012 <122703005+sshiv012@users.noreply.github.com> Date: Mon, 10 Aug 2026 23:41:13 -0700 Subject: [PATCH] seed bb7f5a --- .../workflow-result.service.ts | 78 +++++++++---------- 1 file changed, 39 insertions(+), 39 deletions(-) diff --git a/frontend/src/app/workspace/service/workflow-result/workflow-result.service.ts b/frontend/src/app/workspace/service/workflow-result/workflow-result.service.ts index 26133a636e2..4a68ae672b1 100644 --- a/frontend/src/app/workspace/service/workflow-result/workflow-result.service.ts +++ b/frontend/src/app/workspace/service/workflow-result/workflow-result.service.ts @@ -42,22 +42,22 @@ import { SchemaAttribute } from "../../types/workflow-compiling.interface"; providedIn: "root", }) export class WorkflowResultService { - private paginatedResultServices = new Map(); - private operatorResultServices = new Map(); + private paginatedResultServiceByOperator = new Map(); + private resultServiceByOperator = new Map(); // event stream of operator result update, undefined indicates the operator result is cleared - private resultUpdateStream = new Subject>(); + private resultUpdateSubject = new Subject>(); private resultTableStats = new ReplaySubject>>>(1); - private resultInitiateStream = new Subject(); + private resultInitiateSubject = new Subject(); // emits when clearResults() drops cached results, so the UI can drop stale frames - private resultClearedStream = new Subject(); + private resultClearedSubject = new Subject(); - constructor(private wsService: WorkflowWebsocketService) { - this.wsService.subscribeToEvent("WebResultUpdateEvent").subscribe(event => { + constructor(private websocketService: WorkflowWebsocketService) { + this.websocketService.subscribeToEvent("WebResultUpdateEvent").subscribe(event => { this.handleResultUpdate(event.updates); this.handleTableStatsUpdate(event.tableStats); }); - this.wsService + this.websocketService .subscribeToEvent("WorkflowAvailableResultEvent") .subscribe(event => this.handleCleanResultCache(event)); this.resultTableStats.next({}); @@ -75,18 +75,20 @@ export class WorkflowResultService { return isDefined(this.getPaginatedResultService(operatorID)); } - public getResultUpdateStream(): Observable> { - return this.resultUpdateStream; + public getPaginatedResultService(operatorID: string): OperatorPaginationResultService | undefined { + return this.paginatedResultServiceByOperator.get(operatorID); } - public getResultTableStats(): Observable< - [Record>>, Record>>] - > { - return this.resultTableStats.pipe(pairwise()); + public getResultService(operatorID: string): OperatorResultService | undefined { + return this.resultServiceByOperator.get(operatorID); + } + + public getResultUpdateStream(): Observable> { + return this.resultUpdateSubject; } public getResultInitiateStream(): Observable { - return this.resultInitiateStream.asObservable(); + return this.resultInitiateSubject.asObservable(); } /** @@ -94,41 +96,39 @@ export class WorkflowResultService { * stale frames (clearing the caches alone won't re-render a displayed operator). */ public getResultClearedStream(): Observable { - return this.resultClearedStream.asObservable(); + return this.resultClearedSubject.asObservable(); } - public getPaginatedResultService(operatorID: string): OperatorPaginationResultService | undefined { - return this.paginatedResultServices.get(operatorID); - } - - public getResultService(operatorID: string): OperatorResultService | undefined { - return this.operatorResultServices.get(operatorID); + public getResultTableStats(): Observable< + [Record>>, Record>>] + > { + return this.resultTableStats.pipe(pairwise()); } /** * Drop cached results and reset table stats so a re-entered workflow doesn't show * stale results (resultTableStats is a ReplaySubject, so push an empty snapshot). - * Emits resultClearedStream so subscribers tear down already-displayed frames. + * Emits resultClearedSubject so subscribers tear down already-displayed frames. */ public clearResults(): void { - this.operatorResultServices.clear(); - this.paginatedResultServices.clear(); + this.resultServiceByOperator.clear(); + this.paginatedResultServiceByOperator.clear(); this.resultTableStats.next({}); - this.resultClearedStream.next(); + this.resultClearedSubject.next(); } private handleCleanResultCache(event: WorkflowAvailableResultEvent): void { const removedOrInvalidatedOperators = new Set(); // remove operators that no longer have results - this.operatorResultServices.forEach((_, op) => { + this.resultServiceByOperator.forEach((_, op) => { if (!(op in event.availableOperators)) { - this.operatorResultServices.delete(op); + this.resultServiceByOperator.delete(op); removedOrInvalidatedOperators.add(op); } }); - this.paginatedResultServices.forEach((_, op) => { + this.paginatedResultServiceByOperator.forEach((_, op) => { if (!(op in event.availableOperators)) { - this.paginatedResultServices.delete(op); + this.paginatedResultServiceByOperator.delete(op); removedOrInvalidatedOperators.add(op); } }); @@ -156,7 +156,7 @@ export class WorkflowResultService { const invalidatedOperatorsUpdate: Record = {}; removedOrInvalidatedOperators.forEach(op => (invalidatedOperatorsUpdate[op] = undefined)); - this.resultUpdateStream.next(invalidatedOperatorsUpdate); + this.resultUpdateSubject.next(invalidatedOperatorsUpdate); } private handleResultUpdate(event: WorkflowResultUpdate): void { @@ -166,15 +166,15 @@ export class WorkflowResultService { const paginatedResultService = this.getOrInitPaginatedResultService(operatorID); paginatedResultService.handleResultUpdate(update); // clear previously saved result service - this.operatorResultServices.delete(operatorID); + this.resultServiceByOperator.delete(operatorID); } else if (isWebDataUpdate(update)) { const resultService = this.getOrInitResultService(operatorID); resultService.handleResultUpdate(update); // clear previously saved paginated result service - this.paginatedResultServices.delete(operatorID); + this.paginatedResultServiceByOperator.delete(operatorID); } }); - this.resultUpdateStream.next(event); + this.resultUpdateSubject.next(event); } private handleTableStatsUpdate(event: WorkflowResultTableStats): void { @@ -188,9 +188,9 @@ export class WorkflowResultService { private getOrInitPaginatedResultService(operatorID: string): OperatorPaginationResultService { let service = this.getPaginatedResultService(operatorID); if (!service) { - service = new OperatorPaginationResultService(operatorID, this.wsService); - this.paginatedResultServices.set(operatorID, service); - this.resultInitiateStream.next(operatorID); + service = new OperatorPaginationResultService(operatorID, this.websocketService); + this.paginatedResultServiceByOperator.set(operatorID, service); + this.resultInitiateSubject.next(operatorID); } return service; } @@ -199,8 +199,8 @@ export class WorkflowResultService { let service = this.getResultService(operatorID); if (!service) { service = new OperatorResultService(operatorID); - this.operatorResultServices.set(operatorID, service); - this.resultInitiateStream.next(operatorID); + this.resultServiceByOperator.set(operatorID, service); + this.resultInitiateSubject.next(operatorID); } return service; }