A cursor's lifetime around a read

Toss-up async

Requirement

Read five ledger rows through a fake database cursor whose lifetime must bracket the read: created when reading starts, closed exactly once after the last row — and reading after close throws, so the bracket is load-bearing. Print the rows, then closed: true as the attestation. The cursor is in the code; both versions must print the lines shown under Expected output.

Expected output

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

Side by side

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

Why they differ

They mostly don't — both are ports of the same Rx idea, and FxDart says so: usingAsync came after Rx's using. The shape is the same three-part bracket: acquire, use, release. Rx.using creates the cursor when the stream is listened to and calls the disposer when the stream terminates; usingAsync acquires on the first pull and releases exactly once, after the terminal pull or right before an error propagates. In both, the resource's lifetime is tied to the consumption of the sequence, not to a scope in the caller — which is the whole point.

The edges show each model's temperament. The stream version's disposer also runs on unsubscription — cancel halfway and the cursor still closes, because a subscription is an object with a lifecycle of its own. The pull version has no subscription: release fires on completion or error, so a consumer that silently abandons the iterator would never trigger it — the honest idiom is to bound the pipeline (take, or a finite source like this one) so completion, and therefore release, is guaranteed. Here the read is finite and driven to the end, and both sides close the cursor exactly once, after row five.

A tie by design: this is the pair where the two libraries agree on the abstraction and differ only in what "the iteration ended" means in their model.

Benchmark

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

Async case: the headline scale is N = 10,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

RxDart 385 µs

FxDart 398 µs

Peak memory Tie

RxDart 16.6 MB

FxDart 16.5 MB

N = 10,000

Time Tie

RxDart 30.8 ms

FxDart 32.2 ms

Peak memory Tie

RxDart 24.9 MB

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