RxDart Documentation

When using async* generators, avoid awaiting a never-completing Future immediately after a yield if downstream code relies on the stream's cancellation settling (e.g., using Stream.first, Stream.single, or await for with a break).

If a generator enters an await that never completes, it cannot be cancelled. The cancellationFuture will never complete because the generator body never returns or throws, causing the calling code to hang indefinitely. This is a property of the Dart async* runtime, not a bug in RxDart.

Best Practices:

• Ensure async* generators can eventually return or throw an error so cancellation can complete.
• For long-lived or open-ended sources, use a StreamController-based implementation with explicit onCancel cleanup instead of parking an async* generator on an infinite await. This makes cancellation semantics explicit and ensures the cancellation future settles predictably.

The rxdart_flutter widgets will display error widgets if the provided stream violates certain conditions. To prevent errors, ensure your stream meets these requirements:

1. Always has an initial value: Use BehaviorSubject.seeded(value) instead of a plain BehaviorSubject(). If a stream has no initial value, it will throw ValueStreamHasNoValueError.
2. Never emits errors: Ensure errors are handled before they reach the widget (e.g., using stream.handleError()). If a stream emits an error, it will throw UnhandledStreamError.

Correct Initialization Examples:

// Good - stream has initial value
final goodStream = BehaviorSubject<int>.seeded(0);

// Bad - stream has no initial value (throws ValueStreamHasNoValueError)
final badStream = BehaviorSubject<int>();

// Bad - stream with error (throws UnhandledStreamError)
final errorStream = BehaviorSubject<int>.seeded(0)..addError(Exception());

A known issue exists when composing async* generators that contain infinite await calls (e.g., await Completer<void>().future) using operators like Rx.combineLatest2 and switchMap.

When you call await stream.first, the stream attempts to cancel its subscriptions once the first value is received. However, if one of the source streams is an async* generator suspended at an await (rather than a yield), Dart's runtime does not resume the generator upon cancellation. This causes the generator's cancellationFuture to remain incomplete. Because Rx.combineLatest waits for all source subscriptions to cancel via Future.wait, the entire cancellation chain hangs, and await stream.first will never complete (or will time out).

To avoid this:

• Ensure async* generators do not stay suspended at an infinite await if they are part of a combined stream that needs to be canceled.
• If using long-lived watchers (like Drift watchers) wrapped in async*, ensure they can be properly closed or that the composition doesn't rely on Stream.first which triggers a full subscription cancellation.

import 'dart:async';
import 'dart:io';
import 'package:rxdart/rxdart.dart';

Future<void> main() async {
  final stream = Rx.combineLatest2(
    _emitOnceAndNeverClose('left'),
    Stream.value('right').switchMap(_emitOnceAndNeverClose),
    (left, right) => '$left|$right',
  );

  // This will hang or timeout if the generators are stuck at an await
  final value = await stream.first.timeout(const Duration(milliseconds: 200));
  stdout.writeln(value);
}

Stream<T> _emitOnceAndNeverClose<T>(T value) async* {
  yield value;
  await Completer<void>().future; // This infinite await causes the hang on cancellation
}