Function pullIntoWritableStream

packages/effect/src/Channel.ts:1670–1712

(options: {
  readonly pull: Pull.Pull<Arr.NonEmptyReadonlyArray<A>, IE, unknown>
  readonly writable: WritableStream<A>
  readonly onError: (error: unknown) => E
  readonly closeOnDone?: boolean | undefined
})

/** @internal */
export const pullIntoWritableStream = <A, IE, E>(options: {
  readonly pull: Pull.Pull<Arr.NonEmptyReadonlyArray<A>, IE, unknown>
  readonly writable: WritableStream<A>
  readonly onError: (error: unknown) => E
  readonly closeOnDone?: boolean | undefined
}): Pull.Pull<never, IE | E, unknown> =>
  Effect.acquireUseRelease(
    Effect.sync(() => options.writable.getWriter()),
    (writer) => {
      const loop = options.pull.pipe(
        Effect.flatMap((chunk) =>
          Effect.forEach(
            chunk,
            (value) =>
              Effect.tryPromise({
                try: () => writer.ready.then(() => writer.write(value)),
                catch: options.onError
              }),
            { discard: true }
          )
        ),
        Effect.forever({ disableYield: true })
      )
      const withClose = options.closeOnDone !== false
        ? Pull.catchDone(loop, (done) =>
          Effect.andThen(
            Effect.tryPromise({
              try: () => writer.close(),
              catch: options.onError
            }),
            Cause.done(done)
          ))
        : loop
      return Effect.onError(
        withClose,
        (cause) =>
          Pull.isDoneCause(cause)
            ? Effect.void
            : Effect.promise(() => writer.abort(cause).catch(constVoid))
      )
    },
    (writer) => Effect.sync(() => writer.releaseLock())
  )

/**
 * Creates a channel that writes upstream values to a lazily supplied Web