// ets_tracing: off import type * as CK from "../../../Collections/Immutable/Chunk/index.js" import * as T from "../../../Effect/index.js" import * as E from "../../../Either/index.js" import type * as Ex from "../../../Exit/index.js" import { pipe } from "../../../Function/index.js" import * as H from "../../../Hub/index.js" import * as M from "../../../Managed/index.js" import * as MH from "../Channel/_internal/mergeHelpers.js" import * as CH from "../Channel/index.js" import * as C from "./core.js" import * as UnwrapManaged from "./unwrapManaged.js" /** * Runs both sinks in parallel on the input, returning the result or the error from the * one that finishes first. */ export function raceBoth_( self: C.Sink, that: C.Sink, capacity = 16 ): C.Sink> { const managed = pipe( M.do, M.bind("hub", () => T.toManaged( H.makeBounded, CK.Chunk>>( capacity ) ) ), M.bind("c1", ({ hub }) => CH.fromHubManaged(hub)), M.bind("c2", ({ hub }) => CH.fromHubManaged(hub)), M.let("reader", ({ hub }) => CH.toHub(hub)), M.let("writer", ({ c1, c2 }) => CH.mergeWith_( c1[">>>"](self.channel), c2[">>>"](that.channel), (selfDone) => MH.done(T.map_(T.done(selfDone), E.left)), (thatDone) => MH.done(T.map_(T.done(thatDone), E.right)) ) ), M.let("channel", ({ reader, writer }) => CH.mergeWith_( reader, writer as CH.Channel< R1 & R, unknown, unknown, unknown, OutErr | OutErr1, CK.Chunk, E.Either >, (_) => MH.await_((_) => T.done(_)), (done) => MH.done(T.done(done)) ) ), M.map(({ channel }) => new C.Sink(channel)) ) return UnwrapManaged.unwrapManaged(managed) } /** * Runs both sinks in parallel on the input, returning the result or the error from the * one that finishes first. * * @ets_data_first orElse_ */ export function raceBoth( that: C.Sink, capacity = 16 ) { return (self: C.Sink) => raceBoth_(self, that, capacity) }