Function offer

packages/effect/src/internal/stream.ts:2119–2161

(a: A)

    Effect.flatMap((queuesRef) =>
      Effect.gen(function*() {
        const offer = (a: A): Effect.Effect<void> =>
          pipe(
            decide(a),
            Effect.flatMap((shouldProcess) =>
              pipe(
                Ref.get(queuesRef),
                Effect.flatMap((queues) =>
                  pipe(
                    queues.entries(),
                    Effect.reduce(Chunk.empty<number>(), (acc, [id, queue]) => {
                      if (shouldProcess(id)) {
                        return pipe(
                          Queue.offer(queue, Exit.succeed(a)),
                          Effect.matchCauseEffect({
                            onFailure: (cause) =>
                              // Ignore all downstream queues that were shut
                              // down and remove them later
                              Cause.isInterrupted(cause) ?
                                Effect.succeed(pipe(acc, Chunk.prepend(id))) :
                                Effect.failCause(cause),
                            onSuccess: () => Effect.succeed(acc)
                          })
                        )
                      }
                      return Effect.succeed(acc)
                    }),
                    Effect.flatMap((ids) => {
                      if (Chunk.isNonEmpty(ids)) {
                        return Ref.update(queuesRef, (map) => {
                          for (const id of ids) {
                            map.delete(id)
                          }
                          return map
                        })
                      }
                      return Effect.void
                    })
                  )
                )
              )
            ),
            Effect.asVoid
          )
        const queuesLock = yield* Effect.makeSemaphore(1)
        const newQueue = yield* Ref.make<Effect.Effect<[number, Queue.Queue<Exit.Exit<A, Option.Option<E>>>]>>(
          pipe(