Yicong-Huang commented on code in PR #8301:
URL: https://github.com/apache/texera/pull/8301#discussion_r3919022178
##########
frontend/src/app/workspace/service/workflow-status/workflow-status.service.spec.ts:
##########
@@ -58,32 +63,77 @@ describe("WorkflowStatusService", () => {
service = TestBed.inject(WorkflowStatusService);
});
- it("forwards an OperatorStatisticsUpdateEvent to the status stream", () => {
- const received: Record<string, OperatorStatistics>[] = [];
- service.getStatusUpdateStream().subscribe(s => received.push(s));
+ it("splits an OperatorStatisticsUpdateEvent into the state and statistics
streams", () => {
+ const stateEmissions: Record<string, OperatorState>[] = [];
+ const statisticsEmissions: Record<string, OperatorStatistics>[] = [];
+ service.getStateUpdateStream().subscribe(s => stateEmissions.push(s));
+ service.getStatisticsUpdateStream().subscribe(s =>
statisticsEmissions.push(s));
- websocketEventSubject.next(statsEvent({ op1: sampleStats }));
+ websocketEventSubject.next(statsEvent({ op1: sampleRuntimeStatus }));
- expect(received).toHaveLength(1);
- expect(received[0]).toEqual({ op1: sampleStats });
- expect(service.getCurrentStatus()).toEqual({ op1: sampleStats });
+ expect(stateEmissions).toHaveLength(1);
+ expect(stateEmissions[0]).toEqual({ op1: OperatorState.Running });
+ expect(statisticsEmissions).toHaveLength(1);
+ expect(statisticsEmissions[0]).toEqual({ op1: sampleStatistics });
+ expect(service.getCurrentState()).toEqual({ op1: OperatorState.Running });
+ expect(service.getCurrentStatistics()).toEqual({ op1: sampleStatistics });
+ });
+
+ it("does not leak the operator state into the statistics concept", () => {
+ websocketEventSubject.next(statsEvent({ op1: sampleRuntimeStatus }));
+
expect(service.getCurrentStatistics()["op1"]).not.toHaveProperty("operatorState");
+ });
+
+ it("exposes state and statistics as independent streams", () => {
+ // A consumer subscribed to only one of the two sub-concepts sees exactly
+ // one emission per update, unaffected by the other stream.
+ let stateCount = 0;
+ let statisticsCount = 0;
+ service.getStateUpdateStream().subscribe(() => stateCount++);
+ service.getStatisticsUpdateStream().subscribe(() => statisticsCount++);
+
+ websocketEventSubject.next(statsEvent({ op1: sampleRuntimeStatus }));
+ websocketEventSubject.next(statsEvent({ op1: { ...sampleRuntimeStatus,
operatorState: OperatorState.Paused } }));
+
+ expect(stateCount).toBe(2);
+ expect(statisticsCount).toBe(2);
+ expect(service.getCurrentState()).toEqual({ op1: OperatorState.Paused });
+ });
+
+ it("emits state before statistics, so a statistics subscriber sees the
matching state snapshot", () => {
+ const order: string[] = [];
+ service.getStateUpdateStream().subscribe(() => order.push("state"));
+ service.getStatisticsUpdateStream().subscribe(() => {
+ order.push("statistics");
+ // The licensed read: by the time statistics arrive, the state snapshot
+ // already reflects the same wire event. The converse does not hold.
+ expect(service.getCurrentState()).toEqual({ op1: OperatorState.Running
});
+ });
+
+ websocketEventSubject.next(statsEvent({ op1: sampleRuntimeStatus }));
+
+ expect(order).toEqual(["state", "statistics"]);
Review Comment:
rxjs 7.8.2 does not let the `next()` caller catch a throw from a subscriber
— it re-throws asynchronously — so this assertion cannot fail the test it sits
in. If the snapshot guarantee broke, the order assertion below would still pass
and the failure would surface detached, as an unhandled error. It is the same
hazard the flag at `workflow-editor.component.spec.ts:1963` works around;
capturing the snapshot and asserting after `next()` returns avoids needing the
flag at all.
```suggestion
const order: string[] = [];
let stateAtStatistics: Record<string, OperatorState> | undefined;
service.getStateUpdateStream().subscribe(() => order.push("state"));
service.getStatisticsUpdateStream().subscribe(() => {
order.push("statistics");
// The licensed read: by the time statistics arrive, the state snapshot
// already reflects the same wire event. The converse does not hold.
stateAtStatistics = service.getCurrentState();
});
websocketEventSubject.next(statsEvent({ op1: sampleRuntimeStatus }));
expect(order).toEqual(["state", "statistics"]);
expect(stateAtStatistics).toEqual({ op1: OperatorState.Running });
```
--
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]
