Function scheduledAggregator

packages/effect/src/internal/stream.ts:290–424

(
          sinkFiber: Fiber.RuntimeFiber<readonly [Chunk.Chunk<Chunk.Chunk<A | A2>>, B], E | E2>,
          scheduleFiber: Fiber.RuntimeFiber<C, Option.Option<never>>,
          scope: Scope.Scope
        )

        const timeout = (lastB: Option.Option<B>): Effect.Effect<C, Option.Option<never>, R2 | R3> =>
          scheduleDriver.next(lastB)
        const scheduledAggregator = (
          sinkFiber: Fiber.RuntimeFiber<readonly [Chunk.Chunk<Chunk.Chunk<A | A2>>, B], E | E2>,
          scheduleFiber: Fiber.RuntimeFiber<C, Option.Option<never>>,
          scope: Scope.Scope
        ): Channel.Channel<Chunk.Chunk<Either.Either<B, C>>, unknown, E | E2, unknown, unknown, unknown, R2 | R3> => {
          const forkSink = pipe(
            Ref.set(consumed, false),
            Effect.zipRight(Ref.set(endAfterEmit, false)),
            Effect.zipRight(
              pipe(
                handoffConsumer,
                channel.pipeToOrFail(sink_.toChannel(sink)),
                core.collectElements,
                channel.run,
                Effect.forkIn(scope)
              )
            )
          )
          const handleSide = (
            leftovers: Chunk.Chunk<Chunk.Chunk<A | A2>>,
            b: B,
            c: Option.Option<C>
          ): Channel.Channel<Chunk.Chunk<Either.Either<B, C>>, unknown, E | E2, unknown, unknown, unknown, R2 | R3> =>
            pipe(
              Ref.set(sinkLeftovers, Chunk.flatten(leftovers)),
              Effect.zipRight(
                Effect.map(Ref.get(sinkEndReason), (reason) => {
                  switch (reason._tag) {
                    case SinkEndReason.OP_SCHEDULE_END: {
                      return pipe(
                        Effect.all([
                          Ref.get(consumed),
                          forkSink,
                          pipe(timeout(Option.some(b)), Effect.forkIn(scope))
                        ]),
                        Effect.map(([wasConsumed, sinkFiber, scheduleFiber]) => {
                          const toWrite = pipe(
                            c,
                            Option.match({
                              onNone: (): Chunk.Chunk<Either.Either<B, C>> => Chunk.of(Either.right(b)),
                              onSome: (c): Chunk.Chunk<Either.Either<B, C>> =>
                                Chunk.make(Either.right(b), Either.left(c))
                            })
                          )
                          if (wasConsumed) {
                            return pipe(
                              core.write(toWrite),
                              core.flatMap(() => scheduledAggregator(sinkFiber, scheduleFiber, scope))
                            )
                          }
                          return scheduledAggregator(sinkFiber, scheduleFiber, scope)
                        }),
                        channel.unwrap
                      )
                    }
                    case SinkEndReason.OP_UPSTREAM_END: {
                      return pipe(
                        Ref.get(consumed),