Two paged feeds, concatenated and deduped

FxDart wins async

Requirement

Log events live in two paged stores — a primary and a replica whose pages overlap it (some events shipped to both). Fetch pages of three (simulated calls, fixed data in the code below), read the primary fully first, then the replica, drop events already seen (by id), and stop after the first eight unique events. Report how many of the five pages were actually fetched.

To be precise about what FxDart's concat is: a sequential append, not a merge — the replica is not touched until the primary is exhausted. That is the right tool here, because the task wants primary events to win. Each store becomes an async sequence with range + flatMap (page number → page of events), and uniqBy + take(8) finish the job. Because the chain is pull-based, take stopping also stops the paging: the last replica page is never fetched.

Expected output

first 8 unique events (primary first, then replica):
  e1  boot
  e2  login user 7
  e3  cache miss
  e4  queue drained
  e5  login user 12
  e6  gc pause 18ms
  e7  disk 81% full
  e8  cert renewed
pages fetched: 4 of 5

Side by side

Native Dart

class Event {
  final String id;
  final String msg;
  const Event(this.id, this.msg);
}

// Two log stores, paged 3 per call. The replica overlaps the primary:
// some events were shipped to both.
const primary = [
  [Event('e1', 'boot'), Event('e2', 'login user 7'), Event('e3', 'cache miss')],
  [Event('e4', 'queue drained'), Event('e5', 'login user 12'), Event('e6', 'gc pause 18ms')],
];
const replica = [
  [Event('e5', 'login user 12'), Event('e6', 'gc pause 18ms'), Event('e7', 'disk 81% full')],
  [Event('e8', 'cert renewed'), Event('e9', 'login user 3'), Event('e10', 'backup done')],
  [Event('e11', 'shutdown'), Event('e12', 'boot'), Event('e13', 'login user 9')],
];

int pagesFetched = 0;

Future<List<Event>> fetchPage(List<List<Event>> store, int page) async {
  pagesFetched++;
  await Future.delayed(const Duration(milliseconds: 10));
  return store[page];
}

Future<void> main() async {
  final events = <Event>[];
  final seen = <String>{};
  outer:
  for (final store in [primary, replica]) {
    for (var page = 0; page < store.length; page++) {
      for (final e in await fetchPage(store, page)) {
        if (!seen.add(e.id)) continue;
        events.add(e);
        if (events.length == 8) break outer;
      }
    }
  }
  print('first 8 unique events (primary first, then replica):');
  for (final e in events) {
    print('  ${e.id}  ${e.msg}');
  }
  print('pages fetched: $pagesFetched of 5');
}

FxDart

import 'package:fxdart/fxdart.dart';

class Event {
  final String id;
  final String msg;
  const Event(this.id, this.msg);
}

int pagesFetched = 0;

Future<List<Event>> fetchPage(List<List<Event>> store, int page) async {
  pagesFetched++;
  await Future.delayed(const Duration(milliseconds: 10));
  return store[page];
}

FxAsync<Event> drain(List<List<Event>> store) =>
    fx(range(0, store.length)).toAsync().flatMap((p) => fetchPage(store, p));

Future<void> main() async {
  final events = await drain(primary)
      .concat(drain(replica))
      .uniqBy((e) => e.id)
      .take(8)
      .toList();
  print('first 8 unique events (primary first, then replica):');
  for (final e in events) {
    print('  ${e.id}  ${e.msg}');
  }
  print('pages fetched: $pagesFetched of 5');
}

Why they differ

The native version is three nested loops with a seen set and a labeled break outer; — every piece (pagination, ordering, dedupe, early exit) hand-woven into control flow, and the early exit is the part that keeps the page count at four. It works, but each policy lives in a guard clause rather than a name. The FxDart chain gives every policy its own word — concat for sequencing, uniqBy for dedupe, take for the budget — and the laziness that skips the fifth page is the pipeline's default behavior, not a carefully placed jump.

Benchmark

Apple M1 Max, 32 GB RAM · Dart 3.12.2 (AOT-compiled) · 2026-08-24

Async case: the headline scale is N = 100,000, not 1,000,000. Every element costs an event-loop turn on both sides, so a million real awaits would measure Dart's event loop for minutes — not the pipeline. Delays are zero-length and the example's concurrency limit is kept; what the bars compare is the pipeline machinery.

N = 100

Time Tie

Native Dart 92 µs

FxDart 101 µs

Peak memory Tie

Native Dart 16.4 MB

FxDart 16.6 MB

N = 10,000

Time Native wins

Native Dart 7.60 ms

FxDart 8.37 ms

Peak memory Native wins

Native Dart 43.0 MB

FxDart 47.6 MB

N = 100,000

Time Native wins

Native Dart 78.5 ms

FxDart 85.7 ms

Peak memory Tie

Native Dart 73.5 MB

FxDart 74.0 MB

Bars are medians of repeated timed iterations in fresh processes per side (small N is batched for timer resolution). Sides within 5% of each other — or within 0.6 ms, a difference no person can perceive — count as a tie; close relative races are re-measured up to 5 runs. In an app, anything under a few milliseconds is invisible to the user regardless of which bar is shorter. Memory is peak process RSS. The Dart VM and the dataset are identical on both sides, so the difference between the two bars is what the pipeline itself holds onto.