import { describe, it } from "@effect/vitest" import { assertExitFailure, assertExitSuccess, assertNone, assertSome, deepStrictEqual, strictEqual } from "@effect/vitest/utils" import { Array, Cause, Effect, Option, Ref, Result, Sink, Stream } from "effect" import { constTrue, pipe } from "effect/Function" describe("Sink", () => { describe("reduceWhile", () => { it.effect("empty", () => Effect.gen(function*() { const result = yield* Stream.empty.pipe( Stream.transduce(Sink.reduceWhile(() => 0, constTrue, (x, y) => x + y)), Stream.runCollect ) deepStrictEqual(result, [0]) })) it.effect("termination in the middle", () => Effect.gen(function*() { const result = yield* Stream.range(1, 9).pipe( Stream.run(Sink.reduceWhile(() => 0, (n) => n <= 5, (x, y) => x + y)) ) strictEqual(result, 6) })) it.effect("immediate termination", () => Effect.gen(function*() { const result = yield* Stream.range(1, 9).pipe( Stream.run(Sink.reduceWhile(() => 0, (n) => n <= -1, (x, y) => x + y)) ) strictEqual(result, 0) })) it.effect("no termination", () => Effect.gen(function*() { const result = yield* Stream.range(1, 9).pipe( Stream.run(Sink.reduceWhile(() => 0, (n) => n <= 500, (x, y) => x + y)) ) strictEqual(result, 45) })) }) describe("reduceWhileEffect", () => { it.effect("short circuits", () => Effect.gen(function*() { const empty: Stream.Stream = Stream.empty const single = Stream.make(1) const double = Stream.make(1, 2) const failed = Stream.fail("Ouch") const run = (stream: Stream.Stream) => Ref.make(Array.empty()).pipe( Effect.flatMap((ref) => stream.pipe( Stream.transduce(Sink.reduceWhileEffect( () => 0, constTrue, (_, y: number) => Effect.as(Ref.update(ref, Array.append(y)), 30) )), Stream.runCollect, Effect.flatMap((exit) => Ref.get(ref).pipe( Effect.map((result) => [exit, result]) ) ) ) ), Effect.exit ) const result1 = yield* run(empty) const result2 = yield* run(single) const result3 = yield* run(double) const result4 = yield* run(failed) assertExitSuccess(result1, [[0], []]) assertExitSuccess(result2, [[30], [1]]) assertExitSuccess(result3, [[30], [1, 2]]) assertExitFailure(result4, Cause.fail("Ouch")) })) }) describe("reduce", () => { it.effect("equivalence with Array.reduce", () => Effect.gen(function*() { const stream = Stream.range(1, 9) const result1 = yield* stream.pipe(Stream.run(Sink.reduce(() => "", (s, n) => s + `${n}`))) const result2 = yield* stream.pipe( Stream.runCollect, Effect.map(Array.reduce("", (s, n) => s + `${n}`)) ) strictEqual(result1, result2) })) }) describe("take", () => { it.effect("respects the given limit", () => Effect.gen(function*() { const stream = Stream.make(1, 2, 3, 4).pipe( Stream.transduce(Sink.take(3)) ) const result = yield* Stream.runCollect(stream) deepStrictEqual( result, [[1, 2, 3], [4]] ) })) it.effect("produces empty trailing chunks", () => Effect.gen(function*() { const stream = Stream.make(1, 2, 3, 4).pipe( Stream.transduce(Sink.take(4)) ) const result = yield* Stream.runCollect(stream) deepStrictEqual( result, [[1, 2, 3, 4], []] ) })) it.effect("produces an empty chunk for empty input", () => Effect.gen(function*() { const stream = Stream.empty.pipe( Stream.transduce(Sink.take(3)) ) const result = yield* Stream.runCollect(stream) deepStrictEqual(result, [[]]) })) }) describe("takeWhile", () => { it.effect("takeWhile", () => Effect.gen(function*() { const result = yield* Stream.make(1, 2, 3, 4).pipe( Stream.run(Sink.takeWhile((n) => n < 3)) ) deepStrictEqual(result, [1, 2]) })) it.effect("takeWhileFilter", () => Effect.gen(function*() { const result = yield* Stream.make(1, 2, 3, 4).pipe( Stream.run(Sink.takeWhileFilter((n) => n < 3 ? Result.succeed(n * 2) : Result.failVoid)) ) deepStrictEqual(result, [2, 4]) })) it.effect("takeWhileFilter consumes the first failing input", () => Effect.gen(function*() { const result = yield* Stream.make(1, 2, 3, 4).pipe( Stream.transduce(Sink.takeWhileFilter((n) => n < 3 ? Result.succeed(n) : Result.failVoid)), Stream.runCollect ) deepStrictEqual(result, [[1, 2], [], []]) })) it.effect("takeWhileEffect", () => Effect.gen(function*() { const result = yield* Stream.make(1, 2, 3, 4).pipe( Stream.run(Sink.takeWhileEffect((n) => Effect.succeed(n < 3))) ) deepStrictEqual(result, [1, 2]) })) it.effect("takeWhileFilterEffect", () => Effect.gen(function*() { const result = yield* Stream.make(1, 2, 3, 4).pipe( Stream.run(Sink.takeWhileFilterEffect((n) => Effect.succeed(n < 3 ? Result.succeed(n + 1) : Result.failVoid))) ) deepStrictEqual(result, [2, 3]) })) it.effect("takeWhileFilterEffect consumes the first failing input", () => Effect.gen(function*() { const result = yield* Stream.make(1, 2, 3, 4).pipe( Stream.transduce( Sink.takeWhileFilterEffect((n) => Effect.succeed(n < 3 ? Result.succeed(n) : Result.failVoid)) ), Stream.runCollect ) deepStrictEqual(result, [[1, 2], [], []]) })) }) describe("flatMap", () => { it.effect("flatMap - empty input", () => Effect.gen(function*() { const sink = pipe(Sink.head(), Sink.flatMap(Sink.succeed)) const result = yield* pipe(Stream.empty, Stream.run(sink)) assertNone(result) })) it.effect("flatMap - non-empty input", () => Effect.gen(function*() { const sink = pipe(Sink.head(), Sink.flatMap(Sink.succeed)) const result = yield* pipe(Stream.make(1, 2, 3), Stream.run(sink)) assertSome(result, 1) })) it.effect("flatMap - with leftovers", () => Effect.gen(function*() { const chunks = Array.make( Array.make(1, 2), Array.make(3, 4, 5), Array.empty(), Array.make(7, 8, 9, 10) ) const sink = pipe( Sink.head(), Sink.flatMap((head) => pipe( Sink.count, Sink.map((count) => [head, count] as const) ) ) ) const [option, count] = yield* pipe(Stream.fromArrays(...chunks), Stream.run(sink)) deepStrictEqual(option, Array.head(Array.flatten(chunks))) strictEqual( count + Option.match(option, { onNone: () => 0, onSome: () => 1 }), chunks.flat().length ) })) }) })