1import * as NodeStream from "@effect/platform-node-shared/NodeStream"
2import { assert, describe, it } from "@effect/vitest"
3import { Effect } from "effect"
4import * as Array from "effect/Array"
5import * as ByteSize from "effect/ByteSize"
6import * as Channel from "effect/Channel"
7import * as Console from "effect/Console"
8import * as Stream from "effect/Stream"
9import { Duplex, Readable, Transform } from "node:stream"
10import * as Zlib from "node:zlib"
11
12describe("Stream", () => {
13 it.effect("should read a stream", () =>
14 Effect.gen(function*() {
15 const stream = NodeStream.fromReadable<"error", string>({
16 evaluate: () => Readable.from(["a", "b", "c"]),
17 onError: () => "error"
18 })
19 const items = yield* Stream.runCollect(stream)
20 assert.deepEqual(items, ["a", "b", "c"])
21 }))
22
23 it.effect("fromDuplex", () =>
24 Effect.gen(function*() {
25 const result = yield* Stream.fromArray(["a", "b", "c"]).pipe(
26 Stream.pipeThroughChannelOrFail(NodeStream.fromDuplex({
27 evaluate: () =>
28 new Transform({
29 transform(chunk, _encoding, callback) {
30 callback(null, chunk.toString().toUpperCase())
31 }
32 }),
33 onError: () => "error" as const
34 })),
35 Stream.decodeText(),
36 Stream.mkString
37 )
38
39 assert.strictEqual(result, "ABC")
40 }))
41
42 it.effect("fromDuplex failure", () =>
43 Effect.gen(function*() {
44 const result = yield* Stream.fromArray(["a", "b", "c"]).pipe(
45 Stream.pipeThroughChannelOrFail(NodeStream.fromDuplex({
46 evaluate: () =>
47 new Transform({
48 transform(_chunk, _encoding, callback) {
49 callback(new Error())
50 }
51 }),
52 onError: () => "error" as const
53 })),
54 Stream.runDrain,
55 Effect.flip
56 )
57
58 assert.strictEqual(result, "error")
59 }))
60
61 it.effect("pipeThroughDuplex", () =>
62 Effect.gen(function*() {
63 const result = yield* Stream.fromArray(["a", "b", "c"]).pipe(
64 NodeStream.pipeThroughDuplex({
65 evaluate: () =>
66 new Transform({
67 transform(chunk, _encoding, callback) {
68 callback(null, chunk.toString().toUpperCase())
69 }
70 }),
71 onError: () => "error" as const
72 }),
73 Stream.decodeText(),
74 Stream.mkString
75 )
76
77 assert.strictEqual(result, "ABC")
78 }))
79
80 it.effect("pipeThroughDuplex propagates upstream failure", () =>
81 Effect.gen(function*() {
82 const result = yield* Stream.fail("upstream error").pipe(
83 NodeStream.pipeThroughDuplex({
84 evaluate: () =>
85 new Duplex({
86 read() {},
87 write(_chunk, _encoding, callback) {
88 callback()
89 }
90 })
91 }),
92 Stream.runDrain,
93 Effect.flip
94 )
95
96 assert.strictEqual(result, "upstream error")
97 }))
98
99 it.effect("pipeThroughDuplex write error", () =>
100 Effect.gen(function*() {
101 const result = yield* Stream.fromArray(["a", "b", "c"]).pipe(
102 NodeStream.pipeThroughDuplex({
103 evaluate: () =>
104 new Duplex({
105 read() {},
106 write(_chunk, _encoding, callback) {
107 callback(new Error())
108 }
109 }),
110 onError: () => "error" as const
111 }),
112 Stream.runDrain,
113 Effect.flip
114 )
115 assert.strictEqual(result, "error")
116 }))
117
118 it.effect("pipeThroughSimple", () =>
119 Effect.gen(function*() {
120 const result = yield* Stream.fromArray(["a", Buffer.from("b"), "c"]).pipe(
121 NodeStream.pipeThroughSimple(
122 () =>
123 new Transform({
124 transform(chunk, _encoding, callback) {
125 callback(null, chunk.toString().toUpperCase())
126 }
127 })
128 ),
129 Stream.decodeText(),
130 Stream.mkString
131 )
132
133 assert.strictEqual(result, "ABC")
134 }))
135
136 it.effect("fromDuplex should work with node:zlib", () =>
137 Effect.gen(function*() {
138 const text = "abcdefg1234567890"
139 const encoder = new TextEncoder()
140 const input = encoder.encode(text)
141 const stream = NodeStream.fromReadable<Uint8Array, "error">({
142 evaluate: () => Readable.from([input]),
143 onError: () => "error"
144 })
145 const deflate = NodeStream.fromDuplex({
146 evaluate: () => Zlib.createGzip(),
147 onError: () => "error" as const
148 })
149 const inflate = NodeStream.fromDuplex<never, Uint8Array, Uint8Array, "error">({
150 evaluate: () => Zlib.createUnzip(),
151 onError: () => "error" as const
152 })
153 const channel = Channel.pipeToOrFail(deflate, inflate)
154 const result = yield* stream.pipe(
155 Stream.pipeThroughChannelOrFail(channel),
156 Stream.decodeText(),
157 Stream.mkString
158 )
159 assert.strictEqual(result, text)
160 }))
161
162 it.effect("toReadable roundtrip", () =>
163 Effect.gen(function*() {
164 const stream = Stream.range(0, 10000).pipe(
165 Stream.map((n) => String(n))
166 )
167 const readable = yield* NodeStream.toReadable(stream)
168 const outStream = NodeStream.fromReadable({
169 evaluate: () => readable,
170 onError: () => "error" as const
171 })
172 const items = yield* outStream.pipe(
173 Stream.decodeText(),
174 Stream.runCollect
175 )
176 assert.strictEqual(items.join(""), Array.range(0, 10000).join(""))
177 }))
178
179 it.effect("toReadable with error", () =>
180 Effect.gen(function*() {
181 const stream = Stream.fail("error")
182 const readable = yield* NodeStream.toReadable(stream)
183 const outStream = NodeStream.fromReadable({
184 evaluate: () => readable
185 })
186 const error = yield* outStream.pipe(
187 Stream.runCollect,
188 Effect.tapError((_) => Console.log(_)),
189 Effect.flip
190 )
191 assert.deepEqual(error.cause, "error")
192 }))
193
194 it.effect("toString registers one error listener", () =>
195 Effect.gen(function*() {
196 const stream = new Readable({
197 read() {}
198 })
199 yield* NodeStream.toString(() => stream).pipe(Effect.forkChild)
200 yield* Effect.yieldNow
201 assert.strictEqual(stream.listenerCount("error"), 1)
202 }))
203
204 it.effect("collects strings and array buffers with an Infinity maxBytes limit", () =>
205 Effect.gen(function*() {
206 const chunks = [Buffer.from("hello "), Buffer.from("world")]
207 const text = yield* NodeStream.toString(() => Readable.from(chunks), { maxBytes: Infinity })
208 const buffer = yield* NodeStream.toArrayBuffer(() => Readable.from(chunks), { maxBytes: Infinity })
209
210 assert.strictEqual(text, "hello world")
211 assert.deepStrictEqual(new Uint8Array(buffer), new TextEncoder().encode("hello world"))
212 }))
213
214 it.effect("toString enforces a zero maxBytes limit", () =>
215 Effect.gen(function*() {
216 let emitted = false
217 const stream = new Readable({
218 read() {
219 if (emitted) return
220 emitted = true
221 this.push("a")
222 }
223 })
224 const error = yield* NodeStream.toString(() => stream, {
225 maxBytes: "0 B",
226 onError: () => "maxBytes exceeded" as const
227 }).pipe(Effect.flip)
228
229 assert.strictEqual(error, "maxBytes exceeded")
230 assert.isTrue(stream.destroyed)
231 }))
232
233 it.effect("toArrayBuffer enforces a zero maxBytes limit", () =>
234 Effect.gen(function*() {
235 let emitted = false
236 const stream = new Readable({
237 read() {
238 if (emitted) return
239 emitted = true
240 this.push(Buffer.from("a"))
241 }
242 })
243 const error = yield* NodeStream.toArrayBuffer(() => stream, {
244 maxBytes: ByteSize.zero,
245 onError: () => "maxBytes exceeded" as const
246 }).pipe(Effect.flip)
247
248 assert.strictEqual(error, "maxBytes exceeded")
249 assert.isTrue(stream.destroyed)
250 }))
251})