Function readableStreamToPullUnsafe

packages/effect/src/Channel.ts:1818–1852

(options: {
  readonly scope: Scope.Scope
  readonly exit?: MutableRef.MutableRef<Exit.Exit<never, E | E2 | Cause.Done> | undefined> | undefined
  readonly readable: ReadableStream<A>
  readonly onError: (error: unknown) => E
  readonly releaseLockOnEnd?: boolean | undefined
})

  })

const readableStreamToPullUnsafe = <A, E, E2 = never>(options: {
  readonly scope: Scope.Scope
  readonly exit?: MutableRef.MutableRef<Exit.Exit<never, E | E2 | Cause.Done> | undefined> | undefined
  readonly readable: ReadableStream<A>
  readonly onError: (error: unknown) => E
  readonly releaseLockOnEnd?: boolean | undefined
}): Effect.Effect<Pull.Pull<Arr.NonEmptyReadonlyArray<A>, E | E2>, never> => {
  const reader = options.readable.getReader()
  const exit = options.exit ?? MutableRef.make(undefined)
  const pull = Effect.suspend(() => {
    if (exit.current) return exit.current
    return Effect.matchCauseEffect(
      Effect.tryPromise({
        try: () => reader.read(),
        catch: options.onError
      }),
      {
        onFailure: (cause) => exit.current ?? Effect.failCause(cause),
        onSuccess: ({ done, value }) => {
          if (exit.current) return exit.current
          return done ? Cause.done() : Effect.succeed(Arr.of(value))
        }
      }
    )
  })
  return Effect.as(
    Scope.addFinalizer(
      options.scope,
      options.releaseLockOnEnd
        ? Effect.sync(() => reader.releaseLock())
        : Effect.promise(() => reader.cancel().catch(constVoid))
    ),
    pull
  )
}

/**
 * Creates a channel that pulls values from an `AsyncIterable`.