Un precio, o el precio de lista

Gana FxDart async

Requisito

Cotiza dos órdenes de compra contra el libro de precios promocionales de agosto. La búsqueda asíncrona lanza para los SKU sin precio promocional — esas líneas deben caer a su precio de lista, y cada línea debe aparecer en la cotización. Una orden sin líneas se cotiza como la única línea (no lines to quote). Los datos están en el código; las dos versiones deben imprimir las líneas que aparecen bajo Salida esperada.

Salida esperada

PO-3391:
  SKU-201: 11.50 (promo)
  SKU-333: 9.50 (list)
  SKU-204: 8.25 (promo)
  SKU-207: 19.90 (promo)
  SKU-410: 5.20 (list)
PO-3402:
  (no lines to quote)

Lado a lado

RxDart

import 'package:rxdart/rxdart.dart';

/// August promo price book — looking up any other SKU throws.
const promoPrices = {'SKU-201': 11.50, 'SKU-204': 8.25, 'SKU-207': 19.90};

/// Purchase orders to quote: (order id, lines of sku + list price).
const orders = [
  (
    'PO-3391',
    [
      (sku: 'SKU-201', list: 14.00),
      (sku: 'SKU-333', list: 9.50),
      (sku: 'SKU-204', list: 9.75),
      (sku: 'SKU-207', list: 24.00),
      (sku: 'SKU-410', list: 5.20),
    ]
  ),
  ('PO-3402', <({String sku, double list})>[]),
];

Future<double> promoPrice(String sku) async {
  await Future<void>.delayed(const Duration(milliseconds: 5));
  final price = promoPrices[sku];
  if (price == null) throw StateError('no promo price for $sku');
  return price;
}

Future<void> main() async {
  for (final (id, lines) in orders) {
    print('$id:');
    final quotes = await Stream.fromIterable(lines)
        // One error channel for the whole stream: by the time an
        // onErrorReturnWith after asyncMap saw the failure, the line that
        // caused it would be gone. Recovery that still knows its line
        // means wrapping each lookup in its own inner stream.
        .flatMap((line) => Rx.fromCallable(() => promoPrice(line.sku))
            .map((p) => '  ${line.sku}: ${p.toStringAsFixed(2)} (promo)')
            .onErrorReturnWith((_, _) =>
                '  ${line.sku}: ${line.list.toStringAsFixed(2)} (list)'))
        .defaultIfEmpty('  (no lines to quote)')
        .toList();
    quotes.forEach(print);
  }
}

FxDart

import 'package:fxdart/fxdart.dart';

/// August promo price book — looking up any other SKU throws.
const promoPrices = {'SKU-201': 11.50, 'SKU-204': 8.25, 'SKU-207': 19.90};

Future<void> main() async {
  for (final (id, lines) in orders) {
    print('$id:');
    final quotes = await fx(lines)
        .toAsync()
        // Recovery is plain control flow inside the callback: catch the
        // typed error right beside the call, return the fallback value.
        .map((line) async {
          try {
            final p = await promoPrice(line.sku);
            return '  ${line.sku}: ${p.toStringAsFixed(2)} (promo)';
          } on StateError {
            return '  ${line.sku}: ${line.list.toStringAsFixed(2)} (list)';
          }
        })
        .defaultIfEmpty('  (no lines to quote)')
        .toList();
    quotes.forEach(print);
  }
}

Por qué difieren

Un stream lleva sus valores y sus errores por canales separados, y el canal de errores pertenece al pipeline entero. Para cuando un onErrorReturnWith colocado tras un asyncMap viera el fallo, la línea que lo causó ya habría desaparecido — un evento de error lleva el error, no el elemento. La recuperación idiomática en RxDart es darle a cada búsqueda su propio stream interno — flatMap hacia Rx.fromCallable — de modo que cada fallo sea terminal solo para su propio stream de un elemento, donde la línea sigue a mano para el fallback.

En el modelo pull no hay un segundo canal por el que caer. La búsqueda es un await dentro del callback de map, así que la recuperación es control de flujo ordinario de Dart: atrapa el StateError tipado justo al lado de la llamada y devuelve el precio de lista — el elemento, su fallback y el manejo del error viven todos en las mismas cuatro líneas. Sin envolver, sin volver a mezclar, y el fallo no llega a tocar el pipeline en ningún momento.

La segunda orden, vacía, aterriza igual en ambos lados — defaultIfEmpty, un operador que FxDart tomó de Rx. El veredicto se lo lleva FxDart por el plato principal: la recuperación de errores por elemento es una frase en el modelo pull y una construcción en el modelo push.

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 455 µs

FxDart 453 µs

Memoria pico Empate

RxDart 17.1 MB

FxDart 17.3 MB

N = 10,000

Tiempo Gana FxDart

RxDart 42.7 ms

FxDart 38.9 ms

Memoria pico Gana RxDart

RxDart 49.1 MB

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