PG1204 commented on code in PR #8301:
URL: https://github.com/apache/texera/pull/8301#discussion_r3918322155
##########
frontend/src/app/workspace/service/workflow-status/workflow-status.service.ts:
##########
@@ -27,37 +27,71 @@ import { OperatorPerformanceMetrics,
extractPerformanceMetrics } from "./perform
providedIn: "root",
})
export class WorkflowStatusService {
- // status is responsible for passing websocket responses to other components
- private statusSubject = new Subject<Record<string, OperatorStatistics>>();
- private currentStatus: Record<string, OperatorStatistics> = {};
+ // The engine streams operator state and operator statistics bundled in one
+ // wire object (OperatorRuntimeStatus); this service splits them into two
+ // separate sub-concepts, each with its own stream and snapshot. Derived
+ // performance metrics are the third, separate concept.
+ private stateSubject = new Subject<Record<string, OperatorState>>();
+ private currentState: Record<string, OperatorState> = {};
+
+ private statisticsSubject = new Subject<Record<string,
OperatorStatistics>>();
+ private currentStatistics: Record<string, OperatorStatistics> = {};
// Derived, ground-truth performance metrics for the heat-map overlay.
Backed by
// a BehaviorSubject so a consumer that subscribes after a run already
streamed
// (e.g. the overlay toggled on mid/post-run) still receives the latest
value.
private performanceMetricsSubject = new BehaviorSubject<Record<string,
OperatorPerformanceMetrics>>({});
constructor(private workflowWebsocketService: WorkflowWebsocketService) {
- // Single derivation path: every status emission (websocket, reset, clear,
or
- // externally fed historical stats) recomputes the performance metrics.
- this.getStatusUpdateStream().subscribe(status => {
- this.currentStatus = status;
-
this.performanceMetricsSubject.next(this.buildPerformanceMetrics(status));
+ this.getStateUpdateStream().subscribe(state => {
+ this.currentState = state;
+ });
+
+ // Single derivation path: every statistics emission (websocket, reset, or
+ // clear) recomputes the performance metrics.
+ this.getStatisticsUpdateStream().subscribe(statistics => {
+ this.currentStatistics = statistics;
+
this.performanceMetricsSubject.next(this.buildPerformanceMetrics(statistics));
});
+ // Each wire event produces exactly one emission on each stream, state
+ // first and statistics second (resetStatus/clearStatus follow the same
+ // order), so a statistics subscriber may read getCurrentState() and see
Review Comment:
addressed in
[52e1942](https://github.com/apache/texera/pull/8301/commits/52e19424c8ad57a46807ee1d3d123a886fd3177e)
PR title modified.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
