Function fromQueue

packages/effect/src/internal/stream.ts:3186–3207

(
  queue: Queue.Dequeue<A>,
  options?: {
    readonly maxChunkSize?: number | undefined
    readonly shutdown?: boolean | undefined
  }
)

/** @internal */
export const fromQueue = <A>(
  queue: Queue.Dequeue<A>,
  options?: {
    readonly maxChunkSize?: number | undefined
    readonly shutdown?: boolean | undefined
  }
): Stream.Stream<A> =>
  pipe(
    Queue.takeBetween(queue, 1, options?.maxChunkSize ?? DefaultChunkSize),
    Effect.catchAllCause((cause) =>
      pipe(
        Queue.isShutdown(queue),
        Effect.flatMap((isShutdown) =>
          isShutdown && Cause.isInterrupted(cause) ?
            pull.end() :
            pull.failCause(cause)
        )
      )
    ),
    repeatEffectChunkOption,
    options?.shutdown ? ensuring(Queue.shutdown(queue)) : identity
  )

/** @internal */
export const fromTQueue = <A>(queue: TQueue.TDequeue<A>): Stream.Stream<A> =>