La vida de un cursor alrededor de una lectura

Empate async

Requisito

Lee cinco filas del libro de cuentas a través de un cursor de base de datos falso cuya vida debe abrazar la lectura: creado cuando la lectura empieza, cerrado exactamente una vez después de la última fila — y leer tras el cierre lanza, así que el paréntesis soporta carga real. Imprime las filas y luego closed: true como atestación. El cursor está en el código; las dos versiones deben imprimir las líneas que aparecen bajo Salida esperada.

Salida esperada

2026-08-01  balance 120.00
2026-08-02  balance 84.50
2026-08-03  balance 210.25
2026-08-04  balance 45.00
2026-08-05  balance 99.90
closed: true

Lado a lado

RxDart

import 'package:rxdart/rxdart.dart';

/// A fake database cursor: sequential reads plus a [closed] flag the
/// output can attest to. Reading after close throws.
class LedgerCursor {
  static const _rows = [
    '2026-08-01  balance 120.00',
    '2026-08-02  balance 84.50',
    '2026-08-03  balance 210.25',
    '2026-08-04  balance 45.00',
    '2026-08-05  balance 99.90',
  ];

var closed = false;

int get length => _rows.length;

Future<String> read(int i) async {
    if (closed) throw StateError('read after close');
    await Future<void>.delayed(const Duration(milliseconds: 5));
    return _rows[i];
  }

void close() => closed = true;
}

Future<void> main() async {
  late final LedgerCursor cursor;

// The bracket: the resource is created on listen, read as a stream,
  // and disposed when the stream terminates — however it terminates.
  final rows = await Rx.using<String, LedgerCursor>(
    resourceFactory: () => cursor = LedgerCursor(),
    streamFactory: (c) => Rx.range(0, c.length - 1).asyncMap(c.read),
    disposer: (c) => c.close(),
  ).toList();

rows.forEach(print);
  print('closed: ${cursor.closed}');
}

FxDart

import 'package:fxdart/fxdart.dart';

var closed = false;

int get length => _rows.length;

Future<String> read(int i) async {
    if (closed) throw StateError('read after close');
    await Future<void>.delayed(const Duration(milliseconds: 5));
    return _rows[i];
  }

void close() => closed = true;
}

Future<void> main() async {
  late final LedgerCursor cursor;

// The bracket: the resource is acquired on the first pull, read lazily,
  // and released exactly once — after the last row, or right before an
  // error would propagate.
  final rows = await fxAsync(usingAsync(
    () => cursor = LedgerCursor(),
    (c) => toAsync(Iterable.generate(c.length, c.read)),
    (c) => c.close(),
  )).toList();

rows.forEach(print);
  print('closed: ${cursor.closed}');
}

Por qué difieren

En su mayor parte no difieren — ambos son ports de la misma idea de Rx, y FxDart lo dice: usingAsync llegó después del using de Rx. La forma es el mismo paréntesis de tres partes: adquirir, usar, liberar. Rx.using crea el cursor cuando el stream recibe un listener y llama al liberador cuando el stream termina; usingAsync adquiere en el primer pull y libera exactamente una vez, tras el pull terminal o justo antes de que un error se propague. En ambos, la vida del recurso está atada al consumo de la secuencia, no a un scope en el llamador — que es justamente el sentido de todo esto.

Los bordes muestran el temperamento de cada modelo. El liberador de la versión de stream también corre al cancelar la suscripción — cancela a mitad de camino y el cursor igualmente se cierra, porque una suscripción es un objeto con ciclo de vida propio. La versión pull no tiene suscripción: la liberación se dispara al completar o al errar, así que un consumidor que abandona el iterador en silencio nunca la dispararía — el idioma honesto es acotar el pipeline (take, o una fuente finita como esta) para que la terminación, y por tanto la liberación, esté garantizada. Aquí la lectura es finita y se lleva hasta el final, y ambos lados cierran el cursor exactamente una vez, después de la fila cinco.

Un empate por diseño: este es el par donde las dos bibliotecas coinciden en la abstracción y solo difieren en qué significa «la iteración terminó» en su modelo.

Benchmark

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

Caso async: la escala principal es N = 10,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

RxDart 385 µs

FxDart 398 µs

Memoria pico Empate

RxDart 16.6 MB

FxDart 16.5 MB

N = 10,000

Tiempo Empate

RxDart 30.8 ms

FxDart 32.2 ms

Memoria pico Empate

RxDart 24.9 MB

FxDart 25.4 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í.