fxdart

A functional programming library for Dart, ported from FxTS. Lazy evaluation, concurrent async iteration, and pipeline-style composition — the FxTS programming model, rebuilt on Dart's type system.

// 6 requests of 1s complete in ~2s — not ~6s.
await fx(userIds).toAsync().map(fetchUser).concurrent(3).toList();

---

🚀 Try it in your browser

👆 Click any badge above. Each one is a live, runnable site:

	Site	What it is
📚	FxDart 101	A guided course with an in-browser playground for every function
📒	Daily Ledger	A full app built with fxdart, running live
⚖️	Dart vs FxDart	53 problems solved both ways, with an honest verdict on each
⚡	RxDart vs FxDart	The same 50-example format vs RxDart — push streams vs pull pipelines, including the cases where RxDart is simply the right tool

---

📖 Contents

✨ Why fxdart? · 📦 Install · 🤖 AI agent skills · 🛠️ Usage · 📇 API overview · 🔀 Differences from FxTS · 🧪 Testing · 🙏 Acknowledgments

---

✨ Why fxdart?

🦥 Lazy evaluation

Operators build a pipeline and do no work until a terminal operator runs — so fx(hugeList).map(f).filter(g).take(3) only ever computes 3 results.

🔀 Concurrency you can dial

concurrent(n) evaluates the upstream chain n items at a time while preserving order — turning six 1-second requests into a ~2-second batch with one method call.

🛡️ Type-safe pipelines

The fx() chain keeps full static typing end to end. Sync operators are plain functions over native Iterables, so everything interops with ordinary Dart code.

🧠 One mental model for sync and async

The same operator names work on Iterable (sync) and FxAsyncIterable (async), with Stream bridges in both directions.

🎯 Typed errors

Kotlin Arrow 2.x's Raise/Either approach, ported: straight-line either blocks instead of flatMap pyramids, error accumulation with NonEmptyList, and validation fused directly into the concurrent pipelines above.

⚡ A push side too, when time matters

Pull pipelines model data over demand; fxEvents() models events over time on plain Dart Streams — debounce, throttle, sample, switchMap, combineLatest and friends — then hands you back to the typed pull world with .pull().

🧊 Dart names work too

Every FxTS name that Dart's collections already have a word for is also callable by that word: where, expand, flattened, nonNulls, sorted, indexed, firstWhereOrNull. No dialect to learn before you can read the code.

---

📦 Install

See the installation guide on pub.flutter-io.cn for the latest version.

---

🤖 AI agent skills

fxdart ships three Agent Skills that teach AI coding assistants — Claude Code, Codex, Devin, Antigravity, OpenCode, pi, and anything reading .agents/skills/ — when and how to use fxdart:

Skill	Covers
🔗 skills/fxdart-pipelines/	Collections, concurrent Futures, and complex pull flow logic
⚡ skills/fxdart-events/	Events over time: fxEvents, debounce, switchMap, combineLatest, the pull seam
🎯 skills/fxdart-typed-errors/	The typed-error system: either blocks, error accumulation, Either on pull and on events

Option A — the community skills CLI (auto-detects your IDE/agent):

dart pub global activate skills
skills get fxdart

Option B — fxdart's built-in zero-dependency installer:

# From a project that depends on fxdart:
dart run fxdart:install_skills              # auto-detects agent dirs in the project
dart run fxdart:install_skills claude codex # or name agents explicitly
dart run fxdart:install_skills all --global # per-user dirs (~/.claude/skills, ~/.agents/skills, ...)

# Or standalone:
dart pub global activate fxdart
fxdart_skills --global claude

Supported agents:

Agent	Install dir
claude	.claude/skills/
codex / antigravity / generic	.agents/skills/
devin	.devin/skills/
opencode	.opencode/skills/
pi	.pi/skills/ · global ~/.pi/agent/skills/

💡 --list shows install status · --remove uninstalls.

---

🛠️ Usage

🔗 Sync pipelines

Sync operators are data-first functions over lazy Iterables; the fx() chain composes them with full type inference:

import 'package:fxdart/fxdart.dart';

fx([1, 2, 3, 4, 5])
    .map((a) => a + 10)
    .filter((a) => a % 2 == 0)
    .toList(); // [12, 14]

