Function fromReadableStreamByob

packages/effect/src/internal/stream.ts:3297–3324

(
  ...args: [options: {
    readonly evaluate: LazyArg<ReadableStream<Uint8Array>>
    readonly onError: (error: unknown) => E
    readonly bufferSize?: number | undefined
    readonly releaseLockOnEnd?: boolean | undefined
  }] | [
    evaluate: LazyArg<ReadableStream<Uint8Array>>,
    onError: (error: unknown) => E,
    allocSize?: number | undefined
  ]
)

    allocSize?: number
  ): Stream.Stream<Uint8Array, E>
} = <E>(
  ...args: [options: {
    readonly evaluate: LazyArg<ReadableStream<Uint8Array>>
    readonly onError: (error: unknown) => E
    readonly bufferSize?: number | undefined
    readonly releaseLockOnEnd?: boolean | undefined
  }] | [
    evaluate: LazyArg<ReadableStream<Uint8Array>>,
    onError: (error: unknown) => E,
    allocSize?: number | undefined
  ]
): Stream.Stream<Uint8Array, E> => {
  const evaluate = args.length === 1 ? args[0].evaluate : args[0]
  const onError = args.length === 1 ? args[0].onError : args[1]
  const allocSize = (args.length === 1 ? args[0].bufferSize : args[2]) ?? 4096
  const releaseLockOnEnd = args.length === 1 ? args[0].releaseLockOnEnd === true : false
  return unwrapScoped(Effect.map(
    Effect.acquireRelease(
      Effect.sync(() => evaluate().getReader({ mode: "byob" })),
      (reader) => releaseLockOnEnd ? Effect.sync(() => reader.releaseLock()) : Effect.promise(() => reader.cancel())
    ),
    (reader) =>
      catchAll(
        forever(readChunkStreamByobReader(reader, onError, allocSize)),
        (error) => error === EOF ? empty : fail(error)
      )
  ))
}

const EOF = Symbol.for("effect/Stream/EOF")