Hyperlinkv0.8.0-beta.28

← Channel

Channel.mergeconsteffect/Channel.ts:6130
<OutElem1, OutErr1, OutDone1, InElem1, InErr1, InDone1, Env1>(
  right: Channel<
    OutElem1,
    OutErr1,
    OutDone1,
    InElem1,
    InErr1,
    InDone1,
    Env1
  >,
  options?:
    | { readonly haltStrategy?: HaltStrategy | undefined }
    | undefined
): <OutElem, OutErr, OutDone, InElem, InErr, InDone, Env>(
  left: Channel<OutElem, OutErr, OutDone, InElem, InErr, InDone, Env>
) => Channel<
  OutElem1 | OutElem,
  OutErr | OutErr1,
  OutDone | OutDone1,
  InElem & InElem1,
  InErr & InErr1,
  InDone & InDone1,
  Env1 | Env
>
<
  OutElem,
  OutErr,
  OutDone,
  InElem,
  InErr,
  InDone,
  Env,
  OutElem1,
  OutErr1,
  OutDone1,
  InElem1,
  InErr1,
  InDone1,
  Env1
>(
  left: Channel<OutElem, OutErr, OutDone, InElem, InErr, InDone, Env>,
  right: Channel<
    OutElem1,
    OutErr1,
    OutDone1,
    InElem1,
    InErr1,
    InDone1,
    Env1
  >,
  options?:
    | { readonly haltStrategy?: HaltStrategy | undefined }
    | undefined
): Channel<
  OutElem | OutElem1,
  OutErr | OutErr1,
  OutDone | OutDone1,
  InElem & InElem1,
  InErr & InErr1,
  InDone & InDone1,
  Env | Env1
>

Returns a new channel, which is the merge of this channel and the specified channel.

Example (Merging channels)

import { Channel, Data } from "effect"

class MergeError extends Data.TaggedError("MergeError")<{
  readonly source: string
}> {}

// Create two channels
const leftChannel = Channel.fromIterable([1, 2, 3])
const rightChannel = Channel.fromIterable(["a", "b", "c"])

// Merge them with "either" halt strategy
const mergedChannel = Channel.merge(leftChannel, rightChannel, {
  haltStrategy: "either"
})

// Outputs elements from both channels concurrently
// Order may vary: 1, "a", 2, "b", 3, "c"
combining
Source effect/Channel.ts:6130135 lines
export const merge: {
  <OutElem1, OutErr1, OutDone1, InElem1, InErr1, InDone1, Env1>(
    right: Channel<OutElem1, OutErr1, OutDone1, InElem1, InErr1, InDone1, Env1>,
    options?: {
      readonly haltStrategy?: HaltStrategy | undefined
    } | undefined
  ): <OutElem, OutErr, OutDone, InElem, InErr, InDone, Env>(
    left: Channel<OutElem, OutErr, OutDone, InElem, InErr, InDone, Env>
  ) => Channel<
    OutElem1 | OutElem,
    OutErr | OutErr1,
    OutDone | OutDone1,
    InElem & InElem1,
    InErr & InErr1,
    InDone & InDone1,
    Env1 | Env
  >
  <
    OutElem,
    OutErr,
    OutDone,
    InElem,
    InErr,
    InDone,
    Env,
    OutElem1,
    OutErr1,
    OutDone1,
    InElem1,
    InErr1,
    InDone1,
    Env1
  >(
    left: Channel<OutElem, OutErr, OutDone, InElem, InErr, InDone, Env>,
    right: Channel<OutElem1, OutErr1, OutDone1, InElem1, InErr1, InDone1, Env1>,
    options?: {
      readonly haltStrategy?: HaltStrategy | undefined
    } | undefined
  ): Channel<
    OutElem | OutElem1,
    OutErr | OutErr1,
    OutDone | OutDone1,
    InElem & InElem1,
    InErr & InErr1,
    InDone & InDone1,
    Env | Env1
  >
} = dual((args) => isChannel(args[0]) && isChannel(args[1]), <
  OutElem,
  OutErr,
  OutDone,
  InElem,
  InErr,
  InDone,
  Env,
  OutElem1,
  OutErr1,
  OutDone1,
  InElem1,
  InErr1,
  InDone1,
  Env1
>(
  left: Channel<OutElem, OutErr, OutDone, InElem, InErr, InDone, Env>,
  right: Channel<OutElem1, OutErr1, OutDone1, InElem1, InErr1, InDone1, Env1>,
  options?: {
    readonly haltStrategy?: HaltStrategy | undefined
  } | undefined
): Channel<
  OutElem | OutElem1,
  OutErr | OutErr1,
  OutDone | OutDone1,
  InElem & InElem1,
  InErr & InErr1,
  InDone & InDone1,
  Env | Env1
> =>
  fromTransformBracket(Effect.fnUntraced(function*(upstream, _scope, forkedScope) {
    const strategy = options?.haltStrategy ?? "both"
    const queue = yield* Queue.bounded<OutElem | OutElem1, OutErr | OutErr1 | Cause.Done<OutDone | OutDone1>>(0)
    yield* Scope.addFinalizer(forkedScope, Queue.shutdown(queue))
    let done = 0
    function onExit(
      side: "left" | "right",
      cause: Cause.Cause<OutErr | OutErr1 | Cause.Done<OutDone | OutDone1>>
    ): Effect.Effect<void> {
      done++
      if (!Pull.isDoneCause(cause)) {
        return Queue.failCause(queue, cause)
      }
      switch (strategy) {
        case "both": {
          return done === 2 ? Queue.failCause(queue, cause) : Effect.void
        }
        case "left":
        case "right": {
          return side === strategy ? Queue.failCause(queue, cause) : Effect.void
        }
        case "either": {
          return Queue.failCause(queue, cause)
        }
      }
    }
    const runSide = (
      side: "left" | "right",
      channel: Channel<
        OutElem | OutElem1,
        OutErr | OutErr1,
        OutDone | OutDone1,
        InElem & InElem1,
        InErr & InErr1,
        InDone & InDone1,
        Env | Env1
      >,
      scope: Scope.Closeable
    ) =>
      toTransform(channel)(upstream, scope).pipe(
        Effect.flatMap((pull) =>
          pull.pipe(
            Effect.flatMap((value) => Queue.offer(queue, value)),
            Effect.forever
          )
        ),
        Effect.onError((cause) =>
          Effect.andThen(
            Scope.close(scope, Pull.doneExitFromCause(cause)),
            onExit(side, cause)
          )
        ),
        Effect.forkIn(forkedScope)
      )
    yield* runSide("left", left, Scope.forkUnsafe(forkedScope))
    yield* runSide("right", right, Scope.forkUnsafe(forkedScope))
    return Queue.take(queue)
  })))
Referenced by 3 symbols
Channel.interruptWhenChannel.mergeEffectStream.merge