Class FiberStoreImpl

src/internal/fiberStore.ts:17–110

): FiberStore.FiberStore<R, E, A> => new FiberStoreImpl(runtime)

class FiberStoreImpl<R, E, A> implements FiberStore.FiberStore<R, E, A> {
  constructor(
    readonly runtime: Runtime.Runtime<R>
  ) {}

  // listeners
  private listeners: Array<() => void> = []
  private notify() {
    for (let i = 0; i < this.listeners.length; i++) {
      this.listeners[i]()
    }
  }
  public subscribe = (listener: () => void) => {
    this.listeners.push(listener)
    this.maybeResume()
    return () => {
      const index = this.listeners.indexOf(listener)
      if (index >= 0) {
        this.listeners.splice(index, 1)
      }
      queueMicrotask(() => {
        if (this.listeners.length === 0) {
          this.interruptIfRunning()
        }
      })
    }
  }

  // state
  private trackedProps = TrackedProperties.initial()
  private resultBag: ResultBag.ResultBag<E, A> = ResultBag.make(Result.initial(), this.trackedProps)
  private setResult(result: Result.Result<E, A>) {
    TrackedProperties.updateFromResult(this.trackedProps, result)
    this.resultBag = ResultBag.make(result, this.trackedProps)
    this.notify()
  }
  public snapshot = () => {
    return this.resultBag
  }

  // lifecycle
  private stream: Stream.Stream<R, E, A> | undefined = undefined
  private fiberState:
    | {
      readonly fiber: Fiber.RuntimeFiber<never, void>
      readonly interruptedRef: Ref.Ref<boolean>
    }
    | undefined = undefined
  public run(stream: Stream.Stream<R, E, A>) {
    this.interruptIfRunning()
    this.stream = stream

    const interruptedRef = Ref.unsafeMake(false)
    const maybeSetResult = (result: Result.Result<E, A>) =>
      Effect.flatMap(
        Ref.get(interruptedRef),
        (interrupted) =>
          interrupted