// The same chain as a getter, reached from the collection itself:
[1, 2, 3, 4, 5].fx
    .map((a) => a + 10)
    .filter((a) => a % 2 == 0)
    .toList(); // [12, 14]

// Equivalent with top-level functions:
toList(filter((a) => a % 2 == 0, map((a) => a + 10, [1, 2, 3, 4, 5])));

// Laziness: only 3 squares are ever computed.
fx(range(1, 1000000)).map((a) => a * a).take(3).toList(); // [1, 4, 9]

Every entry point has a getter twin, and the name always carries fx so it is clear which library you are stepping into: .fx on an Iterable, FxAsyncIterable or Stream, .fxAsync on an iterable of futures, .fxEvents and .fxLive on a Stream, .fxShuffle on an Iterable, .fxDebounce / .fxThrottle on a callback. It builds the same thing and reads left to right when the source is itself a call: orders.where(isPaid).fx.groupBy(...). The operators stay on the chain rather than on Iterable — fifteen of them share a name with a member Iterable already has, and an instance member always wins. The docs use the function spellings throughout; the fx() tutorial has the full roster.

⏳ Async pipelines

Async operators work on FxAsyncIterable<T> — a pull-based protocol ported from FxTS's AsyncIterable handling. Lift values in with toAsync / fromStream (or .toAsync() on a chain), and out with .toList() / .toStream():

await fx([1, 2, 3, 4])
    .toAsync()
    .map((a) async => a + 10) // callbacks may be async
    .filter((a) => a % 2 == 0)
    .toList(); // [12, 14]

// Streams bridge both ways.
await fxStream(Stream.fromIterable([1, 2, 3])).map((a) => a * 2).toList();

⚡ Concurrency

concurrent(n) is FxTS's signature feature, ported faithfully: a concurrency marker travels backwards through the pipeline's iterator protocol, so the upstream chain evaluates n items at once while results stay in order.

// 6 requests of 1s complete in ~2s instead of ~6s.
await fx([1, 2, 3, 4, 5, 6])
    .toAsync()
    .map((id) => fetchUser(id))
    .concurrent(3)
    .toList();

• 🥇 concurrentPool(n) — the completion-order variant: faster first results, no ordering guarantee.

ℹ️ This back-channel protocol is why fxdart has its own FxAsyncIterable instead of building on push-based Streams, which cannot express it.

📡 Events (the push side)

Some problems really are events over time, not data over demand. fxEvents() wraps a plain Dart Stream in a chainable, Rx-flavoured API — a thin wrapper, never an extension, so it coexists with rxdart without member conflicts:

final results = await fxEvents(keystrokes)
    .debounce(const Duration(milliseconds: 160))
    .switchMap((q) => search(q).asStream()) // cancels the superseded search
    .toList();

// Cross back into the typed pull world at any point:
await fxEvents(ticks).sampleOn(clock).pull().map(load).concurrent(4).toList();

LiveValue holds a current-value stream, and FxSubscriptions cancels a bag of subscriptions together. See ⚡ RxDart vs FxDart for 50 worked examples — including the cases where RxDart is the better fit.

🎯 Typed errors

