Elixir's Task.async_stream/3, living at its JS-idiomatic home: maps fn over the source with at most maxConcurrency invocations in flight, lazily — nothing starts until a consumer pulls, and the window refills only as results are taken. Results keep source order by default (ordered: false yields completion order — faster when element durations vary). Each element may deadline (timeoutMs) into a declared Failure('AsyncTimeout') element — the railway, not a crash; the element's signal fires so cancellation-aware work stops. A throw from fn stays a bug. Failure elements from the source pass through unmapped.

const doubled = await Stream.asyncStream([1, 2, 3, 4], (n) => n * 2, { maxConcurrency: 2 }).collect();
doubled; // [2, 4, 6, 8] — at most 2 in flight at any moment

The one push source: a handle whose producer emits whenever it likes, and a Stream that consumes what it emitted. Every other builder pulls — unfold, resource and iterate all ask for the next element — but the most common real source in JS cannot be asked: an EventEmitter, a WebSocket, fs.watch, SSE, a browser running tests.

Backpressure is not on offer for such a producer, so the ruling is explicit. Node's own adapter assumes the producer is pausable (events.on(emitter, 'x', { highWaterMark: 2 }) throws emitter.pause is not a function on a plain EventEmitter, and without it queues without bound); web streams report desiredSize but enforce nothing. GenStage is the one design that makes the caller choose, and this is its :buffer_size / :buffer_keep pair: buffer up to capacity, and past it drop from one end or the other. There is no third option — you cannot have both "no loss" and "bounded memory" from a producer that will not slow down.

onDemand is what makes the good case reachable: it fires when a consumer actually attaches, so a producer that can defer starting keeps the buffer near empty instead of racing ahead of a consumer that is not there yet. Measured against an unslowable producer of 5000 events, attaching from the first event held the peak buffer at 50 with nothing dropped; attaching 80ms late dropped 2250 at capacity: 1000, or held 3650 in memory uncapped.

A consumer attaches when its Task is awaited, not when it is built. channel.stream .collect() returns a lazy Task like every other consumer here, so nothing drains until something awaits it, and everything emitted in the meantime goes to the buffer. That is the module's laziness working as designed rather than an exception to it — but a live producer is where it becomes visible, so onDemand is the hook that ties the two together.

A producer that can slow down should honour emit's return value: if (!emit(x)) await ready() moves 200 elements through a buffer of 10 without losing one. Merely yielding to the microtask queue between emits does not — the consumer's own path through the generator costs several turns per element, so it is outrun about three to one and overflows anyway.

fail puts a declared failure in the flow (the railway); abort rejects the consuming Task — the two-tier rule, kept. For overflow that must be fatal rather than lossy, call either one from onDiscard; that is why there is no third overflow mode, and why the failure type stays whatever you declared instead of widening for a case you may not use.

One consumer: the buffer is drained, not replayed, so a second pass would take the elements the first was owed. Opening one is a bug and throws — this being silent is exactly the class of defect a push source should not have.

const channel = Stream.channel<number>();
const collected = channel.stream.take(2).collect();
channel.emit(1);
channel.emit(2);
await collected; // [1, 2]

StreamClass#chunkBy as a static: Stream.chunkBy(source, …) is Stream.from(source).chunkBy(…).

await Stream.chunkBy([1, 1, 2], (n) => n).collect(); // [[1, 1], [2]]

StreamClass#chunkEvery as a static: Stream.chunkEvery(source, …) is Stream.from(source).chunkEvery(…).

await Stream.chunkEvery([1, 2, 3], 2).collect(); // [[1, 2], [3]]

StreamClass#chunkWhile as a static: Stream.chunkWhile(source, …) is Stream.from(source).chunkWhile(…).

await Stream.chunkWhile([1, 2], 0, (v, acc) => ({ acc: acc + v, emit: [v] })).collect(); // [[1], [2]]

StreamClass#collect as a static: Stream.collect(source, …) is Stream.from(source).collect(…).

await Stream.collect([1, 2]); // [1, 2]

Concatenates sources in order — Elixir's Stream.concat, variadic.

await Stream.concat([1, 2], [3]).collect(); // [1, 2, 3]

Repeats a re-iterable source forever — always bound it (take, takeWhile).

await Stream.cycle([1, 2]).take(5).collect(); // [1, 2, 1, 2, 1]

StreamClass#dedup as a static: Stream.dedup(source, …) is Stream.from(source).dedup(…).

