Function mapEffectConcurrent

packages/effect/src/Channel.ts:2113–2210

(
  self: Channel<OutElem, OutErr, OutDone, InElem, InErr, InDone, Env>,
  f: (o: OutElem, i: number) => Effect.Effect<OutElem2, EX, RX>,
  options: {
    readonly concurrency: number | "unbounded"
    readonly unordered?: boolean | undefined
  }
)

  })

const mapEffectConcurrent = <
  OutElem,
  OutErr,
  OutDone,
  InElem,
  InErr,
  InDone,
  Env,
  OutElem2,
  EX,
  RX
>(
  self: Channel<OutElem, OutErr, OutDone, InElem, InErr, InDone, Env>,
  f: (o: OutElem, i: number) => Effect.Effect<OutElem2, EX, RX>,
  options: {
    readonly concurrency: number | "unbounded"
    readonly unordered?: boolean | undefined
  }
): Channel<OutElem2, OutErr | EX, OutDone, InElem, InErr, InDone, Env | RX> =>
  fromTransformBracket(
    Effect.fnUntraced(function*(upstream, scope, forkedScope) {
      let i = 0
      const pull = yield* toTransform(self)(upstream, scope)
      const concurrencyN = options.concurrency === "unbounded"
        ? Number.MAX_SAFE_INTEGER
        : options.concurrency
      const queue = yield* Queue.bounded<OutElem2, OutErr | EX | Cause.Done<OutDone>>(0)
      yield* Scope.addFinalizer(forkedScope, Queue.shutdown(queue))

      const runFork = Effect.runForkWith(yield* Effect.context<RX>())
      const trackFiber = Fiber.runIn(forkedScope)

      if (options.unordered) {
        const semaphore = Semaphore.makeUnsafe(concurrencyN)
        const release = constant(semaphore.release(1))
        const handle = Effect.matchCauseEffect({
          onFailure: (cause: Cause.Cause<EX>) => Effect.flatMap(Queue.failCause(queue, cause), release),
          onSuccess: (value: OutElem2) => Effect.flatMap(Queue.offer(queue, value), release)
        })
        yield* semaphore.take(1).pipe(
          Effect.flatMap(() => pull),
          Effect.flatMap((value) => {
            trackFiber(runFork(handle(f(value, i++))))
            return Effect.void
          }),
          Effect.forever({ disableYield: true }),
          Effect.catchCause((cause) =>
            semaphore.withPermits(concurrencyN - 1)(
              Queue.failCause(queue, cause)
            )
          ),
          Effect.forkIn(forkedScope)
        )
      } else {
        // capacity is n - 2 because
        // - 1 for the offer *after* starting a fiber
        // - 1 for the current processing fiber
        const effects = yield* Queue.bounded<