The either builder runs a block in a Raise<E> scope: each r.bind unwraps a success or short-circuits the whole block with a typed failure — the Kotlin Arrow 2.x model, ported (no TaskEither/IO wrapper tower, no Option; Dart's T? plus the nullable builder covers absence):

Either<String, int> parsePort(String raw) => either((r) {
  final n = r.ensureNotNull(int.tryParse(raw), () => '"$raw" is not a number');
  r.ensure(n > 0 && n < 65536, () => '$n is out of range');
  return n;
});

// Validation accumulates EVERY failure into a NonEmptyList, not just the first:
final user = either<Nel<String>, User>((r) => r.zipOrAccumulate2(
    (r) => validateName(r, input), (r) => validateAge(r, input), User.new));

// And it fuses with pipelines — fail-slow, 8 records in flight, order kept:
final result = await fxStream(records)
    .mapOrAccumulate<String, User>((r, rec) => parseUser(r, rec), concurrency: 8);

📚 Every subject has a detailed tutorial with an in-browser playground:

🗺️ overview · ↔️ Either · 🎬 either & the Raise scope · ❓ nullable · 📋 NonEmptyList · ➕ accumulation · 🔗 Either × pipelines

---

📇 API overview

	Category	Functions
🌱	Generate	range, repeat, cycle, entries, keys, values
🔄	Transform (lazy)	map, mapWithIndex, mapEffect, flatMap, flatMapWithIndex, flat, scan, scan1, peek, pluck, attach, using, mapAccum, mapCatching
🔍	Filter (lazy)	filter, filterWithIndex, reject, compact, uniq, uniqBy, uniqAdjacent, uniqAdjacentBy, takeUniqBy (strict; filter().uniqBy().take() in one inlinable call), difference, differenceBy, intersection, intersectionBy, compress, mapNotNull
✂️	Slice (lazy)	take, takeRight, takeWhile, takeWhileRight, takeUntilInclusive, drop, dropRight, dropWhile, dropWhileRight, dropUntil, slice, chunk, windowed, pairwise, split
🧩	Combine (lazy)	append, prepend, concat, zip, zip3, zipWith, zipWithIndex, transpose, reverse, fork, tee, tee3, unzip
📊	Aggregate	reduce, fold, foldWithIndex, foldRight, foldRightWithIndex, reduceLazy, toList, sum, sumBy, sumStrings, average, averageBy, min, minBy, max, maxBy, size, join, groupBy, indexBy, countBy, sort, sortBy, sortByDesc, toSorted, partition, each, consume, product, productBy, topBy, bottomBy
🎯	Access	head, last, nth, find, findIndex, includes, isEmpty, defaultIfEmpty, ifEmpty, every, some, none, firstNotNullOf
🗂️	Object (Map)	omit, pick, omitBy, pickBy, prop, props, evolve, fromEntries, mapKeys, mapValues, mapEntries, compactObject, resolveProps, isMatch, matches, toPairs
🧮	Function	pipe, pipe1, pipeLazy, identity, always, noop, tap, apply, juxt, memoize, negate, not, when, unless, throwError, throwIf, cases, add, gt, gte, lt, lte, delay, sleep, unicodeToList, .curried/.uncurried (extension getters, arity 2–5)
✅	Predicates	isNull, isNotNull, isNil, isBoolean/isBool, isNumber/isNum, isString, isDate/isDateTime, isList, isMap
🧊	Dart-idiomatic aliases	Every FxTS name that Dart's own collections already have a word for is also callable by that word: where, whereNot, expand, flattened, nonNulls, distinct, distinctBy, sorted, indexed, skip, skipWhile, takeLast, count, countWhere, any, forEach, firstOrNull, lastOrNull, firstWhereOrNull, elementAtOrNull, indexWhere
⏳	Async	Every lazy/aggregate operator has an *Async twin (mapAsync, toListAsync, …), plus toAsync, fromStream, mapConcurrent, concurrentAsync, concurrentPoolAsync, parallel / mapParallel (CPU, isolate pool), asyncEmpty
⚡	Events (push)	fxEvents() / FxEvents — an Rx-flavoured chain over plain Streams: debounce, throttle, sample, sampleOn, delay, spaceBy, startWith, startOn, stopOn, chunk, chunkEvery, chunkOn, switchMap, mergeMap, concatMap, exhaustMap, asyncMap, merge, mergeWith, race, raceWith, zip, zipWith, combineLatest, combineLatestAll, withLatestFrom, waitAll, share, retry, onErrorResume, attempt, mapEither, separated, onErrorReturn, scan, uniqAdjacent, uniqAdjacentBy, pairwise, take, drop/skip, head/firstOrNull, pull (back into the typed pull world). Plus LiveValue and FxSubscriptions
🎯	Typed errors	Either (Left/Right, fold, map, flatMap, recover, Either.catching, toEitherNel), either/eitherAsync, eitherCatching, nullable/nullableAsync, catching, foldRaise, NonEmptyList/Nel; in a Raise scope: r.bind, r.bindNel, r.ensure, r.ensureNotNull, r.accumulate, r.zipOrAccumulate2..5, r.mapOrAccumulate; free functions rights, lefts, separateEither, sequenceEither, mapOrAccumulate, flattenOrAccumulate; chain terminals rights(), lefts(), separated(), sequence(), mapOrAccumulate()
🧰	Util	debounce, throttle, retry, shuffle, createSeededRandom
⚙️	Config	FxDart.config / FxConfig — process-wide switches, read when a pipeline starts iterating
⛓️	Chains	fx() (sync, extends Iterable; .fx / .fxAsync getter twins), fxAsync(), fxStream(), fxEvents(); Fx<num>/FxAsync<num> gain sum/average/min/max/product

---

🔀 Differences from FxTS

Dart has no function overloads, variadic generics, or conditional types, so some APIs deliberately deviate:

	FxTS	fxdart
🍛	curried data-last (map(f) inside pipe)	fx() chain (typed) or dynamic pipe(value, [closures])
🔄	one map dispatching sync/async	map (Iterable) / mapAsync (FxAsyncIterable); chains use plain names
📊	reduce(f, seed, iter) overload	fold(seed, f, iter) (unseeded reduce(f, iter) unchanged)
📦	tuples (zip, entries, partition)	Dart records: (A, B)
🗂️	TS objects (omit, pick, evolve, …)	Map-based equivalents
❓	undefined	null (head/find/nth return T?)
📋	toArray / toArrayAsync	toList / toListAsync (Dart has no array type)
⏳	AsyncIterable / for await	FxAsyncIterable + toStream() / fromStream() bridges
🎛️	variadic zip/juxt/cases	fixed arities (zip/zip3) or list/record parameters
🍛	curry(f)	.curried / .uncurried extension getters — see WHY_CURRIED.md

🍛 Why .curried instead of curry?

FxTS's curry needs arity reflection and recursive conditional types, which Dart lacks — so fxdart curries through per-arity extensions instead, resolved statically and fully typed:

int add(int a, int b) => a + b;
final addOne = add.curried(1); // int Function(int)
fx([1, 2, 3]).map(addOne).toList(); // [2, 3, 4]

📖 WHY_CURRIED.md tells the full design story: why the direct port is impossible, how static extension resolution stands in for overloading, why the getter is named curried, and how the same port-the-meaning philosophy resolves the other unportable APIs.

⚠️ Those APIs keep @Deprecated stubs (curry, isUndefined, isArray, isObject, takeUntil) so migrating code gets analyzer guidance instead of silent breakage.

---

🧪 Testing

The FxTS spec suite has been ported alongside the library, and grown well past it: 1,800+ tests across 170 files, covering sync/async behavior, error propagation, laziness, typed errors, the events layer, and concurrency timing across every operator.

dart test

📈 Coverage is measured on every push and pull request and reported to Codecov. To reproduce locally:

dart run coverage:test_with_coverage   # writes coverage/lcov.info

---

📊 Benchmarks

Two suites back the comparison sites linked at the top: Dart vs FxDart (53 cases) and RxDart vs FxDart (41 cases). Every case is AOT-compiled (dart compile exe) and each side runs as a fresh process, interleaved, so thermal drift lands on both equally. ./benchmark.sh is the entry point.

Which command, when

1 · While developing — "did my change move this case?"

./benchmark.sh --ab ledger-diff               # against HEAD
./benchmark.sh --ab --ref v0.8.5 ledger-diff  # against a tag or commit

The one you will reach for most. It builds both variants of lib/ and runs them interleaved in one session, so drift hits both sides. The native side is the control: it links no fxdart code, so a library-only change must leave it identical — if it moved, the row is void.

It runs 20 rounds rather than ab_bench's default 12, because 12 is not enough. Four readings in the 0.8.6 pass looked like solid ±3-4% results against clean controls and every one of them was gone at 20.

2 · Before merging or releasing — "did anything regress?"

./benchmark.sh --ab --all

The same instrument across every case, as a gate: if a control drifts past its limit the run fails rather than printing a number. --all exists because without it every slug had to be typed by hand, which made "nothing regressed by 3%" a claim rather than a check. Give it an idle machine — it takes a while.

3 · Publishing — updating the numbers the site shows

./benchmark.sh --docs        # sweep, regenerate the report, rebuild docs/
./benchmark.sh --docs --rx   # the RxDart family

The only output fit to publish: native and fxdart are measured in the same session, so each row's ratio is sound. It writes benchmark/results/results.json (the bar charts), SUMMARY.md, and perf_ratio_report.md — every case ordered slowest to fastest. The RxDart family writes results-rx.json and SUMMARY-RX.md; its pages carry bars but no ranking table, so there is no report to regenerate there.

Skip --docs and the site keeps showing the old numbers. Skip the report regeneration — which is why this mode always does it for you — and results.json and the report drift apart silently, which has happened, and surfaced months later looking like a regression that had just landed.

⚠️ Reading the numbers

Do not compare two sweeps to judge a change. Cross-run noise is about 5%, and the proof is built in: the native side must be byte-identical across a library-only change, yet its measured cross-run delta is a median −2.1%, ranging −27% to +4%. Judge changes with 1, publish with 3.

--smoke is for "does this still run" — one un-warmed iteration, and the script restores results.json afterwards precisely so those numbers cannot leak into anything.

Two checks cost seconds and are worth running freely:

./benchmark.sh --verify   # is the ratio report in step with results.json?
./benchmark.sh --check    # do the cases still match their published examples?

Adding or changing a case? benchmark/AUTHORING.md has the rules — the first being that a case must measure the same pipeline its published example shows, which CI enforces.

---

🙏 Acknowledgments

Great thanks to Indong Yoo, CTO of Marpple, the creator of FxTS (and FxJS before it), whose functional programming model — lazy iteration with first-class, order-preserving concurrency — this library ports to Dart. All core ideas, operator semantics, and the original test suite come from the marpple/FxTS repository.

---

👤 Author

Bansook Nam

• 🌐 Website: https://github.com/bansooknam
• 🐙 Github: @bansooknam

🚢 Publishing the docs site

Normally there is nothing to do. docs/ is generated and untracked; .github/workflows/pages.yml builds and publishes it on every push to main. A lib/ change does not require rebuilding the bundle, re-stamping pages or committing anything.

The one part worth steering by hand is how many playground snippets get precompiled. Precompiling is what makes ▶ Run take ~200 ms instead of ~2.5 s, because the page ships the compiled JS instead of calling the DartPad compile service in the browser. It is also the slow part of the build — about a second per snippet, four at a time — so the default only covers the first playground on each page.

Trigger a publish by hand

gh workflow run pages.yml                  # republish at the default scope
gh workflow run pages.yml -f pg_scope=all  # …precompiling every snippet
gh workflow run pages.yml -f pg_scope=none # …skipping precompilation entirely
gh run watch                               # follow it

or Actions → pages → Run workflow in the browser.

pg_scope	Snippets precompiled	Cold build
first (default)	446 of 784 — the first playground on each page	~8 min
all	all 784	~13 min
none	none; every Run compiles over the network	seconds

Artifacts are content-addressed and cached per scope, so a repeat run only compiles what actually changed. A change to lib/ re-keys every snippet at once — that is inherent, since an artifact is keyed by the snippet plus the library it was compiled against, and it is what stops a stale artifact outliving the library.

A manual all run is temporary

A run publishes exactly the scope it was given. So pg_scope=all holds only until the next push to main, which republishes at first and drops the extra artifacts from the deployed site. Nothing breaks — those pages just fall back to the compile service and get slower on Run.

If you want all permanently, change the default in .github/workflows/pages.yml rather than re-running by hand:

env:
  PG_SCOPE: ${{ github.event.inputs.pg_scope || 'first' }}   # ← 'all'

Build it locally to look at it

None of these commit anything — docs/ is ignored.

./run.sh                  # build the site and serve it (-o opens a browser)
./run.sh -s               # serve what is already built, no rebuild
./deploy.sh               # build exactly what CI builds, then stop
dart run tool/precompile_playgrounds.dart --status   # coverage report, no network
dart run tool/rebuild_page.dart tutorials/map.html   # one page's artifacts, for a fast local preview

🤝 Contributing

Contributions, issues and feature requests are welcome! Feel free to check the issues page.

CONTRIBUTING.md has the working rules: branch naming and how to keep a long-running feature from becoming a merge event, what a PR description has to answer, and the gates a branch passes before review — dart analyze, a 100%-passing dart test, the playground-bundle check that catches wrapper drift, and the docs and translation checks — all of which CI also runs, so nothing silently depends on you having run them. Performance claims need a paired A/B, not a sweep; the ~5% noise floor and the instrument for seeing past it are documented there too.

📝 License

Copyright © 2023 Bansook Nam.

This project is MIT licensed.