Function unsafeReadableRead

packages/platform-node-shared/src/internal/stream.ts:254–311

(
  readable: Readable | NodeJS.ReadableStream,
  onError: (error: unknown) => E,
  exit: MutableRef.MutableRef<Exit.Exit<void, E> | undefined>,
  options: FromReadableOptions | undefined
)

    })

const unsafeReadableRead = <A, E>(
  readable: Readable | NodeJS.ReadableStream,
  onError: (error: unknown) => E,
  exit: MutableRef.MutableRef<Exit.Exit<void, E> | undefined>,
  options: FromReadableOptions | undefined
) => {
  if (!readable.readable) {
    return Channel.void
  }

  const latch = Effect.unsafeMakeLatch(false)
  function onReadable() {
    latch.unsafeOpen()
  }
  function onErr(err: unknown) {
    exit.current = Exit.fail(onError(err))
    latch.unsafeOpen()
  }
  function onEnd() {
    exit.current = Exit.void
    latch.unsafeOpen()
  }
  readable.on("readable", onReadable)
  readable.on("error", onErr)
  readable.on("end", onEnd)

  const chunkSize = options?.chunkSize ? Number(options.chunkSize) : undefined
  const read = Channel.suspend(function loop(): Channel.Channel<Chunk.Chunk<A>, unknown, E> {
    let item = readable.read(chunkSize) as A | null
    if (item === null) {
      if (exit.current) {
        return Channel.fromEffect(exit.current)
      }
      latch.unsafeClose()
      return Channel.flatMap(latch.await, loop)
    }
    const arr = [item as A]
    while (true) {
      item = readable.read(chunkSize)
      if (item === null) {
        return Channel.flatMap(Channel.write(Chunk.unsafeFromArray(arr)), loop)
      }
      arr.push(item as A)
    }
  })

  return Channel.ensuring(
    read,
    Effect.sync(() => {
      readable.off("readable", onReadable)
      readable.off("error", onErr)
      readable.off("end", onEnd)
      if (options?.closeOnDone !== false && "closed" in readable && !readable.closed) {
        readable.destroy()
      }
    })
  )
}