Function fromQueue

packages/effect/src/Sink.ts:514–526

(
  queue: Queue.Queue<A, Cause.Done>
)

 * @since 2.0.0
 */
export const fromQueue = <A>(
  queue: Queue.Queue<A, Cause.Done>
): Sink<void, A> =>
  fromTransform((upstream) =>
    upstream.pipe(
      Effect.flatMap((arr) => Queue.offerAll(queue, arr)),
      Effect.forever({ disableYield: true }),
      Pull.catchDone((_) => {
        Queue.endUnsafe(queue)
        return endVoid
      })
    )
  )

/**
 * Creates a sink that publishes every consumed input element to a `PubSub`.