Class DrainChildExecutors

packages/effect/src/internal/channel/subexecutor.ts:155–212

 * @internal
 */
export class DrainChildExecutors<in out R> implements Subexecutor<R> {
  readonly _tag: OP_DRAIN_CHILD_EXECUTORS = OP_DRAIN_CHILD_EXECUTORS

  constructor(
    readonly upstreamExecutor: ErasedExecutor<R>,
    readonly lastDone: unknown,
    readonly activeChildExecutors: ReadonlyArray<PullFromChild<R> | undefined>,
    readonly upstreamDone: Exit.Exit<unknown, unknown>,
    readonly combineChildResults: (x: unknown, y: unknown) => unknown,
    readonly combineWithChildResult: (x: unknown, y: unknown) => unknown,
    readonly onPull: (
      request: UpstreamPullRequest.UpstreamPullRequest<unknown>
    ) => UpstreamPullStrategy.UpstreamPullStrategy<unknown>
  ) {
  }

  close(exit: Exit.Exit<unknown, unknown>): Effect.Effect<unknown, never, R> | undefined {
    const fin1 = this.upstreamExecutor.close(exit)
    const fins = [
      ...this.activeChildExecutors.map((child) => (child !== undefined ?
        child.childExecutor.close(exit) :
        undefined)
      ),
      fin1
    ]
    const result = fins.reduce(
      (acc: Effect.Effect<Exit.Exit<unknown, unknown>, never, R> | undefined, next) => {
        if (acc !== undefined && next !== undefined) {
          return Effect.zipWith(
            acc,
            Effect.exit(next),
            (exit1, exit2) => Exit.zipRight(exit1, exit2)
          )
        } else if (acc !== undefined) {
          return acc
        } else if (next !== undefined) {
          return Effect.exit(next)
        } else {
          return undefined
        }
      },
      undefined
    )
    return result === undefined ? result : result
  }

  enqueuePullFromChild(child: PullFromChild<R>): Subexecutor<R> {
    return new DrainChildExecutors(
      this.upstreamExecutor,
      this.lastDone,
      [...this.activeChildExecutors, child],
      this.upstreamDone,
      this.combineChildResults,
      this.combineWithChildResult,
      this.onPull
    )
  }
}