Dos feeds paginados, concatenados y deduplicados

Gana FxDart async

Requisito

Los eventos de log viven en dos almacenes paginados: uno primario y una réplica cuyas páginas se solapan con las del primario (algunos eventos se enviaron a ambos). Trae páginas de tres (llamadas simuladas, datos fijos en el código de abajo), lee primero por completo el primario y después la réplica, descarta los eventos ya vistos (por id) y para tras los primeros ocho eventos únicos. Informa de cuántas de las cinco páginas se llegaron a pedir realmente.

Para ser precisos sobre qué es el concat de FxDart: un añadido secuencial, no una mezcla — la réplica no se toca hasta que el primario se ha agotado. Aquí esa es la herramienta correcta, porque la tarea quiere que ganen los eventos del primario. Cada almacén se convierte en una secuencia asíncrona con range + flatMap (número de página → página de eventos), y uniqBy + take(8) rematan el trabajo. Como la cadena está basada en pull, que take se detenga detiene también la paginación: la última página de la réplica no se pide nunca.

Salida esperada

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

Lado a lado

Dart nativo

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');
}

Por qué difieren

La versión nativa son tres bucles anidados con un conjunto seen y un break outer; etiquetado — cada pieza (paginación, orden, deduplicación, salida temprana) tejida a mano dentro del flujo de control, y la salida temprana es la parte que mantiene el recuento de páginas en cuatro. Funciona, pero cada política vive en una cláusula de guarda en lugar de en un nombre. La cadena de FxDart le da a cada política su propia palabra — concat para la secuenciación, uniqBy para la deduplicación, take para el presupuesto — y la pereza que se salta la quinta página es el comportamiento por defecto del pipeline, no un salto cuidadosamente colocado.

Benchmark

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

Caso async: la escala principal es N = 100,000, no 1,000,000. Cada elemento cuesta una vuelta del event loop en ambos lados, así que un millón de awaits reales mediría el event loop de Dart durante minutos — no el pipeline. Los retardos son de longitud cero y se conserva el límite de concurrencia del ejemplo; lo que comparan las barras es la maquinaria del pipeline.

N = 100

Tiempo Empate

Dart nativo 92 µs

FxDart 101 µs

Memoria pico Empate

Dart nativo 16.4 MB

FxDart 16.6 MB

N = 10,000

Tiempo Gana nativo

Dart nativo 7.60 ms

FxDart 8.37 ms

Memoria pico Gana nativo

Dart nativo 43.0 MB

FxDart 47.6 MB

N = 100,000

Tiempo Gana nativo

Dart nativo 78.5 ms

FxDart 85.7 ms

Memoria pico Empate

Dart nativo 73.5 MB

FxDart 74.0 MB

Las barras son medianas de iteraciones cronometradas repetidas en procesos nuevos por lado (los N pequeños se agrupan por resolución del temporizador). Dos lados a menos del 5% entre sí — o a menos de 0.6 ms, una diferencia que nadie puede percibir — cuentan como empate; las carreras relativas ajustadas se vuelven a medir hasta 5 veces. En una app, cualquier cosa por debajo de unos pocos milisegundos es invisible para el usuario, gane la barra que gane. La memoria es el RSS pico del proceso. La VM de Dart y el dataset son idénticos en ambos lados, así que la diferencia entre las dos barras es lo que retiene el pipeline en sí.