Stamp each request with the latest config

Toss-up async

Requirement

An app fires four API requests while, in the background, deploys bump the config version v1 → v2 → v3 at fixed offsets. Each request must be stamped with the config version that was current when the request fired — and a config bump on its own must not emit anything. Print the four stamped requests. Both schedules are simulated in the code; both versions must print the lines shown under Expected output.

Expected output

GET /orders (config v1)
GET /stock (config v2)
GET /prices (config v2)
GET /refunds (config v3)

Side by side

RxDart

import 'dart:async';

import 'package:rxdart/rxdart.dart';

/// Simulated deploys: the config version bumps between requests.
Stream<String> configs() {
  final c = StreamController<String>();
  const events = [(0, 'v1'), (200, 'v2'), (500, 'v3')];
  for (final (ms, v) in events) {
    Timer(Duration(milliseconds: ms), () => c.add(v));
  }
  Timer(const Duration(milliseconds: 700), c.close);
  return c.stream;
}

/// Simulated API calls, fired at fixed offsets.
Stream<String> requests() {
  final c = StreamController<String>();
  const events = [
    (80, 'GET /orders'),
    (300, 'GET /stock'),
    (400, 'GET /prices'),
    (600, 'GET /refunds'),
  ];
  for (final (ms, r) in events) {
    Timer(Duration(milliseconds: ms), () => c.add(r));
  }
  Timer(const Duration(milliseconds: 700), c.close);
  return c.stream;
}

Future<void> main() async {
  final stamped = await requests()
      .withLatestFrom(configs(), (String req, String cfg) => '$req (config $cfg)')
      .toList();

stamped.forEach(print);
}

FxDart

import 'dart:async';

import 'package:fxdart/fxdart.dart';

Future<void> main() async {
  final stamped = await fxEvents(requests())
      .withLatestFrom(configs(), (req, cfg) => '$req (config $cfg)')
      .toList();

stamped.forEach(print);
}

Why they differ

They don't. This is combineLatest's asymmetric sibling — one stream drives, the other is only consulted — and both panels name it the same way: withLatestFrom emits per request, stamped with the freshest config seen so far, and stays silent when only the config changes. No tagged-merge scaffolding, no scan fold — the two chains are operator-for-operator identical.

fxdart's events layer absorbs the Rx approach for the push side: fxEvents is a thin wrapper chain over plain Streams — never an extension, so it collides with nothing, rxdart included. RxDart's operator catalog remains far larger; fxdart keeps the events core small and crosses into the typed pull pipeline with .pull() when per-value processing grows. For stamping one live stream with the latest value of another, the two sides are equivalent: a tie.

Benchmark

No benchmark for this example. This operator is defined by elapsed time, not by work done: both sides wait out the same windows, so a bar would be measuring Dart's timers rather than either library. Shrinking the windows does not help either — how many values a burst produces would then depend on how fast the machine is, and the two sides could no longer be checked against an identical result. The examples built on timing operators are left unmeasured for that reason; the ones whose delays merely stand in for I/O are benchmarked with those delays set to zero.