Class D2

packages/db-ivm/src/d2.ts:10–64

import type { ID2, IStreamBuilder, PipedOperator } from './types.js'

export class D2 implements ID2 {
  #operators: Array<UnaryOperator<any> | BinaryOperator<any>> = []
  #nextOperatorId = 0
  #finalized = false

  constructor() {}

  #checkNotFinalized(): void {
    if (this.#finalized) {
      throw new Error(`Graph already finalized`)
    }
  }

  getNextOperatorId(): number {
    this.#checkNotFinalized()
    return this.#nextOperatorId++
  }

  newInput<T>(): RootStreamBuilder<T> {
    this.#checkNotFinalized()
    const writer = new DifferenceStreamWriter<T>()
    // Use the root stream builder that exposes the sendData and sendFrontier methods
    const streamBuilder = new RootStreamBuilder<T>(this, writer)
    return streamBuilder
  }

  addOperator(operator: UnaryOperator<any> | BinaryOperator<any>): void {
    this.#checkNotFinalized()
    this.#operators.push(operator)
  }

  finalize() {
    this.#checkNotFinalized()
    this.#finalized = true
  }

  step(): void {
    if (!this.#finalized) {
      throw new Error(`Graph not finalized`)
    }
    for (const op of this.#operators) {
      op.run()
    }
  }

  pendingWork(): boolean {
    return this.#operators.some((op) => op.hasPendingWork())
  }

  run(): void {
    while (this.pendingWork()) {
      this.step()
    }
  }
}

export class StreamBuilder<T> implements IStreamBuilder<T> {
  #graph: ID2