Function terminate

packages/effect/src/internal/stream.ts:1137–1151

(take: Take.Take<A, E>)

    ref: Ref.Ref<Deferred.Deferred<void>>
  ): Channel.Channel<never, Chunk.Chunk<A>, never, E, unknown, unknown, R> => {
    const terminate = (take: Take.Take<A, E>): Channel.Channel<never, Chunk.Chunk<A>, never, E, unknown, unknown, R> =>
      pipe(
        Ref.get(ref),
        Effect.tap(Deferred.await),
        Effect.zipRight(Deferred.make<void>()),
        Effect.flatMap((deferred) =>
          pipe(
            Queue.offer(queue, [take, deferred] as const),
            Effect.zipRight(Ref.set(ref, deferred)),
            Effect.zipRight(Deferred.await(deferred))
          )
        ),
        Effect.asVoid,
        core.fromEffect
      )
    return core.readWithCause({
      onInput: (input: Chunk.Chunk<A>) =>
        pipe(