1/**
2 * Adapters between Node streams and Effect streams, channels, and readables.
3 *
4 * This module is the stream boundary for Node APIs. It wraps `Readable` and
5 * `Duplex` values as Effect `Stream`s and `Channel`s, pipes Effect streams
6 * through Node duplex streams, exposes an Effect `Stream` back to Node as a
7 * `Readable`, and collects readable payloads into strings, array buffers, or
8 * `Uint8Array`s with optional byte limits.
9 *
10 * @since 4.0.0
11 */
12import * as Arr from "effect/Array"
13import * as ByteSize from "effect/ByteSize"
14import * as Cause from "effect/Cause"
15import * as Channel from "effect/Channel"
16import * as Context from "effect/Context"
17import * as Effect from "effect/Effect"
18import * as Exit from "effect/Exit"
19import * as Fiber from "effect/Fiber"
20import { dual, type LazyArg } from "effect/Function"
21import * as Latch from "effect/Latch"
22import * as MutableRef from "effect/MutableRef"
23import * as Pull from "effect/Pull"
24import * as Scope from "effect/Scope"
25import * as Stream from "effect/Stream"
26import type { Duplex } from "node:stream"
27import { Readable } from "node:stream"
28import { pullIntoWritable } from "./NodeSink.ts"
29
30/**
31 * Converts a Node readable stream into an Effect `Stream`, reading chunks with
32 * an optional chunk size, mapping stream errors with `onError`, and destroying
33 * the readable on completion unless `closeOnDone` is `false`.
34 *
35 * @category constructors
36 * @since 4.0.0
37 */
38export const fromReadable = <A = Uint8Array, E = Cause.UnknownError>(options: {
39 readonly evaluate: LazyArg<Readable | NodeJS.ReadableStream>
40 readonly onError?: (error: unknown) => E
41 readonly chunkSize?: number | undefined
42 readonly closeOnDone?: boolean | undefined
43}): Stream.Stream<A, E> => Stream.fromChannel(fromReadableChannel<A, E>(options))
44
45/**
46 * Creates a `Channel` that pulls chunks from a Node readable stream, mapping
47 * errors with `onError` and destroying the readable on completion unless
48 * `closeOnDone` is `false`.
49 *
50 * @category constructors
51 * @since 4.0.0
52 */
53export const fromReadableChannel = <A = Uint8Array, E = Cause.UnknownError>(options: {
54 readonly evaluate: LazyArg<Readable | NodeJS.ReadableStream>
55 readonly onError?: (error: unknown) => E
56 readonly chunkSize?: number | undefined
57 readonly closeOnDone?: boolean | undefined
58}): Channel.Channel<Arr.NonEmptyReadonlyArray<A>, E> =>
59 Channel.fromTransform((_, scope) =>
60 readableToPullUnsafe({
61 scope,
62 readable: options.evaluate(),
63 onError: options.onError ?? defaultOnError as any,
64 chunkSize: options.chunkSize,
65 closeOnDone: options.closeOnDone
66 })
67 )
68
69/**
70 * Creates a `Channel` over a Node `Duplex`, writing upstream chunks with
71 * backpressure while emitting chunks read from the duplex and optionally ending
72 * the writable side when upstream completes.
73 *
74 * @category constructors
75 * @since 4.0.0
76 */
77export const fromDuplex = <IE, I = Uint8Array, O = Uint8Array, E = Cause.UnknownError>(
78 options: {
79 readonly evaluate: LazyArg<Duplex>
80 readonly onError?: (error: unknown) => E
81 readonly chunkSize?: number | undefined
82 readonly endOnDone?: boolean | undefined
83 readonly encoding?: BufferEncoding | undefined
84 }
85): Channel.Channel<Arr.NonEmptyReadonlyArray<O>, IE | E, void, Arr.NonEmptyReadonlyArray<I>, IE> =>
86 Channel.fromTransform((upstream, scope) => {
87 const duplex = options.evaluate()
88 const exit = MutableRef.make<Exit.Exit<never, IE | E | Cause.Done> | undefined>(undefined)
89 const latch = Latch.makeUnsafe(false)
90
91 return pullIntoWritable({
92 pull: upstream,
93 writable: duplex,
94 onError: options.onError ?? defaultOnError as any,
95 endOnDone: options.endOnDone,
96 encoding: options.encoding
97 }).pipe(
98 Effect.catchCause((cause) => {
99 if (Pull.isDoneCause(cause)) return Effect.void
100 exit.current = Exit.failCause(cause as Cause.Cause<IE | E | Cause.Done>)
101 latch.openUnsafe()
102 return Effect.void
103 }),
104 Effect.forkIn(scope),
105 Effect.flatMap(() =>
106 readableToPullUnsafe({
107 scope,
108 exit,
109 latch,
110 readable: duplex,
111 onError: options.onError ?? defaultOnError as any,
112 chunkSize: options.chunkSize
113 })
114 )
115 )
116 })
117
118/**
119 * Pipes an Effect `Stream` through a Node `Duplex`, writing the stream's
120 * chunks to the duplex and emitting chunks read back from it.
121 *
122 * @category combinators
123 * @since 4.0.0
124 */
125export const pipeThroughDuplex: {
126 <B = Uint8Array, E2 = Cause.UnknownError>(
127 options: {
128 readonly evaluate: LazyArg<Duplex>
129 readonly onError?: (error: unknown) => E2
130 readonly chunkSize?: number | undefined
131 readonly endOnDone?: boolean | undefined
132 readonly encoding?: BufferEncoding | undefined
133 }
134 ): <R, E, A>(self: Stream.Stream<A, E, R>) => Stream.Stream<B, E2 | E, R>
135 <R, E, A, B = Uint8Array, E2 = Cause.UnknownError>(
136 self: Stream.Stream<A, E, R>,
137 options: {
138 readonly evaluate: LazyArg<Duplex>
139 readonly onError?: (error: unknown) => E2
140 readonly chunkSize?: number | undefined
141 readonly endOnDone?: boolean | undefined
142 readonly encoding?: BufferEncoding | undefined
143 }
144 ): Stream.Stream<B, E | E2, R>
145} = dual(2, <R, E, A, B = Uint8Array, E2 = Cause.UnknownError>(
146 self: Stream.Stream<A, E, R>,
147 options: {
148 readonly evaluate: LazyArg<Duplex>
149 readonly onError?: (error: unknown) => E2
150 readonly chunkSize?: number | undefined
151 readonly endOnDone?: boolean | undefined
152 readonly encoding?: BufferEncoding | undefined
153 }
154): Stream.Stream<B, E | E2, R> =>
155 Stream.pipeThroughChannelOrFail(
156 self,
157 fromDuplex(options)
158 ))
159
160/**
161 * Pipes a stream of strings or bytes through a Node `Duplex` using default
162 * options and `Cause.UnknownError` for stream failures.
163 *
164 * @category combinators
165 * @since 4.0.0
166 */
167export const pipeThroughSimple: {
168 (
169 duplex: LazyArg<Duplex>
170 ): <R, E>(self: Stream.Stream<string | Uint8Array, E, R>) => Stream.Stream<Uint8Array, E | Cause.UnknownError, R>
171 <R, E>(
172 self: Stream.Stream<string | Uint8Array, E, R>,
173 duplex: LazyArg<Duplex>
174 ): Stream.Stream<Uint8Array, Cause.UnknownError | E, R>
175} = dual(2, <R, E>(
176 self: Stream.Stream<string | Uint8Array, E, R>,
177 duplex: LazyArg<Duplex>
178): Stream.Stream<Uint8Array, Cause.UnknownError | E, R> => pipeThroughDuplex(self, { evaluate: duplex }))
179
180/**
181 * Converts an Effect `Stream` into a Node `Readable`, using the caller's
182 * Effect context to run the stream and destroying the readable if the stream
183 * fails.
184 *
185 * @category converting
186 * @since 4.0.0
187 */
188export const toReadable = <E, R>(stream: Stream.Stream<string | Uint8Array, E, R>): Effect.Effect<Readable, never, R> =>
189 Effect.map(
190 Effect.context<R>(),
191 (context) => new StreamAdapter(context, stream)
192 )
193
194/**
195 * Converts a service-free Effect `Stream` into a Node `Readable` using an
196 * empty Effect context.
197 *
198 * @category converting
199 * @since 4.0.0
200 */
201export const toReadableNever = <E>(stream: Stream.Stream<string | Uint8Array, E, never>): Readable =>
202 new StreamAdapter(
203 Context.empty(),
204 stream
205 )
206
207/**
208 * Consumes a Node readable stream into a string using the selected encoding,
209 * failing through `onError` on stream errors or when `maxBytes` is exceeded
210 * and destroying the stream on interruption or failure.
211 *
212 * @category converting
213 * @since 4.0.0
214 */
215export const toString = <E = Cause.UnknownError>(
216 readable: LazyArg<Readable | NodeJS.ReadableStream>,
217 options?: {
218 readonly onError?: (error: unknown) => E
219 readonly encoding?: BufferEncoding | undefined
220 readonly maxBytes?: ByteSize.Input | undefined
221 }
222): Effect.Effect<string, E> => {
223 const maxBytesNumber = toMaxBytes(options?.maxBytes)
224 const onError = options?.onError ?? defaultOnError
225 const encoding = options?.encoding ?? "utf8"
226 return Effect.callback((resume) => {
227 const stream = readable() as Readable
228 stream.setEncoding(encoding)
229
230 stream.once("error", (err) => {
231 if ("closed" in stream && !stream.closed) {
232 stream.destroy()
233 }
234 resume(Effect.fail(onError(err) as E))
235 })
236
237 let string = ""
238 let bytes = 0
239 stream.once("end", () => {
240 resume(Effect.succeed(string))
241 })
242 stream.on("data", (chunk) => {
243 string += chunk
244 bytes += Buffer.byteLength(chunk)
245 if (maxBytesNumber !== undefined && bytes > maxBytesNumber) {
246 if ("closed" in stream && !stream.closed) {
247 stream.destroy()
248 }
249 resume(Effect.fail(onError(new Error("maxBytes exceeded")) as E))
250 }
251 })
252 return Effect.sync(() => {
253 if ("closed" in stream && !stream.closed) {
254 stream.destroy()
255 }
256 })
257 })
258}
259
260/**
261 * Consumes a Node readable stream into an `ArrayBuffer`, failing through
262 * `onError` on stream errors or when `maxBytes` is exceeded and destroying the
263 * stream on interruption or failure.
264 *
265 * @category converting
266 * @since 4.0.0
267 */
268export const toArrayBuffer = <E = Cause.UnknownError>(
269 readable: LazyArg<Readable | NodeJS.ReadableStream>,
270 options?: {
271 readonly onError?: (error: unknown) => E
272 readonly maxBytes?: ByteSize.Input | undefined
273 }
274): Effect.Effect<ArrayBuffer, E> => {
275 const maxBytesNumber = toMaxBytes(options?.maxBytes)
276 const onError = options?.onError ?? defaultOnError
277 return Effect.callback((resume) => {
278 const stream = readable() as Readable
279 const buffers: Array<Uint8Array> = []
280 let bytes = 0
281 stream.once("error", (err) => {
282 if ("closed" in stream && !stream.closed) {
283 stream.destroy()
284 }
285 resume(Effect.fail(onError(err) as E))
286 })
287 stream.once("end", () => {
288 const buffer = buffers.length === 1 ? buffers[0] : Buffer.concat(buffers)
289 if (buffer.byteOffset === 0 && buffer.buffer.byteLength === buffer.byteLength) {
290 return resume(Effect.succeed(buffer.buffer as ArrayBuffer))
291 }
292 resume(
293 Effect.succeed(buffer.buffer.slice(buffer.byteOffset, buffer.byteOffset + buffer.byteLength) as ArrayBuffer)
294 )
295 })
296 stream.on("data", (chunk) => {
297 buffers.push(chunk)
298 bytes += chunk.length
299 if (maxBytesNumber !== undefined && bytes > maxBytesNumber) {
300 if ("closed" in stream && !stream.closed) {
301 stream.destroy()
302 }
303 resume(Effect.fail(onError(new Error("maxBytes exceeded")) as E))
304 }
305 })
306 return Effect.sync(() => {
307 if ("closed" in stream && !stream.closed) {
308 stream.destroy()
309 }
310 })
311 })
312}
313
314/**
315 * Consumes a Node readable stream into a `Uint8Array`, using the same error
316 * mapping and `maxBytes` handling as `toArrayBuffer`.
317 *
318 * @category converting
319 * @since 4.0.0
320 */
321export const toUint8Array = <E = Cause.UnknownError>(
322 readable: LazyArg<Readable | NodeJS.ReadableStream>,
323 options?: {
324 readonly onError?: (error: unknown) => E
325 readonly maxBytes?: ByteSize.Input | undefined
326 }
327): Effect.Effect<Uint8Array, E> => Effect.map(toArrayBuffer(readable, options), (buffer) => new Uint8Array(buffer))
328
329// ----------------------------------------------------------------------------
330// internal
331// ----------------------------------------------------------------------------
332
333const readableToPullUnsafe = <A, E>(options: {
334 readonly scope: Scope.Scope
335 readonly exit?: MutableRef.MutableRef<Exit.Exit<never, E | Cause.Done> | undefined> | undefined
336 readonly latch?: Latch.Latch | undefined
337 readonly readable: Readable | NodeJS.ReadableStream
338 readonly onError: (error: unknown) => E
339 readonly chunkSize: number | undefined
340 readonly closeOnDone?: boolean | undefined
341}) => {
342 const readable = options.readable as Readable
343
344 const closeOnDone = options.closeOnDone ?? true
345 const exit = options.exit ?? MutableRef.make(undefined)
346 const latch = options.latch ?? Latch.makeUnsafe(false)
347 function onReadable() {
348 latch.openUnsafe()
349 }
350 function onError(error: unknown) {
351 exit.current = Exit.fail(options.onError(error))
352 latch.openUnsafe()
353 }
354 function onEnd() {
355 exit.current = Exit.fail(Cause.Done())
356 latch.openUnsafe()
357 }
358 readable.on("readable", onReadable)
359 readable.once("error", onError)
360 readable.once("end", onEnd)
361
362 const pull = Effect.suspend(function loop(): Pull.Pull<Arr.NonEmptyReadonlyArray<A>, E> {
363 let item = options.readable.read(options.chunkSize) as A | null
364 if (item === null) {
365 if (exit.current) {
366 return exit.current
367 }
368 if (readable.readableEnded) {
369 return Effect.fail(Cause.Done())
370 }
371 latch.closeUnsafe()
372 return Effect.flatMap(latch.await, loop)
373 }
374 const chunk = Arr.of(item as A)
375 while (true) {
376 item = options.readable.read(options.chunkSize)
377 if (item === null) break
378 chunk.push(item)
379 }
380 return Effect.succeed(chunk)
381 })
382
383 return Effect.as(
384 Scope.addFinalizer(
385 options.scope,
386 Effect.sync(() => {
387 readable.off("readable", onReadable)
388 readable.off("error", onError)
389 readable.off("end", onEnd)
390 if (closeOnDone && "closed" in options.readable && !options.readable.closed) {
391 options.readable.destroy()
392 }
393 })
394 ),
395 pull
396 )
397}
398
399class StreamAdapter<E, R> extends Readable {
400 private readonly readLatch: Latch.Latch
401 private fiber: Fiber.Fiber<void, E> | undefined = undefined
402
403 constructor(
404 context: Context.Context<R>,
405 stream: Stream.Stream<Uint8Array | string, E, R>
406 ) {
407 super({})
408 this.readLatch = Latch.makeUnsafe(false)
409 this.fiber = Stream.runForEachArray(stream, (chunk) =>
410 this.readLatch.whenOpen(Effect.sync(() => {
411 this.readLatch.closeUnsafe()
412 for (let i = 0; i < chunk.length; i++) {
413 const item = chunk[i]
414 if (typeof item === "string") {
415 this.push(item, "utf8")
416 } else {
417 this.push(item)
418 }
419 }
420 }))).pipe(
421 this.readLatch.whenOpen,
422 Effect.provideContext(context),
423 Effect.runFork
424 )
425 this.fiber.addObserver((exit) => {
426 this.fiber = undefined
427 if (Exit.isSuccess(exit)) {
428 this.push(null)
429 } else {
430 this.destroy(Cause.squash(exit.cause) as any)
431 }
432 })
433 }
434
435 override _read(_size: number): void {
436 this.readLatch.openUnsafe()
437 }
438
439 override _destroy(error: Error | null, callback: (error?: Error | null | undefined) => void): void {
440 if (!this.fiber) {
441 return callback(error)
442 }
443 Effect.runFork(Fiber.interrupt(this.fiber)).addObserver((exit) => {
444 callback(exit._tag === "Failure" ? Cause.squash(exit.cause) as any : error)
445 })
446 }
447}
448
449const defaultOnError = (error: unknown): Cause.UnknownError => new Cause.UnknownError(error)
450
451const toMaxBytes = (maxBytes: ByteSize.Input | undefined): number | undefined =>
452 maxBytes === undefined || maxBytes === Infinity ? undefined : Number(ByteSize.fromInputUnsafe(maxBytes))