Class PullFromUpstream

packages/effect/src/internal/channel/subexecutor.ts:87–147

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

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

  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 PullFromUpstream(
      this.upstreamExecutor,
      this.createChild,
      this.lastDone,
      [...this.activeChildExecutors, child],
      this.combineChildResults,
      this.combineWithChildResult,
      this.onPull,
      this.onEmit