1/**
2 * Sink adapters for writing Effect chunks into Node writable streams.
3 *
4 * `fromWritable` creates a `Sink`, `fromWritableChannel` creates a lower-level
5 * `Channel`, and `pullIntoWritable` writes from an existing pull loop. All
6 * three adapters respect writable-stream backpressure, map writable errors with
7 * the supplied `onError` function, and can end the writable when the upstream
8 * data is done.
9 *
10 * @since 4.0.0
11 */
12import type { NonEmptyReadonlyArray } from "effect/Array"
13import * as Cause from "effect/Cause"
14import * as Channel from "effect/Channel"
15import * as Effect from "effect/Effect"
16import { identity, type LazyArg } from "effect/Function"
17import * as Pull from "effect/Pull"
18import * as Sink from "effect/Sink"
19import type { Writable } from "node:stream"
20
21/**
22 * Creates a `Sink` that writes chunks to a Node writable stream, respecting
23 * backpressure, mapping writable errors with `onError`, and ending the stream
24 * on completion unless `endOnDone` is `false`.
25 *
26 * @category constructors
27 * @since 4.0.0
28 */
29export const fromWritable = <E, A = Uint8Array | string>(
30 options: {
31 readonly evaluate: LazyArg<Writable | NodeJS.WritableStream>
32 readonly onError: (error: unknown) => E
33 readonly endOnDone?: boolean | undefined
34 readonly encoding?: BufferEncoding | undefined
35 }
36): Sink.Sink<void, A, never, E> =>
37 Sink.fromChannel(Channel.mapDone(fromWritableChannel<never, E, A>(options), (_) => [_]))
38
39/**
40 * Creates a `Channel` that pulls chunks from upstream and writes them to a
41 * Node writable stream, respecting backpressure and optionally ending the
42 * writable when upstream is done.
43 *
44 * @category constructors
45 * @since 4.0.0
46 */
47export const fromWritableChannel = <IE, E, A = Uint8Array | string>(
48 options: {
49 readonly evaluate: LazyArg<Writable | NodeJS.WritableStream>
50 readonly onError: (error: unknown) => E
51 readonly endOnDone?: boolean | undefined
52 readonly encoding?: BufferEncoding | undefined
53 }
54): Channel.Channel<never, IE | E, void, NonEmptyReadonlyArray<A>, IE> =>
55 Channel.fromTransform((pull: Pull.Pull<NonEmptyReadonlyArray<A>, IE, unknown>) => {
56 const writable = options.evaluate() as Writable
57 return Effect.succeed(pullIntoWritable({ ...options, writable, pull }))
58 })
59
60/**
61 * Writes Effect chunks into a Node writable stream.
62 *
63 * **When to use**
64 *
65 * Use to implement custom Node stream adapters that already have an upstream
66 * pull and need direct control over a writable stream.
67 *
68 * **Details**
69 *
70 * The loop waits for `drain` when needed, fails on writable errors, and ends
71 * the writable on upstream completion unless `endOnDone` is `false`.
72 *
73 * @category converting
74 * @since 4.0.0
75 */
76export const pullIntoWritable = <A, IE, E>(options: {
77 readonly pull: Pull.Pull<NonEmptyReadonlyArray<A>, IE, unknown>
78 readonly writable: Writable
79 readonly onError: (error: unknown) => E
80 readonly endOnDone?: boolean | undefined
81 readonly encoding?: BufferEncoding | undefined
82}): Pull.Pull<never, IE | E, unknown> =>
83 options.pull.pipe(
84 Effect.flatMap((chunk) => {
85 let i = 0
86 return Effect.callback<void, E>((resume) => {
87 let cancelled = false
88 const loop = () => {
89 for (; i < chunk.length;) {
90 if (cancelled) {
91 return
92 }
93 const success = options.writable.write(chunk[i++], options.encoding as any)
94 if (!success) {
95 if (!cancelled) {
96 options.writable.once("drain", loop)
97 }
98 return
99 }
100 }
101 if (!cancelled) {
102 resume(Effect.void)
103 }
104 }
105 loop()
106 return Effect.sync(() => {
107 cancelled = true
108 options.writable.off("drain", loop)
109 })
110 })
111 }),
112 Effect.forever({ disableYield: true }),
113 options.endOnDone !== false ?
114 Pull.catchDone((_) => {
115 if ("closed" in options.writable && options.writable.closed) {
116 return Cause.done(_)
117 }
118 return Effect.callback<never, E | Cause.Done<unknown>>((resume) => {
119 const onFinish = () => resume(Cause.done(_))
120 options.writable.once("finish", onFinish)
121 options.writable.end()
122 return Effect.sync(() => {
123 options.writable.off("finish", onFinish)
124 })
125 })
126 }) :
127 identity,
128 Effect.raceFirst(Effect.callback<never, E>((resume) => {
129 const onError = (error: unknown) => resume(Effect.fail(options.onError(error)))
130 options.writable.once("error", onError)
131 return Effect.sync(() => {
132 options.writable.off("error", onError)
133 })
134 }))
135 )