1import { describe, it } from "@effect/vitest"
2import {
3 assertExitFailure,
4 assertExitSuccess,
5 assertNone,
6 assertSome,
7 deepStrictEqual,
8 strictEqual
9} from "@effect/vitest/utils"
10import { Array, Cause, Deferred, Duration, Effect, Fiber, Option, Ref, Result, Sink, Stream } from "effect"
11import { constTrue, pipe } from "effect/Function"
12import { TestClock } from "effect/testing"
13
14describe("Sink", () => {
15 describe("constructors", () => {
16 it.effect("fromWritableStream - aborts instead of closing on upstream failure", () =>
17 Effect.gen(function*() {
18 const error = new Error("upstream failed")
19 let abortReason: unknown = undefined
20 let closes = 0
21 const writable = new WritableStream<number>({
22 close() {
23 closes++
24 },
25 abort(reason) {
26 abortReason = reason
27 }
28 })
29 const exit = yield* Stream.make(1).pipe(
30 Stream.concat(Stream.fail(error)),
31 Stream.run(Sink.fromWritableStream({
32 evaluate: () => writable,
33 onError: (cause) => cause
34 })),
35 Effect.exit
36 )
37
38 assertExitFailure(exit, Cause.fail(error))
39 strictEqual(closes, 0)
40 deepStrictEqual(Cause.squash(abortReason as Cause.Cause<Error>), error)
41 }))
42
43 it.effect("fromWritableStream - aborts instead of closing on interruption", () =>
44 Effect.gen(function*() {
45 const started = yield* Deferred.make<void>()
46 let abortReason: unknown = undefined
47 let closes = 0
48 const writable = new WritableStream<number>({
49 close() {
50 closes++
51 },
52 abort(reason) {
53 abortReason = reason
54 }
55 })
56 const sink = Sink.fromWritableStream({
57 evaluate: () => writable,
58 onError: (cause) => cause
59 })
60 const fiber = yield* Stream.fromEffect(
61 Deferred.succeed(started, void 0).pipe(Effect.andThen(Effect.never))
62 ).pipe(
63 Stream.run(sink),
64 Effect.forkChild
65 )
66
67 yield* Deferred.await(started)
68 yield* Fiber.interrupt(fiber)
69
70 strictEqual(closes, 0)
71 strictEqual(abortReason === undefined, false)
72 }))
73 })
74
75 describe("reduceWhile", () => {
76 it.effect("empty", () =>
77 Effect.gen(function*() {
78 const result = yield* Stream.empty.pipe(
79 Stream.transduce(Sink.reduceWhile<number, number>(() => 0, constTrue, (x, y) => x + y)),
80 Stream.runCollect
81 )
82 deepStrictEqual(result, [0])
83 }))
84
85 it.effect("termination in the middle", () =>
86 Effect.gen(function*() {
87 const result = yield* Stream.range(1, 9).pipe(
88 Stream.run(Sink.reduceWhile<number, number>(() => 0, (n) => n <= 5, (x, y) => x + y))
89 )
90 strictEqual(result, 6)
91 }))
92
93 it.effect("immediate termination", () =>
94 Effect.gen(function*() {
95 const result = yield* Stream.range(1, 9).pipe(
96 Stream.run(Sink.reduceWhile<number, number>(() => 0, (n) => n <= -1, (x, y) => x + y))
97 )
98 strictEqual(result, 0)
99 }))
100
101 it.effect("no termination", () =>
102 Effect.gen(function*() {
103 const result = yield* Stream.range(1, 9).pipe(
104 Stream.run(Sink.reduceWhile<number, number>(() => 0, (n) => n <= 500, (x, y) => x + y))
105 )
106 strictEqual(result, 45)
107 }))
108 })
109
110 describe("reduceWhileArray", () => {
111 it.effect("applies the reducer once per non-empty input array", () =>
112 Effect.gen(function*() {
113 const result = yield* Stream.fromArrays([1, 2, 3]).pipe(
114 Stream.run(Sink.reduceWhileArray(() => 0, constTrue, (count) => count + 1))
115 )
116
117 strictEqual(result, 1, "the reducer must run once for each input array")
118 }))
119 })
120
121 describe("reduceWhileEffect", () => {
122 it.effect("short circuits", () =>
123 Effect.gen(function*() {
124 const empty: Stream.Stream<number> = Stream.empty
125 const single = Stream.make(1)
126 const double = Stream.make(1, 2)
127 const failed = Stream.fail("Ouch")
128 const run = <E>(stream: Stream.Stream<number, E>) =>
129 Ref.make(Array.empty<number>()).pipe(
130 Effect.flatMap((ref) =>
131 stream.pipe(
132 Stream.transduce(Sink.reduceWhileEffect(
133 () => 0,
134 constTrue,
135 (_, y: number) => Effect.as(Ref.update(ref, Array.append(y)), 30)
136 )),
137 Stream.runCollect,
138 Effect.flatMap((exit) =>
139 Ref.get(ref).pipe(
140 Effect.map((result) => [exit, result])
141 )
142 )
143 )
144 ),
145 Effect.exit
146 )
147 const result1 = yield* run(empty)
148 const result2 = yield* run(single)
149 const result3 = yield* run(double)
150 const result4 = yield* run(failed)
151 assertExitSuccess(result1, [[0], []])
152 assertExitSuccess(result2, [[30], [1]])
153 assertExitSuccess(result3, [[30], [1, 2]])
154 assertExitFailure(result4, Cause.fail("Ouch"))
155 }))
156 })
157
158 describe("reduce", () => {
159 it.effect("equivalence with Array.reduce", () =>
160 Effect.gen(function*() {
161 const stream = Stream.range(1, 9)
162 const result1 = yield* stream.pipe(Stream.run(Sink.reduce(() => "", (s, n) => s + `${n}`)))
163 const result2 = yield* stream.pipe(
164 Stream.runCollect,
165 Effect.map(Array.reduce("", (s, n) => s + `${n}`))
166 )
167 strictEqual(result1, result2)
168 }))
169 })
170
171 describe("foldUntil", () => {
172 it.effect("normalizes the maximum element count", () =>
173 Effect.gen(function*() {
174 const results = yield* Effect.forEach([Number.NaN, -1, 0, 0.5, 1.9, 2.9], (max) =>
175 Stream.make(1, 2, 3).pipe(
176 Stream.run(Sink.foldUntil(() => 0, max, (count) => Effect.succeed(count + 1)))
177 ))
178 deepStrictEqual(results, [0, 0, 0, 0, 1, 2])
179 }))
180
181 it.effect("zero normalized counts do not evaluate the upstream", () =>
182 Effect.gen(function*() {
183 let initializations = 0
184 const sink = Sink.foldUntil(
185 () => {
186 initializations++
187 return 42
188 },
189 Number.NaN,
190 (state: number) => Effect.succeed(state + 1)
191 )
192
193 strictEqual(initializations, 0)
194 const result = yield* Stream.fromEffect(Effect.die("upstream evaluated")).pipe(Stream.run(sink))
195 strictEqual(result, 42)
196 strictEqual(initializations, 1)
197 }))
198 })
199
200 describe("take", () => {
201 it.effect("normalizes the element count", () =>
202 Effect.gen(function*() {
203 const results = yield* Effect.all([
204 Stream.make(1, 2, 3).pipe(Stream.run(Sink.take(Number.NaN))),
205 Stream.make(1, 2, 3).pipe(Stream.run(Sink.take(-1))),
206 Stream.make(1, 2, 3).pipe(Stream.run(Sink.take(1.9)))
207 ])
208 deepStrictEqual(results, [[], [], [1]])
209 }))
210
211 it.effect("respects the given limit", () =>
212 Effect.gen(function*() {
213 const stream = Stream.make(1, 2, 3, 4).pipe(
214 Stream.transduce(Sink.take<number>(3))
215 )
216 const result = yield* Stream.runCollect(stream)
217 deepStrictEqual(
218 result,
219 [[1, 2, 3], [4]]
220 )
221 }))
222
223 it.effect("produces empty trailing chunks", () =>
224 Effect.gen(function*() {
225 const stream = Stream.make(1, 2, 3, 4).pipe(
226 Stream.transduce(Sink.take<number>(4))
227 )
228 const result = yield* Stream.runCollect(stream)
229 deepStrictEqual(
230 result,
231 [[1, 2, 3, 4], []]
232 )
233 }))
234
235 it.effect("produces an empty chunk for empty input", () =>
236 Effect.gen(function*() {
237 const stream = Stream.empty.pipe(
238 Stream.transduce(Sink.take<number>(3))
239 )
240 const result = yield* Stream.runCollect(stream)
241 deepStrictEqual(result, [[]])
242 }))
243 })
244
245 describe("takeWhile", () => {
246 it.effect("takeWhile", () =>
247 Effect.gen(function*() {
248 const result = yield* Stream.make(1, 2, 3, 4).pipe(
249 Stream.run(Sink.takeWhile((n) => n < 3))
250 )
251 deepStrictEqual(result, [1, 2])
252 }))
253
254 it.effect("takeWhileFilter", () =>
255 Effect.gen(function*() {
256 const result = yield* Stream.make(1, 2, 3, 4).pipe(
257 Stream.run(Sink.takeWhileFilter((n) => n < 3 ? Result.succeed(n * 2) : Result.failVoid))
258 )
259 deepStrictEqual(result, [2, 4])
260 }))
261
262 it.effect("takeWhileFilter consumes the first failing input", () =>
263 Effect.gen(function*() {
264 const result = yield* Stream.make(1, 2, 3, 4).pipe(
265 Stream.transduce(Sink.takeWhileFilter((n) => n < 3 ? Result.succeed(n) : Result.failVoid)),
266 Stream.runCollect
267 )
268 deepStrictEqual(result, [[1, 2], [], []])
269 }))
270
271 it.effect("takeWhileEffect", () =>
272 Effect.gen(function*() {
273 const result = yield* Stream.make(1, 2, 3, 4).pipe(
274 Stream.run(Sink.takeWhileEffect((n) => Effect.succeed(n < 3)))
275 )
276 deepStrictEqual(result, [1, 2])
277 }))
278
279 it.effect("takeWhileFilterEffect", () =>
280 Effect.gen(function*() {
281 const result = yield* Stream.make(1, 2, 3, 4).pipe(
282 Stream.run(Sink.takeWhileFilterEffect((n) => Effect.succeed(n < 3 ? Result.succeed(n + 1) : Result.failVoid)))
283 )
284 deepStrictEqual(result, [2, 3])
285 }))
286
287 it.effect("takeWhileFilterEffect consumes the first failing input", () =>
288 Effect.gen(function*() {
289 const result = yield* Stream.make(1, 2, 3, 4).pipe(
290 Stream.transduce(
291 Sink.takeWhileFilterEffect((n) => Effect.succeed(n < 3 ? Result.succeed(n) : Result.failVoid))
292 ),
293 Stream.runCollect
294 )
295 deepStrictEqual(result, [[1, 2], [], []])
296 }))
297 })
298
299 describe("withDuration", () => {
300 it.effect("uses monotonic time when wall time moves backward", () =>
301 Effect.gen(function*() {
302 yield* TestClock.setTime(1_000)
303 const [, duration] = yield* Stream.empty.pipe(
304 Stream.run(
305 Sink.fromEffect(Effect.gen(function*() {
306 yield* TestClock.adjust("100 millis")
307 yield* TestClock.setTime(0)
308 })).pipe(Sink.withDuration)
309 )
310 )
311
312 strictEqual(Duration.toMillis(duration), 100)
313 }))
314 })
315
316 describe("flatMap", () => {
317 it.effect("preserves leftovers when the next sink consumes no input", () =>
318 Effect.gen(function*() {
319 const sink = Sink.take<number>(1).pipe(
320 Sink.flatMap(() => Sink.take<number>(0)),
321 Sink.flatMap(() => Sink.collect<number>())
322 )
323 const result = yield* Stream.fromArrays([1, 2, 3], [4, 5]).pipe(Stream.run(sink))
324 deepStrictEqual(result, [2, 3, 4, 5], "pending leftovers should be preserved")
325 }))
326
327 it.effect("flatMap - empty input", () =>
328 Effect.gen(function*() {
329 const sink = pipe(Sink.head<number>(), Sink.flatMap(Sink.succeed))
330 const result = yield* pipe(Stream.empty, Stream.run(sink))
331 assertNone(result)
332 }))
333
334 it.effect("flatMap - non-empty input", () =>
335 Effect.gen(function*() {
336 const sink = pipe(Sink.head<number>(), Sink.flatMap(Sink.succeed))
337 const result = yield* pipe(Stream.make(1, 2, 3), Stream.run(sink))
338 assertSome(result, 1)
339 }))
340
341 it.effect("flatMap - with leftovers", () =>
342 Effect.gen(function*() {
343 const chunks = Array.make(
344 Array.make(1, 2),
345 Array.make(3, 4, 5),
346 Array.empty<number>(),
347 Array.make(7, 8, 9, 10)
348 )
349 const sink = pipe(
350 Sink.head<number>(),
351 Sink.flatMap((head) =>
352 pipe(
353 Sink.count,
354 Sink.map((count) => [head, count] as const)
355 )
356 )
357 )
358 const [option, count] = yield* pipe(Stream.fromArrays(...chunks), Stream.run(sink))
359 deepStrictEqual(option, Array.head(Array.flatten(chunks)))
360 strictEqual(
361 count + Option.match(option, {
362 onNone: () => 0,
363 onSome: () => 1
364 }),
365 chunks.flat().length
366 )
367 }))
368 })
369})