await Stream.dedup([1, 1, 2, 1]).collect(); // [1, 2, 1]

StreamClass#dedupBy as a static: Stream.dedupBy(source, …) is Stream.from(source).dedupBy(…).

await Stream.dedupBy([1, 1, 2], (n) => n).collect(); // [1, 2]

StreamClass#drop as a static: Stream.drop(source, …) is Stream.from(source).drop(…).

await Stream.drop([1, 2, 3], 1).collect(); // [2, 3]

StreamClass#dropEvery as a static: Stream.dropEvery(source, …) is Stream.from(source).dropEvery(…).

await Stream.dropEvery([1, 2, 3, 4], 2).collect(); // [2, 4]

StreamClass#dropWhile as a static: Stream.dropWhile(source, …) is Stream.from(source).dropWhile(…).

await Stream.dropWhile([1, 2, 3], (n) => n < 3).collect(); // [3]

count copies of one value.

await Stream.duplicate('x', 3).collect(); // ['x', 'x', 'x']

StreamClass#every as a static: Stream.every(source, …) is Stream.from(source).every(…).

await Stream.every([2, 4], (n) => n % 2 === 0); // true

StreamClass#filter as a static: Stream.filter(source, …) is Stream.from(source).filter(…).

await Stream.filter([1, 2, 3], (n) => n % 2 === 1).collect(); // [1, 3]

StreamClass#find as a static: Stream.find(source, …) is Stream.from(source).find(…).

await Stream.find([1, 2], (n) => n > 1); // 2

StreamClass#flatMap as a static: Stream.flatMap(source, …) is Stream.from(source).flatMap(…).

await Stream.flatMap([1, 2], (n) => [n, n]).collect(); // [1, 1, 2, 2]

StreamClass#forEach as a static: Stream.forEach(source, …) is Stream.from(source).forEach(…).

await Stream.forEach([1, 2], () => {}); // undefined

Lifts a source into a Stream. Elements that are Failures are the E channel from the start; everything else is T.

const doubled = await Stream.from([1, 2, 3]).map((n) => n * 2).collect();
doubled; // [2, 4, 6]

The infinite integers offset, offset + 1, … — Elixir's Stream.from_index/1.

await Stream.fromIndex(10).take(3).collect(); // [10, 11, 12]

StreamClass#intersperse as a static: Stream.intersperse(source, …) is Stream.from(source).intersperse(…).

await Stream.intersperse([1, 2], 0).collect(); // [1, 0, 2]

Emits 0, 1, 2, … every ms milliseconds, forever — Elixir's Stream.interval/1, on setTimeout (universal). Infinite: always bound it.

await Stream.interval(1).take(3).collect(); // [0, 1, 2]

StreamClass#into as a static: Stream.into(source, …) is Stream.from(source).into(…).

await Stream.into([1, 2], new WritableStream()).collect(); // [1, 2]

seed, fn(seed), fn(fn(seed)), … — Elixir's Stream.iterate/2, infinite.

await Stream.iterate(1, (n) => n * 2).take(4).collect(); // [1, 2, 4, 8]

Decodes a byte stream into text lines — the front door for NDJSON, logs, and SSE-ish feeds. Handles multi-byte characters and lines split across chunk boundaries; the final unterminated line is flushed.

const bytes = [new TextEncoder().encode('a\nb'), new TextEncoder().encode('c\nd')];
await Stream.lines(bytes).collect(); // ['a', 'bc', 'd']

StreamClass#map as a static: Stream.map(source, …) is Stream.from(source).map(…).

await Stream.map([1, 2], (n) => n * 2).collect(); // [2, 4]

StreamClass#mapConcurrent as a static: Stream.mapConcurrent(source, …) is Stream.from(source).mapConcurrent(…).

await Stream.mapConcurrent([1, 2], (n) => n * 2, { maxConcurrency: 2 }).collect(); // [2, 4]

StreamClass#mapEvery as a static: Stream.mapEvery(source, …) is Stream.from(source).mapEvery(…).

await Stream.mapEvery([1, 2, 3, 4], 2, (n) => n * 10).collect(); // [10, 2, 30, 4]

StreamClass#partition as a static: Stream.partition(source, …) is Stream.from(source).partition(…).

(await Stream.partition([1, 2])).values; // [1, 2]

StreamClass#reduce as a static: Stream.reduce(source, …) is Stream.from(source).reduce(…).

await Stream.reduce([1, 2, 3], (sum, n) => sum + n, 0); // 6

