Function race

packages/effect/src/internal/sink.ts:1660–1695

(scope: Scope.Scope)

    }
  ): Sink.Sink<A3 | A4, In & In2, L2 | L, E2 | E, R2 | R> => {
    function race(scope: Scope.Scope) {
      return Effect.gen(function*() {
        const pubsub = yield* PubSub.bounded<
          Either.Either<Chunk.Chunk<In & In2>, Exit.Exit<unknown>>
        >(options?.capacity ?? 16)
        const subscription1 = yield* Scope.extend(PubSub.subscribe(pubsub), scope)
        const subscription2 = yield* Scope.extend(PubSub.subscribe(pubsub), scope)
        const reader = channel.toPubSub(pubsub)
        const writer = channel.fromQueue(subscription1).pipe(
          core.pipeTo(toChannel(self)),
          channel.zipLeft(core.fromEffect(Queue.shutdown(subscription1))),
          channel.mergeWith({
            other: channel.fromQueue(subscription2).pipe(
              core.pipeTo(toChannel(options.other)),
              channel.zipLeft(core.fromEffect(Queue.shutdown(subscription2)))
            ),
            onSelfDone: options.onSelfDone,
            onOtherDone: options.onOtherDone
          })
        )
        const racedChannel = channel.mergeWith(reader, {
          other: writer,
          onSelfDone: () => mergeDecision.Await(identity),
          onOtherDone: (exit) => mergeDecision.Done(exit)
        }) as Channel.Channel<
          Chunk.Chunk<L | L2>,
          Chunk.Chunk<In & In2>,
          E | E2,
          never,
          A3 | A4,
          unknown,
          R | R2
        >
        return new SinkImpl(racedChannel)
      })
    }
    return unwrapScopedWith(race)
  }
)