Function finalize

packages/effect/src/internal/stream.ts:2175–2215

(endTake: Exit.Exit<never, Option.Option<E>>)

          )
        )
        const finalize = (endTake: Exit.Exit<never, Option.Option<E>>): Effect.Effect<void> =>
          // Make sure that no queues are currently being added
          queuesLock.withPermits(1)(
            pipe(
              Ref.set(
                newQueue,
                pipe(
                  // All newly created queues should end immediately
                  Queue.bounded<Exit.Exit<A, Option.Option<E>>>(1),
                  Effect.tap((queue) => Queue.offer(queue, endTake)),
                  Effect.flatMap((queue) => {
                    const id = newDistributedWithDynamicId()
                    return pipe(
                      Ref.update(queuesRef, (map) => map.set(id, queue)),
                      Effect.as(Tuple.make(id, queue))
                    )
                  })
                )
              ),
              Effect.zipRight(
                pipe(
                  Ref.get(queuesRef),
                  Effect.flatMap((map) =>
                    pipe(
                      Chunk.fromIterable(map.values()),
                      Effect.forEach((queue) =>
                        pipe(
                          Queue.offer(queue, endTake),
                          Effect.catchSomeCause((cause) =>
                            Cause.isInterrupted(cause) ? Option.some(Effect.void) : Option.none()
                          )
                        )
                      )
                    )
                  )
                )
              ),
              Effect.zipRight(done(endTake)),
              Effect.asVoid
            )
          )
        yield* pipe(
          self,
          runForEachScoped(offer),