StreamClass#reject as a static: Stream.reject(source, …) is Stream.from(source).reject(…).

await Stream.reject([1, 2, 3], (n) => n === 2).collect(); // [1, 3]

Calls fn for each pull, forever — Elixir's Stream.repeatedly/1.

let n = 0;
await Stream.repeatedly(() => ++n).take(3).collect(); // [1, 2, 3]

unfold with lifecycle hooks — Elixir's Stream.resource/3: start runs on first pull, next is unfold's step, and after ALWAYS runs — normal end, early take, or a throw. (Plain generators already get this via try/finally; resource is for when the setup/teardown pair deserves to be explicit.)

const opened: string[] = [];
const rows = Stream.resource(
  () => (opened.push('open'), { cursor: 0 }),
  (db) => (db.cursor < 2 ? ([`row-${db.cursor}`, { cursor: db.cursor + 1 }] as const) : null),
  () => void opened.push('close'),
);
await rows.collect(); // ['row-0', 'row-1'] — and opened is ['open', 'close']

StreamClass#results as a static: Stream.results(source, …) is Stream.from(source).results(…).

(await Stream.results([1, 2])).length; // 2

StreamClass#run as a static: Stream.run(source, …) is Stream.from(source).run(…).

await Stream.run([1, 2]); // undefined

StreamClass#some as a static: Stream.some(source, …) is Stream.from(source).some(…).

await Stream.some([1, 2], (n) => n > 1); // true

StreamClass#take as a static: Stream.take(source, …) is Stream.from(source).take(…).

await Stream.take([1, 2, 3], 2).collect(); // [1, 2]

StreamClass#takeEvery as a static: Stream.takeEvery(source, …) is Stream.from(source).takeEvery(…).

await Stream.takeEvery([1, 2, 3, 4], 2).collect(); // [1, 3]

StreamClass#takeWhile as a static: Stream.takeWhile(source, …) is Stream.from(source).takeWhile(…).

await Stream.takeWhile([1, 2, 3], (n) => n < 3).collect(); // [1, 2]

StreamClass#tap as a static: Stream.tap(source, …) is Stream.from(source).tap(…).

await Stream.tap([1, 2], () => {}).collect(); // [1, 2]

StreamClass#through as a static: Stream.through(source, …) is Stream.from(source).through(…).

await Stream.through([1, 2], async function* (els) { yield* els; }).collect(); // [1, 2]

Emits a single 0 after ms milliseconds, then ends — Elixir's Stream.timer/1.

await Stream.timer(1).collect(); // [0]

Elixir's Stream.unfold/2: grow a lazy stream from a seed. next(state) returns [element, nextState] to emit and continue, or null to end; return [failure, null] to emit a declared failure and stop. Nothing runs until a consumer pulls — pagination only fetches the pages the consumer actually reaches.

const countdown = Stream.unfold(3, (n) => (n === 0 ? null : [n, n - 1] as const));
await countdown.collect(); // [3, 2, 1]

StreamClass#uniq as a static: Stream.uniq(source, …) is Stream.from(source).uniq(…).

await Stream.uniq([1, 1, 2, 1]).collect(); // [1, 2]

StreamClass#uniqBy as a static: Stream.uniqBy(source, …) is Stream.from(source).uniqBy(…).

await Stream.uniqBy([1, 1, 2], (n) => n).collect(); // [1, 2]

StreamClass#withIndex as a static: Stream.withIndex(source, …) is Stream.from(source).withIndex(…).

await Stream.withIndex(['a', 'b']).collect(); // [['a', 0], ['b', 1]]

