1import * as NodeSink from "@effect/platform-node-shared/NodeSink"
2import * as NodeStream from "@effect/platform-node-shared/NodeStream"
3import { assert, describe, it } from "@effect/vitest"
4import { Deferred, Effect, Exit, Fiber, Option } from "effect"
5import * as Data from "effect/Data"
6import * as Latch from "effect/Latch"
7import * as Queue from "effect/Queue"
8import * as Stream from "effect/Stream"
9import { TestClock } from "effect/testing"
10import { createReadStream } from "fs"
11import { join } from "path"
12import { Writable } from "stream"
13import * as Tar from "tar"
14
15const TEST_TARBALL = join(__dirname, "fixtures", "helloworld.tar.gz")
16
17describe("Sink", () => {
18 it.effect("does not resume a selected drain callback after interruption", () =>
19 Effect.gen(function*() {
20 const items: Array<string> = []
21 const entered = yield* Deferred.make<() => void>()
22 const drained = yield* Deferred.make<void>()
23 const controller = new AbortController()
24 const writable = new Writable({
25 highWaterMark: 1,
26 write(chunk, _encoding, callback) {
27 items.push(chunk.toString())
28 if (chunk.toString() === "a") {
29 queueMicrotask(() => Deferred.doneUnsafe(entered, Effect.succeed(callback)))
30 } else {
31 callback()
32 }
33 }
34 })
35 yield* Effect.addFinalizer(() =>
36 Effect.sync(() => {
37 controller.abort()
38 writable.destroy()
39 })
40 )
41 writable.once("drain", () => controller.abort())
42 Effect.runCallback(
43 Stream.make("a", "b").pipe(
44 Stream.run(NodeSink.fromWritable({ evaluate: () => writable, onError: (error) => error }))
45 ),
46 { signal: controller.signal, onExit: () => {} }
47 )
48 const release = yield* Deferred.await(entered)
49 assert.deepEqual(items, ["a"])
50 writable.once("drain", () => {
51 queueMicrotask(() => Deferred.doneUnsafe(drained, Effect.void))
52 })
53 yield* Effect.sync(release)
54 yield* Deferred.await(drained)
55 assert.deepEqual(items, ["a"])
56 }))
57
58 it.effect.each(["normal_final", "write_error", "final_error"] as const)(
59 "writable lifecycle: %s",
60 (mode) =>
61 Effect.gen(function*() {
62 const error = new Error(mode)
63 const mapped = { _tag: "WritableError", cause: error }
64 const observed: Array<unknown> = []
65 const mappedErrors: Array<unknown> = []
66 const closed = yield* Deferred.make<void>()
67 const items: Array<string> = []
68 let finalized = false
69 const writable = new Writable({
70 write(chunk, _encoding, callback) {
71 items.push(chunk.toString())
72 callback(mode === "write_error" ? error : null)
73 },
74 final(callback) {
75 finalized = true
76 callback(mode === "final_error" ? error : null)
77 }
78 })
79 // Observe Node errors without handling them on behalf of the sink.
80 writable.on("error", (error) => observed.push(error))
81 writable.once("close", () => Deferred.doneUnsafe(closed, Effect.void))
82 yield* Effect.addFinalizer(() => Effect.sync(() => writable.destroy()))
83 const fiber = yield* Stream.make("a", "b").pipe(
84 Stream.run(NodeSink.fromWritable({
85 evaluate: () => writable,
86 onError: (error) => {
87 mappedErrors.push(error)
88 return mapped
89 }
90 })),
91 Effect.exit,
92 Effect.timeoutOption("1 second"),
93 Effect.forkScoped
94 )
95 yield* Deferred.await(closed)
96 // The writable has settled; advance only the virtual non-completion guard.
97 yield* TestClock.adjust("1 second")
98 assert.deepEqual(observed, mode === "normal_final" ? [] : [error])
99 assert.strictEqual(finalized, mode !== "write_error")
100 assert.deepEqual(items, mode === "write_error" ? ["a"] : ["a", "b"])
101 assert.deepEqual({ result: yield* Fiber.join(fiber), mappedErrors }, {
102 result: Option.some(mode === "normal_final" ? Exit.void : Exit.fail(mapped)),
103 mappedErrors: mode === "normal_final" ? [] : [error]
104 })
105 })
106 )
107
108 it.effect.each(["error", "interrupt"] as const)(
109 "cleans up finalization listeners on %s",
110 (mode) =>
111 Effect.gen(function*() {
112 const finalizing = yield* Deferred.make<(error?: Error | null) => void>()
113 const error = new Error("finalization failed")
114 const writable = new Writable({
115 write(_chunk, _encoding, callback) {
116 callback()
117 },
118 final(callback) {
119 queueMicrotask(() => Deferred.doneUnsafe(finalizing, Effect.succeed(callback)))
120 }
121 })
122 const observed: Array<unknown> = []
123 writable.on("error", (error) => observed.push(error))
124 yield* Effect.addFinalizer(() => Effect.sync(() => writable.destroy()))
125 const fiber = yield* Stream.make("a").pipe(
126 Stream.run(NodeSink.fromWritable({ evaluate: () => writable, onError: (error) => error })),
127 Effect.forkScoped
128 )
129 const release = yield* Deferred.await(finalizing)
130 assert.strictEqual(writable.listenerCount("finish"), 1)
131 if (mode === "error") {
132 yield* Effect.sync(() => release(error))
133 assert.deepEqual(yield* Fiber.await(fiber), Exit.fail(error))
134 } else {
135 yield* Fiber.interrupt(fiber)
136 }
137 assert.strictEqual(writable.listenerCount("finish"), 0)
138 assert.strictEqual(writable.listenerCount("error"), 1)
139 assert.deepEqual(observed, mode === "error" ? [error] : [])
140 if (mode === "interrupt") {
141 yield* Effect.sync(() => release())
142 }
143 })
144 )
145
146 it.effect("should write to a stream", () =>
147 Effect.gen(function*() {
148 const items: Array<string> = []
149 const destroyLatch = yield* Latch.make()
150 yield* Stream.make("a", "b", "c").pipe(
151 Stream.run(NodeSink.fromWritable({
152 evaluate: () =>
153 new Writable({
154 construct(callback) {
155 callback()
156 },
157 write(chunk, _encoding, callback) {
158 items.push(chunk.toString())
159 callback()
160 },
161 destroy(_error, callback) {
162 destroyLatch.openUnsafe()
163 callback(null)
164 }
165 }),
166 onError: () => "error"
167 }))
168 )
169 assert.deepEqual(items, ["a", "b", "c"])
170 yield* destroyLatch.await
171 }))
172
173 it.effect("write error", () =>
174 Effect.gen(function*() {
175 const items: Array<string> = []
176 const sink = NodeSink.fromWritable({
177 evaluate: () =>
178 new Writable({
179 construct(callback) {
180 callback()
181 },
182 write(chunk, _encoding, callback) {
183 items.push(chunk.toString())
184 callback()
185 },
186 destroy(_error, callback) {
187 callback(null)
188 }
189 }),
190 onError: () => "error"
191 })
192 const result = yield* Stream.fail("a").pipe(Stream.run(sink), Effect.flip)
193 assert.deepEqual(items, [])
194 assert.strictEqual(result, "a")
195 }))
196
197 it.live("endOnClose false", () =>
198 Effect.gen(function*() {
199 const items: Array<string> = []
200 let destroyed = false
201 const sink = NodeSink.fromWritable({
202 evaluate: () =>
203 new Writable({
204 construct(callback) {
205 callback()
206 },
207 write(chunk, _encoding, callback) {
208 items.push(chunk.toString())
209 callback()
210 },
211 destroy(_error, callback) {
212 destroyed = true
213 callback(null)
214 }
215 }),
216 onError: () => "error",
217 endOnDone: false
218 })
219 yield* Stream.run(Stream.make("a", "b", "c"), sink)
220 yield* Effect.sleep(10)
221 assert.deepEqual(items, ["a", "b", "c"])
222 assert.strictEqual(destroyed, false)
223 }))
224
225 it.effect("should handle non-compliant node streams", () =>
226 Effect.gen(function*() {
227 const stream = NodeStream.fromReadable<Uint8Array, "error">({
228 evaluate: () => createReadStream(TEST_TARBALL),
229 onError: () => "error"
230 })
231 const items = yield* entries(stream).pipe(
232 Stream.flatMap((entry) =>
233 NodeStream.fromReadable({
234 evaluate: () => (entry as any),
235 onError: (error) => new TarError({ error })
236 }).pipe(
237 Stream.map((content) => ({
238 path: entry.path,
239 content: Buffer.from(content).toString("utf-8")
240 }))
241 )
242 ),
243 Stream.runCollect
244 )
245 assert.deepEqual(items, [
246 { path: "./tar/world.txt", content: "world\n" },
247 { path: "./tar/hello.txt", content: "hello\n" }
248 ])
249 }))
250})
251
252class TarError extends Data.TaggedError("TarError")<{
253 readonly error: unknown
254}> {}
255
256const entries = <R, E>(
257 input: Stream.Stream<Uint8Array, E, R>
258): Stream.Stream<Tar.ReadEntry, TarError | E, R> =>
259 Effect.gen(function*() {
260 const parser = new Tar.Parser()
261
262 yield* input.pipe(
263 Stream.run(
264 NodeSink.fromWritable({
265 evaluate: () => parser,
266 onError: (error) => new TarError({ error })
267 })
268 ),
269 Effect.forkScoped
270 )
271
272 return Stream.callback<Tar.ReadEntry, TarError>((queue) =>
273 Effect.sync(() => {
274 parser.on("entry", (entry) => {
275 Queue.offerUnsafe(queue, entry)
276 })
277 parser.on("close", () => {
278 Queue.endUnsafe(queue)
279 })
280 })
281 )
282 }).pipe(Stream.unwrap)