* Run the D2 graph until quiescence, then emit accumulated events once. * * All output across the entire while-loop is accumulated into a single * batch so that users see one `onBatchProcessed` invocation per scheduler * run, even when ordered loading causes multiple graph steps.
()
| 726 | * run, even when ordered loading causes multiple graph steps. |
| 727 | */ |
| 728 | private runGraph(): void { |
| 729 | if (this.isGraphRunning || this.disposed || !this.graph) return |
| 730 | |
| 731 | this.isGraphRunning = true |
| 732 | try { |
| 733 | while (this.graph.pendingWork()) { |
| 734 | this.graph.run() |
| 735 | // A handler (via onBatchProcessed) or source error callback may have |
| 736 | // called dispose() during graph.run(). Stop early to avoid operating |
| 737 | // on stale state. TS narrows disposed to false from the guard above |
| 738 | // but it can change during graph.run() via callbacks. |
| 739 | // eslint-disable-next-line @typescript-eslint/no-unnecessary-condition |
| 740 | if (this.disposed) break |
| 741 | // After each step, check if ordered queries need more data. |
| 742 | // loadMoreIfNeeded may send data to D2 inputs (via requestLimitedSnapshot), |
| 743 | // causing pendingWork() to return true for the next iteration. |
| 744 | this.loadMoreIfNeeded() |
| 745 | } |
| 746 | // Emit all accumulated events once the graph reaches quiescence |
| 747 | this.flushPendingChanges() |
| 748 | } finally { |
| 749 | this.isGraphRunning = false |
| 750 | // If dispose() was called during this graph run, it deferred the heavy |
| 751 | // cleanup (clearing graph/inputs/pipeline) to avoid nulling references |
| 752 | // mid-loop. Complete that cleanup now. |
| 753 | if (this.deferredCleanup) { |
| 754 | this.deferredCleanup = false |
| 755 | this.finalCleanup() |
| 756 | } |
| 757 | } |
| 758 | } |
| 759 | |
| 760 | /** Classify accumulated changes into DeltaEvents and invoke the callback */ |
| 761 | private flushPendingChanges(): void { |
no test coverage detected