Zips sources in lockstep, ending at the shortest (Elixir's Stream.zip) and closing the longer sources so their cleanup runs. JS divergences: sources are pulled sequentially per slot, and a failure element passes through bare without consuming from the other sources — failures cannot pair.

await Stream.zip([1, 2, 3], ['a', 'b']).collect(); // [[1, 'a'], [2, 'b']]

zip plus a combiner — Elixir's Stream.zip_with; the tuple arrives spread.

await Stream.zipWith([[1, 2], [10, 20]], (a, b) => a + b).collect(); // [11, 22]

Streams are for await-able; elements arrive bare (T | E), so a hand-rolled loop discriminates with Failure.is exactly like any other union consumer.

const seen: number[] = [];
for await (const element of Stream.from([1, 2])) seen.push(element);
seen; // [1, 2]

Chunks consecutive values sharing a key — Elixir's chunk_by/2; a key change closes the chunk. Failures are transparent: they pass through without closing the chunk around them, consistent with dedup.

await Stream.from([1, 3, 2, 4, 5]).chunkBy((n) => n % 2).collect(); // [[1, 3], [2, 4], [5]]

Groups values into arrays of count, sliding by step (Elixir's full chunk_every/4): step < count overlaps windows, step > count skips between them. leftover rules the trailing partial chunk: emitted as-is by default, 'discard' drops it, an array pads it up to count. Failures pass through between chunks without breaking the one being assembled — a bad row never voids the batch around it.

await Stream.from([1, 2, 3, 4, 5]).chunkEvery(2).collect(); // [[1, 2], [3, 4], [5]]
await Stream.from([1, 2, 3, 4, 5]).chunkEvery(3, 2, 'discard').collect(); // [[1, 2, 3], [3, 4, 5]]
await Stream.from([1, 2, 3, 4]).chunkEvery(3, 3, [0, 0]).collect(); // [[1, 2, 3], [4, 0, 0]]

The general chunker — Elixir's chunk_while/4, object-shaped for JS: step returns { acc } to keep accumulating, { acc, emit } to emit a chunk, plus halt: true to end the stream; flush may emit one last chunk from the final accumulator. Values only; failures pass through.

const batchesOfTwo = Stream.from([1, 2, 3, 4, 5]).chunkWhile(
  [] as number[],
  (n, acc) => (acc.length === 1 ? { acc: [], emit: [...acc, n] } : { acc: [...acc, n] }),
  (acc) => (acc.length > 0 ? acc : undefined),
);
await batchesOfTwo.collect(); // [[1, 2], [3, 4], [5]]

Collects every value, fail-fast: the first failure element rejects the Task with it (declared — so .result() reflects it bare and typed). The Result.all of streams.

NOT values(): in JS .values() produces an iterator (Array, Map, Set, ReadableStream) rather than consuming one, and partition().values filters failures out while this throws on them — the same word for opposite behaviour inside one module.

NOT all() either, tempting as the Result.all mirror was: every other language in this module's lineage spends "all" on the predicate — Rust's StreamExt::all(pred), Elixir's Enum.all?/2 — which is StreamClass#every here. collect is Rust's word for gathering and collides with nothing.

Rust gets one collect for both this and StreamClass#results because the target type picks the behaviour; JS cannot dispatch on return type, so the two need two names.

await Stream.from(['a', 'b']).collect(); // ['a', 'b']

Drops consecutive duplicate values — dedupBy with identity.

await Stream.from([1, 1, 2, 1]).dedup().collect(); // [1, 2, 1]

Drops consecutive duplicate values by key — failures are transparent: they pass through without resetting the last-seen memory.

await Stream.from([1, 1, 2, 2, 1]).dedupBy((n) => n).collect(); // [1, 2, 1]

Discards the first count elements — positional like StreamClass#take, so values and failures both count (dropping a prefix is an explicit consumer decision).

await Stream.from([1, 2, 3, 4]).drop(2).collect(); // [3, 4]

Drops the values at positions 0, every, 2·every, … (Elixir's drop_every/2, counting values; failures pass through).

await Stream.from([0, 1, 2, 3, 4]).dropEvery(2).collect(); // [1, 3]

Discards values while predicate holds, then everything flows. Failures met during the dropping phase still pass through — a declared failure is never silently swallowed by a value predicate.

await Stream.from([1, 2, 9, 1]).dropWhile((n) => n < 5).collect(); // [9, 1]

Whether every value satisfies predicate — short-circuiting on the first that does not.

This is the member Rust and Elixir both call all; JS calls it every, and JS wins here for the same reason forEach did. StreamClass#collect is the collector, and the two are told apart by their arguments as much as their names: every takes a predicate, collect takes nothing.

Vacuously true on an empty stream, matching Array.prototype.every.

await Stream.from([2, 4]).every((n) => n % 2 === 0); // true

Keeps the values predicate accepts; failure elements always pass through — dropping a declared failure is a decision for a consumer (partition), never a side effect of filtering values.

await Stream.from([1, 2, 3, 4]).filter((n) => n % 2 === 0).collect(); // [2, 4]

The first value satisfying predicate, or undefined — short-circuiting, so an infinite source is fine as long as a match exists.

undefined rather than a Failure for "not found": absence is an ordinary answer to a search, not a failure of the run, and Array.prototype.find sets the expectation.

await Stream.from([1, 2, 3]).find((n) => n > 1); // 2

Expands each value into a source and flattens it in order — pages into rows, rows into cells. Failure elements pass through unexpanded.

await Stream.from([[1, 2], [3]]).flatMap((page) => page).collect(); // [1, 2, 3]

Drains the stream, running fn per value — fail-fast like StreamClass#collect. The Task resolves when the source ends; use it when the work is the side effect.

Elixir's Stream.each/2 is the lazy tap, which lives here as StreamClass#tap because that is the name JS gave it. This is the terminal drain, and it takes JS's name for that: every iterable in the language spells it forEach — Array, Map, Set, and Iterator.prototype since ES2025 — as do Rust's StreamExt::for_each and TC39's pending async iterator helpers. Array.prototype.each has never existed.

Sequential, for the same reason StreamClass#map is: the next element is not pulled until fn has settled. That is what makes it safe to write to a database from here, and what makes it the wrong place to fan out — StreamClass.asyncStream is.

const seen: number[] = [];
await Stream.from([1, 2, 3]).forEach((n) => void seen.push(n));
seen; // [1, 2, 3]

Emits separator between every two elements — positional, so it also separates a value from a passing failure.

await Stream.from([1, 2, 3]).intersperse(0).collect(); // [1, 0, 2, 0, 3]

Tees every value into a web WritableStream while passing all elements through — Elixir's Stream.into/2, with the platform's own sink as the Collectable. The sink is closed on normal completion and released on early exit; failures pass the tee but are not written (the sink types T, not T | E).

const written: number[] = [];
const sink = new WritableStream<number>({ write: (n) => void written.push(n) });
await Stream.from([1, 2, 3]).into(sink).collect(); // [1, 2, 3]
written; // [1, 2, 3]

Transforms each value; failure elements flow past untouched (the railway, per element). fn may itself return a Failure — that widens the stream's E, which is how a parse step turns raw input into Row | BadRow without a wrapper per element. A throw inside fn is a bug and rejects the consumer.

fn is awaited one element at a time. map((url) => fetch(url)) issues its requests strictly in sequence — six 60ms calls take 360ms, not 60ms. That is not an oversight to route around: a stream is one suspended generator, so it has exactly one resume point, and the same constraint is why TC39's async iterator helpers are still unshipped over precisely this question. For concurrent per-element work use StreamClass.asyncStream, whose maxConcurrency bounds the window and whose ordered decides whether results keep source order or arrive as they finish.

const tagged = await Stream.from([1, 2]).map((n, i) => `${i}:${n}`).collect();
tagged; // ['0:1', '1:2']

StreamClass.asyncStream as a stage: the same bounded fan-out, usable in the middle of a pipeline instead of only at the start of one.

const rows = await Stream.from([1, 2, 3, 4])
  .filter((n) => n % 2 === 0)
  .mapConcurrent((n) => n * 10, { maxConcurrency: 2 })
  .collect();
rows; // [20, 40]

Why concurrency is a stage rather than a .limit(n) you append. The obvious API is TC39's — .map(fetch).limit(5) — and it cannot work on top of async generators. A generator has one suspended body, so it has exactly one resume point, and it serializes .next() no matter how many calls are in flight. Measured: a downstream limiter aggressively keeping three .next() calls outstanding over a generator-backed map of six 40ms elements still took 242ms at peak concurrency 1 — identical to serial. Nothing downstream can parallelise work that upstream has already awaited.

So the stage that performs the async work is the only place that can introduce concurrency, which is what this is. It is also why TC39's helpers are still unshipped: to make .limit composable they must first respecify the helpers as something other than async generators.

Transforms the values at positions 0, every, 2·every, …, passing the rest — and every failure — through unchanged (Elixir's map_every/3).

await Stream.from([1, 1, 1, 1]).mapEvery(2, (n) => n * 10).collect(); // [10, 1, 10, 1]

Splits the stream into { values, errors }, keeping both — Result.partition, fed by the stream. Literally results() piped through partition: the three modules share one vocabulary, so the composition is one line.

import * as Failure from '../result/failure.ts';

const Bad = Failure.define('Bad', 'bad element');
const { values, errors } = await Stream.from([1, Bad(), 2]).partition();
values; // [1, 2]
errors.length; // 1

Folds every value into one, fail-fast like StreamClass#collect — the terminal counterpart to StreamClass#scan's lazy running fold, and the reason a ten-million-row stream can answer with a single number.

Without it the only route is (await stream.collect()).reduce(…), which buffers the whole source to produce one value and gives back everything the module exists to avoid. Elixir does not need a member here because Enum.reduce accepts any Enumerable; JS has no such fallback.

initial is required. The seedless form would have to raise on an empty stream — the way [].reduce(fn) does — and a stream's emptiness is not knowable before it is drained, so the failure would arrive at the worst possible moment. Naming the seed also names A.

const total = await Stream.from([1, 2, 3]).reduce((sum, n) => sum + n, 0);
total; // 6

Drops the values predicate accepts — filter's complement; failures pass through.

await Stream.from([1, 2, 3, 4]).reject((n) => n % 2 === 0).collect(); // [1, 3]

Collects every element, positionally, failures included — the Task.results of streams. Each failure is reported to the observation seam as it is classified into the value world, so tracing sees per-element failures a consumer chose to keep.

import * as Failure from '../result/failure.ts';

const Odd = Failure.define('Odd', (d: { n: number }) => `${d.n} is odd`);
const outcomes = await Stream.from([2, Odd({ n: 3 }), 4]).results();
outcomes.length; // 3 — nothing lost, order kept

Forces the stream for its side effects alone — Elixir's Stream.run/1. Fail-fast like StreamClass#forEach: pair with StreamClass#tap for the effects.

const seen: number[] = [];
await Stream.from([1, 2]).tap((n) => void seen.push(n)).run();
seen; // [1, 2]

Emits the running fold of the values — Elixir's Stream.scan; without initial the first value seeds the accumulator. Failures pass through and leave the accumulator untouched.

await Stream.from([1, 2, 3]).scan((acc, n) => acc + n).collect(); // [1, 3, 6]

Whether any value satisfies predicate — short-circuiting: the source is not pulled past the first match. JS's name for Elixir's Enum.any?/2 and Rust's StreamExt::any.

Fail-fast like the other terminals: a failure element rejects, because a stream that could not produce one of its values cannot honestly answer a question about all of them.

await Stream.from([1, 2, 3]).some((n) => n > 2); // true

Ends the stream after count elements (values and failures both count — take bounds work, it does not editorialize). The source is never pulled past the cut, which is the whole point over an eager slice: unfold pagination stops fetching.

let pulled = 0;
const firstTwo = Stream.unfold(0, (n) => ((pulled += 1), [n, n + 1] as const)).take(2);
await firstTwo.collect(); // [0, 1] — and pulled is 2, not ∞

Keeps the values at positions 0, every, 2·every, … (Elixir's take_every/2, counting values; failures pass through). every of 0 keeps nothing.

await Stream.from([0, 1, 2, 3, 4]).takeEvery(2).collect(); // [0, 2, 4]

Ends the stream at the first value predicate refuses. Failures inside the window pass through and are never tested — the predicate speaks about values only.

await Stream.from([1, 2, 9, 1]).takeWhile((n) => n < 5).collect(); // [1, 2]

Runs fn per value as a lazy side effect and passes everything through unchanged — Elixir's lazy Stream.each/2, under the idiomatic JS name. Nothing runs until a consumer pulls; pair with StreamClass#run for side effects alone.

const seen: number[] = [];
await Stream.from([1, 2]).tap((n) => void seen.push(n)).collect(); // [1, 2]
seen; // [1, 2]

The general engine — Elixir's Stream.transform, JS-shaped: hand the raw element flow to your own async generator and yield whatever you want. The one transform where failures do not auto-pass: your generator sees them and owns the ruling. Every missing combinator is three lines away through this.

const pairs = Stream.from([1, 2, 3, 4]).through(async function* (elements) {
  let previous: number | undefined;
  for await (const n of elements) {
    if (previous !== undefined) yield [previous, n] as const;
    previous = n;
  }
});
await pairs.collect(); // [[1, 2], [2, 3], [3, 4]]

Keeps the first occurrence of each value — uniqBy with identity.

await Stream.from([1, 2, 1, 2, 3]).uniq().collect(); // [1, 2, 3]

Keeps the first occurrence of each value by key, stream-wide (holds a Set of seen keys — bound infinite streams). Failures pass through.

await Stream.from([1, 2, 1, 3, 2]).uniqBy((n) => n).collect(); // [1, 2, 3]

Pairs each value with its index: [value, index]. Failures pass bare and consume no index — pairs stay contiguous.

await Stream.from(['a', 'b']).withIndex(1).collect(); // [['a', 1], ['b', 2]]