diff --git a/README.md b/README.md index 21d2a6c..384c17b 100644 --- a/README.md +++ b/README.md @@ -31,6 +31,7 @@ array.filter(predicate).map(transform) Pipelean gives you: - `series` for sequential work over arrays **and async iterables** — live `onProgress`, `pause` rate limits, `take`, first-class error strategies +- `stopWhen` to cancel or predicate-stop any source — composes with every consumer, no option changes - `scan` / `reduce` for stateful accumulation across many items - `flow` for stateful accumulation across one input — each operation enriches the same state - `pipe` for vertical composition diff --git a/docs/architecture.md b/docs/architecture.md index a9e198d..1cb0b9b 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -21,7 +21,7 @@ Our Approach: Eager Execution. When you run series or scan, the work happens immediately and you get a structured report `{ results, errors, failure }` back. No surprises. -This is a focus, not a rejection of lazy iterators. Pipelean *consumes* async iterables (arrays, generators, anything `for await` can consume) through `series`/`scan`/`reduce`/`filter`. When you need to *build* a lazy producer — for example an async generator that yields pages — pipelean steps aside: write a standard JavaScript generator and let it yield. The `no-loop-without-yield` lint rule encodes exactly this boundary: the only loops pipelean allows are the ones inside a generator that yields. +This is a focus, not a rejection of lazy iterators. Pipelean *consumes* async iterables (arrays, generators, anything `for await` can consume) through `series`/`scan`/`reduce`/`filter`. When you need to *build* a lazy producer — for example an async generator that yields pages — pipelean steps aside: write a standard JavaScript generator and let it yield. The `no-loop-without-yield` lint rule encodes exactly this boundary: the only loops pipelean allows are the ones inside a generator that yields. `stopWhen` follows the same rule: it is a tiny yielding adapter (a producer, not an iterator), shipped because every consumer needed the same five lines of stop logic. ## Terminology diff --git a/docs/examples.md b/docs/examples.md index 6bf1f82..ed46d65 100644 --- a/docs/examples.md +++ b/docs/examples.md @@ -73,6 +73,30 @@ const {value: totalDuration} = await reduce( // totalDuration = 22 — no .at(-1), no fallback ``` +#### Example: stopWhen (Cancellable Pipeline) + +Scenario: You are enriching albums against the MusicBrainz API. The job takes minutes, users hit Cancel, and every request must respect a rate limit. Cancellation, pacing, error strategy and progress each live in exactly one place: + +```js +import { series, stopWhen } from 'pipelean' + +const enrichLibrary = async (albums, { shouldStop = () => false, onProgress }) => { + await series(enrichOne, { + total: albums.length, + pause: ENRICH_DELAY_MS, // rate limit lives in series + pauseOnErrors: true, + onProgress: onItem, // progress lives in series + })(stopWhen(albums, shouldStop)) // cancellation lives in the source +} + +const { results } = await enrichLibrary(albums, { + shouldStop: () => cancelRequested, + onProgress: ({ index, total }) => updateBar(index + 1, total), +}) +``` + +Before `stopWhen`, this needed an async generator with inline `shouldStop()` checks and the rate-limit delay tangled into its `finally` block — which is how double-yield bugs are born. Now each concern has one home. + #### Example: tryCatch as an App-Layer Primitive Scenario: You want a reusable "Error Boundary" for your application that automatically logs errors to a monitoring service (like Sentry) and pushes a notification to your UI state (e.g., a Svelte store), ensuring the app never crashes silently. diff --git a/docs/guide.md b/docs/guide.md index 8584db3..c6a87a0 100644 --- a/docs/guide.md +++ b/docs/guide.md @@ -16,6 +16,7 @@ Pipelean provides core tools grouped by **data flow direction** (horizontal vs v 6. pipe (Vertical / Composition) 7. flow (Vertical / Stateful accumulation — one input, many enrichments, final accumulated value) 8. assign (Utility for creating conditional property assignments for flow) +9. stopWhen (Source adapter — predicate-based early exit for any iterable) > **Sync variants** — The iteration functions above also have synchronous > counterparts: `seriesSync`, `filterSync`, `findSync`, `scanSync`, and @@ -95,6 +96,11 @@ Errors thrown by the **iteration itself** (e.g. an async generator dying mid-str - `total` for progress math; omitted when the size is unknown. - `scan` and `reduce` do **not** have `onProgress`, `pause`, or `take`. +* **Source adapters** + - `stopWhen(items, predicate)` stops pulling as soon as the predicate is truthy — checked *before* the item is offered downstream. + - Composes with every consumer (`series`, `scan`, `reduce`, `filter`, raw `for await`) with no option changes: `series(fn, opts)(stopWhen(items, shouldStop))`. + - Cancel is a clean shorter run (`failure: false`), never a source error. + * **Order Guarantee** - Because execution is sequential, output order strictly matches input order (no race conditions). diff --git a/docs/migration.md b/docs/migration.md index 53db301..2945c1f 100644 --- a/docs/migration.md +++ b/docs/migration.md @@ -2,6 +2,28 @@ How to replace common imperative and error-prone patterns with pipelean equivalents. +## 0.9.2: `stopWhen` source adapter + +`stopWhen(items, predicate)` stops pulling from any iterable once the predicate is truthy, checked before the item is offered downstream. Additive — no migration needed, but delete your hand-rolled stop logic when you see it: + +```js +// Before: a generator with inline checks (and cleanup tangled into finally) +async function * events () { + for (const album of albums) { + if (shouldStop()) + return + yield await enrichOne(album) + } +} + +// After: cancellation lives in the source, work lives in the operation +await series(enrichOne, {pause: DELAY})(stopWhen(albums, shouldStop)) +``` + +The predicate receives `(item, index)` and may close over cancel flags or counters (`() => count >= limit`). Stopping is a clean completion (`failure: false`), not an error; the abandoned source's cleanup still runs. See [stopWhen](reference.md#stopwhen). + +Also in 0.9.2: iterable objects (generators, custom adapters) passed to `filter` / `filterSync` / `findSync` are now correctly treated as sources instead of `where()` patterns — `filter(pages(), pred)` used to silently misbehave. + ## 0.8.4: Node engine requirement Pipelean now requires Node `>22.7` (package.json `engines`). diff --git a/docs/patterns.md b/docs/patterns.md index db011c3..55972b4 100644 --- a/docs/patterns.md +++ b/docs/patterns.md @@ -156,3 +156,33 @@ for (const rawAlbum of rawAlbums) { ``` Operations are reusable building blocks. The same `processAlbum` can be called from a CLI import script, a REST endpoint, or a background job — and the error shapes are normalized (`operation` is the function's `name` or `operation-${index}`), so error reports are consistent across entry points. + +### Pattern 9: Cancel / limit a run with `stopWhen()` + +Use `stopWhen()` when a run should stop pulling items once a condition is met: a user-requested cancel, a batch limit, or a content-based stop. The predicate is checked **before** each item is offered downstream, so the triggering item is never processed. It composes with every pipelean consumer — wrap the source, keep your options untouched: + +```javascript +import { series, stopWhen } from 'pipelean' + +const { results } = await series(enrichOne, { + total: albums.length, // forwarded from the array — optional with stopWhen + pause: ENRICH_DELAY_MS, + pauseOnErrors: true, + onProgress: onItem, +})(stopWhen(albums, shouldStop)) +``` + +The predicate receives `(item, index)` and may close over external state instead of looking at the item at all — that covers cancel flags and limits with one combinator: + +```javascript +stopWhen(albums, shouldStop) // cancel flag +stopWhen(folders, () => count >= limit) // batch limit +stopWhen(pages, page => page.isLast) // content-based stop +``` + +Stopping is a clean completion for the consumer: `failure` stays `false`, `sourceErrors` stays empty — cancel is a shorter run, not an error. Once stopped, the underlying source is abandoned mid-stream and its cleanup (`finally`, `iterator.return()`) still runs. + +Two things `stopWhen` does **not** do: + +- it does not abort work already started on an item — intra-item cancellation stays in your operation (e.g. `AbortController`); +- it is not a `series` option on purpose: wrapping the source also serves raw `for await` loops and `reduce`/`scan` pipelines, where a consumer option would not reach. diff --git a/docs/reference.md b/docs/reference.md index a0d3a64..3b82395 100644 --- a/docs/reference.md +++ b/docs/reference.md @@ -30,6 +30,7 @@ - [scan](#scan) - Stateful Sequential Transformation - [reduce](#reduce) - Pure Reduction - [filter](#filter) - Stateless Selection +- [stopWhen](#stopwhen) - Source Adapter for Predicate-Based Early Exit (alias of sorts: the `while` you don't need) ### Composition @@ -453,7 +454,7 @@ const {value: totalDuration} = await reduce( **Key Characteristics**: - The predicate's return value is never placed into `results` — only truthiness is checked, and the original `item` is what gets kept or dropped. -- Pattern objects are supported: `filter(users, {active: true})` works via `where()`. +- Pattern objects are supported: `filter(users, {active: true})` works via `where()`. Anything implementing the iteration protocols (arrays, generators, source adapters like `stopWhen`) is always treated as a source, never as a pattern. **Usage Example**: ```javascript @@ -468,6 +469,42 @@ const adults = await filter( --- +### stopWhen + +**Purpose**: Source adapter that stops pulling from an iterable as soon as a predicate is truthy. The standard way to cancel or predicate-stop any source consumed by `series`, `scan`, `reduce`, `filter`, or a raw `for await`. + +**Type**: `(items, predicate?) => asyncIterable` +Sync: `(items, predicate?) => iterable` + +**Parameters**: +- `items`: An array or (async) iterable — generators and paging sources included +- `predicate(item, index)`: Checked **before** each item is yielded. Truthy → stop. Defaults to `() => false` (never stops). Sync only. + +**Key Characteristics**: +- **Check before yield**: the triggering item is pulled from the source but never offered downstream — "cancel before work" +- **Pull-lazy**: once stopped, the underlying source is abandoned mid-stream; native cleanup (`iterator.return()`, generator `finally`) still runs +- **Clean completion, not an error**: consumers see a finished source — `failure: false`, empty `sourceErrors`. Cancel is a shorter run, not a source death +- **Composable**: zero changes in `series` / `scan` / `reduce` / `filter`; they just see an iterable that ends early. Works with raw `for await` too +- **Length forwarding**: when the source has a numeric `length` (arrays), it is forwarded on the wrapper so `series` keeps progress totals without an explicit `total`. Generators have no cheap size — `total` stays omitted +- **Predicate throws are source errors**: reported through `onSourceError` / `sourceErrors`, never `onError` +- This is **not** intra-item abort: work already started on an item is not interrupted (use `AbortController` in your operation for that) + +**Usage Example**: +```javascript +import { series } from 'pipelean' + +const {results} = await series(enrichOne, { + total: albums.length, + pause: ENRICH_DELAY_MS, + onProgress: onItem, +})(stopWhen(albums, shouldStop)) + +// limit-style: close over counters instead of the item +stopWhen(folders, () => count >= limit) +``` + +--- + ## Composition ### pipe @@ -765,6 +802,7 @@ want Pipelean's structured error collection. - `pipeSync` composes synchronous functions left-to-right - `flowSync` returns `{value, errors, failure}` directly and runs a state-enrichment pipeline synchronously - `tryCatchSync` wraps a synchronous function with lifecycle hooks +- `stopWhenSync` is the sync source adapter — same contract as `stopWhen`, sync iterables only The sync variants handle **source errors** with the same semantics as their async twins: `seriesSync`, `filterSync`, `scanSync`, and `reduceSync` accept `onSourceError({error, index})` and report errors thrown by the iteration itself in the `sourceErrors` field. See [Source errors](#source-errors) under `series`. diff --git a/package.json b/package.json index 79989d7..fffeeda 100644 --- a/package.json +++ b/package.json @@ -1,6 +1,6 @@ { "name": "pipelean", - "version": "0.9.1", + "version": "0.9.2", "description": "A pragmatic library for sequential async operations with first-class error handling.", "type": "module", "license": "MIT", diff --git a/skills/core/SKILL.md b/skills/core/SKILL.md index 3e6e840..3a76b09 100644 --- a/skills/core/SKILL.md +++ b/skills/core/SKILL.md @@ -67,7 +67,8 @@ const {results, errors, sourceErrors, failure} = await series( 8. **`tryCatch(fn, {onStart, onSuccess, onError, onFinally})`**: Single-function lifecycle. Returns `null` on error. 9. **`where(pattern)`**: Strict-equality object predicate. Used with `filter` / `findSync`. 10. **`assign(property, parse)`**: `flow` step. Sets `{[property]: value}` unless `parse(state)` is `undefined` (returns `{}`). -11. **`*Sync`**: `seriesSync`, `filterSync`, `findSync`, `scanSync`, `reduceSync`, `pipeSync`, `flowSync`, `tryCatchSync`. Same strategies and shapes, no Promises. No `pause` (needs async delay). No async iterables. `findSync` is sync-only early-exit: `{result, errors, failure}`. +11. **`stopWhen(items, predicate?)`** / **`stopWhenSync`**: Source adapter — stops pulling once the predicate `(item, index)` is truthy, checked before the item is yielded. Cancel flags, limits (`() => count >= limit`), content stops. Clean completion (`failure: false`, no sourceErrors), source cleanup still runs, predicate throws are source errors. Composes with every consumer: `series(fn, opts)(stopWhen(items, shouldStop))`. Not intra-item abort. +12. **`*Sync`**: `seriesSync`, `filterSync`, `findSync`, `scanSync`, `reduceSync`, `pipeSync`, `flowSync`, `tryCatchSync`. Same strategies and shapes, no Promises. No `pause` (needs async delay). No async iterables. `findSync` is sync-only early-exit: `{result, errors, failure}`. ## Error strategies diff --git a/src/functional-sync.js b/src/functional-sync.js index 68393cd..b9b5bff 100644 --- a/src/functional-sync.js +++ b/src/functional-sync.js @@ -1,6 +1,6 @@ /* eslint-disable max-lines */ /* eslint-disable max-lines-per-function */ -import {getPlannedTotal, withTotal} from './shared.js' +import {getPlannedTotal, isPatternObject, withTotal} from './shared.js' import { collect, failFast, normalizeOperationError, where, } from './functional.js' @@ -138,11 +138,8 @@ export const seriesSync = (...args) => { } export const filterSync = (...args) => { - const isPattern = x => x !== null && - typeof x === 'object' && - !Array.isArray(x) - const toPredicate = x => isPattern(x) ? where(x) : x - const immediate = typeof args[0] !== 'function' && !isPattern(args[0]) + const toPredicate = x => isPatternObject(x) ? where(x) : x + const immediate = typeof args[0] !== 'function' && !isPatternObject(args[0]) const [items, rawPredicate, opts] = immediate ? args : [null, args[0], args[1]] @@ -158,11 +155,8 @@ export const filterSync = (...args) => { } export const findSync = (...args) => { - const isPattern = x => x !== null && - typeof x === 'object' && - !Array.isArray(x) - const toPredicate = x => isPattern(x) ? where(x) : x - const immediate = typeof args[0] !== 'function' && !isPattern(args[0]) + const toPredicate = x => isPatternObject(x) ? where(x) : x + const immediate = typeof args[0] !== 'function' && !isPatternObject(args[0]) const [items, rawPredicate, opts = {}] = immediate ? args : [null, args[0], args[1]] diff --git a/src/functional.js b/src/functional.js index 3198926..de361ee 100644 --- a/src/functional.js +++ b/src/functional.js @@ -1,6 +1,6 @@ /* eslint-disable max-lines */ /* eslint-disable max-lines-per-function */ -import {getPlannedTotal, withTotal} from './shared.js' +import {getPlannedTotal, isPatternObject, withTotal} from './shared.js' export const failFast = Object.freeze({name: 'failFast'}) export const collect = Object.freeze({name: 'collect'}) @@ -189,11 +189,8 @@ export const series = (...args) => { } export const filter = (...args) => { - const isPattern = x => x !== null && - typeof x === 'object' && - !Array.isArray(x) - const toPredicate = x => isPattern(x) ? where(x) : x - const immediate = typeof args[0] !== 'function' && !isPattern(args[0]) + const toPredicate = x => isPatternObject(x) ? where(x) : x + const immediate = typeof args[0] !== 'function' && !isPatternObject(args[0]) const [items, rawPredicate, opts] = immediate ? args : [null, args[0], args[1]] diff --git a/src/index.js b/src/index.js index 0db5ef6..c476ffa 100644 --- a/src/index.js +++ b/src/index.js @@ -1,2 +1,3 @@ export * from './functional.js' export * from './functional-sync.js' +export * from './iterables.js' diff --git a/src/iterables.js b/src/iterables.js new file mode 100644 index 0000000..6f16d78 --- /dev/null +++ b/src/iterables.js @@ -0,0 +1,36 @@ +// Source adapters: lazy iterable producers, consumed by series/reduce/scan, +// filter, findSync, or any raw for await / for of. +// +// stopWhen stops pulling from the source as soon as the predicate is truthy. +// The predicate is checked BEFORE yielding: the triggering item is pulled but +// not offered downstream ("cancel before work"). Stopping is a clean +// completion for every consumer — cancel is a shorter run, not a source error. + +const forwardedLength = items => + typeof items?.length === 'number' ? {length: items.length} : {} + +export const stopWhen = (items, predicate = () => false) => ({ + ...forwardedLength(items), + async * [Symbol.asyncIterator] () { + let index = 0 + for await (const item of items) { + if (predicate(item, index)) + return + yield item + index++ + } + }, +}) + +export const stopWhenSync = (items, predicate = () => false) => ({ + ...forwardedLength(items), + * [Symbol.iterator] () { + let index = 0 + for (const item of items) { + if (predicate(item, index)) + return + yield item + index++ + } + }, +}) diff --git a/src/shared.js b/src/shared.js index 31398fe..a425c01 100644 --- a/src/shared.js +++ b/src/shared.js @@ -1,6 +1,15 @@ export const getKnownTotal = (items, total) => total !== undefined ? total : items.length +// A where() pattern is a plain data object. Anything that can be iterated +// (arrays, generators, source adapters) is an input, not a pattern. +export const isPatternObject = x => + x !== null && + typeof x === 'object' && + !Array.isArray(x) && + !(Symbol.asyncIterator in x) && + !(Symbol.iterator in x) + export const getPlannedTotal = ({items, take, total}) => { const knownTotal = getKnownTotal(items, total) diff --git a/tests/stop-when-sync.test.js b/tests/stop-when-sync.test.js new file mode 100644 index 0000000..575ea26 --- /dev/null +++ b/tests/stop-when-sync.test.js @@ -0,0 +1,151 @@ +import {test, expect} from 'vitest' +import {filterSync, findSync, seriesSync} from '$src/functional-sync' +import {stopWhenSync} from '$src/iterables' + +test('yields items until the predicate fires', () => { + const items = [...stopWhenSync([1, 2, 3, 4], n => n === 3)] + expect(items).toEqual([1, 2]) +}) + +test('yields everything when the predicate never fires', () => { + const items = [...stopWhenSync([1, 2], () => false)] + expect(items).toEqual([1, 2]) +}) + +test('yields nothing when the predicate fires immediately', () => { + const items = [...stopWhenSync([1, 2], () => true)] + expect(items).toEqual([]) +}) + +test('passes each examined item to the predicate', () => { + const seen = [] + const items = [...stopWhenSync([1, 2, 3], item => { + seen.push(item) + return false + })] + expect(items).toEqual([1, 2, 3]) + expect(seen).toEqual([1, 2, 3]) +}) + +test('passes the index as second argument', () => { + const seen = [] + const items = [...stopWhenSync(['a', 'b'], (item, index) => { + seen.push([item, index]) + return false + })] + expect(items).toEqual(['a', 'b']) + expect(seen).toEqual([['a', 0], ['b', 1]]) +}) + +test('default predicate is never true', () => { + const items = [...stopWhenSync([1, 2, 3])] + expect(items).toEqual([1, 2, 3]) +}) + +test('works over any sync iterable', () => { + const items = [...stopWhenSync(new Set(['a', 'b', 'c']), l => l === 'c')] + expect(items).toEqual(['a', 'b']) +}) + +test('stops pulling from the source once the predicate fired', () => { + const produced = [] + function * source () { + produced.push(1) + yield 1 + produced.push(2) + yield 2 + produced.push(3) + yield 3 + } + + const items = [...stopWhenSync(source(), n => n === 2)] + + expect(items).toEqual([1]) + expect(produced).toEqual([1, 2]) +}) + +test('early stop still runs the source cleanup', () => { + const state = {cleanedUp: false} + function * gen () { + try { + yield 1 + yield 2 + yield 3 + } finally { + state.cleanedUp = true + } + } + const items = [...stopWhenSync(gen(), n => n === 2)] + expect(items).toEqual([1]) + expect(state.cleanedUp).toBe(true) +}) + +test('seriesSync consumes a stopped source and reports a clean run', () => { + const result = seriesSync( + stopWhenSync([1, 2, 3, 4], n => n === 3), + x => x * 10, + ) + expect(result).toEqual({ + results: [10, 20], errors: [], sourceErrors: [], failure: false, + }) +}) + +test('forwards array length so seriesSync keeps progress totals', () => { + const progress = [] + seriesSync(stopWhenSync([1, 2, 3, 4], n => n === 3), x => x * 10, { + onProgress: value => progress.push(value), + }) + expect(progress).toEqual([ + { + item: 1, result: 10, index: 0, total: 4, + }, + { + item: 2, result: 20, index: 1, total: 4, + }, + ]) +}) + +test('a throwing predicate surfaces as a source error downstream', () => { + const bang = new Error('bang') + const result = seriesSync( + stopWhenSync([1, 2, 3], item => { + if (item === 2) + throw bang + return false + }), + x => x, + ) + + expect(result.results).toEqual([1]) + expect(result.sourceErrors).toEqual([{error: bang, index: 1}]) +}) + +test('filterSync inherits the stopped source', () => { + const result = filterSync( + stopWhenSync([1, 2, 3, 4], n => n === 3), + x => x % 2 === 0, + ) + expect(result.results).toEqual([2]) +}) + +// isPatternObject fix — iterable objects were misread as where() patterns + +test('filterSync accepts sync generator sources directly', () => { + function * numbers () { + yield 1 + yield 2 + yield 3 + } + const result = filterSync(numbers(), x => x % 2 === 1) + expect(result.results).toEqual([1, 3]) +}) + +test('findSync accepts generator sources directly', () => { + function * numbers () { + yield 1 + yield 2 + yield 3 + } + const result = findSync(numbers(), x => x > 1) + expect(result.result).toBe(2) +}) diff --git a/tests/stop-when.test.js b/tests/stop-when.test.js new file mode 100644 index 0000000..d677fc2 --- /dev/null +++ b/tests/stop-when.test.js @@ -0,0 +1,199 @@ +import {test, expect, vi} from 'vitest' +import {filter, reduce, series} from '$src/functional' +import {stopWhen} from '$src/iterables' +import {trackedSource} from './source-errors-helpers.js' + +const collect = async iterable => { + const items = [] + for await (const item of iterable) + items.push(item) + return items +} + +// Adapter contract — ported from nucube-app stop-when.test.js + +test('yields items until the predicate fires', async () => { + const items = await collect(stopWhen([1, 2, 3, 4], n => n === 3)) + expect(items).toEqual([1, 2]) +}) + +test('yields everything when the predicate never fires', async () => { + const items = await collect(stopWhen([1, 2], () => false)) + expect(items).toEqual([1, 2]) +}) + +test('yields nothing when the predicate fires immediately', async () => { + const items = await collect(stopWhen([1, 2], () => true)) + expect(items).toEqual([]) +}) + +test('passes each examined item to the predicate', async () => { + const seen = [] + await collect(stopWhen([1, 2, 3], item => { + seen.push(item) + return false + })) + expect(seen).toEqual([1, 2, 3]) +}) + +test('works over async iterables', async () => { + const source = (async function * () { + yield 'a' + yield 'b' + }()) + const items = await collect(stopWhen(source, letter => letter === 'b')) + expect(items).toEqual(['a']) +}) + +test('stops pulling from the source once the predicate fired', async () => { + const produced = [] + const source = (async function * () { + produced.push(1) + yield 1 + produced.push(2) + yield 2 + produced.push(3) + yield 3 + }()) + + const items = await collect(stopWhen(source, n => n === 2)) + + expect(items).toEqual([1]) + expect(produced).toEqual([1, 2]) +}) + +// Pipelean additions + +test('passes the index as second argument', async () => { + const seen = [] + await collect(stopWhen(['a', 'b', 'c'], (item, index) => { + seen.push([item, index]) + return false + })) + expect(seen).toEqual([['a', 0], ['b', 1], ['c', 2]]) +}) + +test('default predicate is never true', async () => { + const items = await collect(stopWhen([1, 2, 3])) + expect(items).toEqual([1, 2, 3]) +}) + +test('predicate may close over counters like shouldHalt', async () => { + let checks = 0 + const items = await collect(stopWhen([1, 2, 3, 4], () => ++checks > 2)) + expect(items).toEqual([1, 2]) +}) + +test('early stop still runs the source cleanup', async () => { + const {gen, state} = trackedSource([1, 2, 3]) + await collect(stopWhen(gen(), n => n === 2)) + expect(state.cleanedUp).toBe(true) +}) + +test('series consumes a stopped source and reports a clean run', async () => { + const result = await series(stopWhen([1, 2, 3, 4], n => n === 3), x => x * 10) + expect(result).toEqual({ + results: [10, 20], errors: [], sourceErrors: [], failure: false, + }) +}) + +test('reduce consumes a stopped source without a new option', async () => { + const { + value, errors, sourceErrors, failure, + } = await reduce( + stopWhen([1, 2, 3, 4], n => n === 3), + (acc, x) => acc + x, + 0, + ) + expect(value).toBe(3) + expect(errors).toEqual([]) + expect(sourceErrors).toEqual([]) + expect(failure).toBe(false) +}) + +test('forwards array length so series keeps progress totals', async () => { + const progress = [] + const result = await series( + stopWhen([1, 2, 3, 4], n => n === 3), + x => x * 10, + {onProgress: value => progress.push(value)}, + ) + + expect(result.results).toEqual([10, 20]) + expect(progress).toEqual([ + { + item: 1, result: 10, index: 0, total: 4, + }, + { + item: 2, result: 20, index: 1, total: 4, + }, + ]) +}) + +test('generator sources still have no known total', async () => { + async function * numbers () { + yield 1 + yield 2 + } + + const progress = [] + await series(stopWhen(numbers(), () => false), x => x, { + onProgress: value => progress.push(value), + }) + expect(progress).toHaveLength(2) + expect(Object.hasOwn(progress[0], 'total')).toBe(false) +}) + +test('a throwing predicate surfaces as a source error downstream', async () => { + const bang = new Error('bang') + const onSourceError = vi.fn() + const onError = vi.fn() + const result = await series( + stopWhen([1, 2, 3], item => { + if (item === 2) + throw bang + return false + }), + x => x, + {onError, onSourceError}, + ) + + expect(result.results).toEqual([1]) + expect(result.sourceErrors).toEqual([{error: bang, index: 1}]) + expect(onSourceError).toHaveBeenCalledWith({error: bang, index: 1}) + expect(onError).not.toHaveBeenCalled() +}) + +test('composes with take', async () => { + const result = await series( + stopWhen([1, 2, 3, 4, 5], n => n === 5), + x => x * 10, + {take: 2}, + ) + expect(result.results).toEqual([10, 20]) +}) + +test('filter inherits the stopped source', async () => { + const result = await filter( + stopWhen([1, 2, 3, 4], n => n === 3), + x => x % 2 === 1, + ) + expect(result.results).toEqual([1]) +}) + +// isPatternObject fix — iterable objects were misread as where() patterns + +test('filter accepts async generator sources directly', async () => { + async function * numbers () { + yield 1 + yield 2 + yield 3 + } + const result = await filter(numbers(), x => x % 2 === 1) + expect(result.results).toEqual([1, 3]) +}) + +test('filter still accepts pattern objects', async () => { + const result = await filter([{active: true}, {active: false}], {active: true}) + expect(result.results).toEqual([{active: true}]) +})