Why I Use RxDart Selectively Instead of Making Everything Reactive

Dart already has excellent streams. For a network status listener or a sequence of database updates, Stream and StreamController are often enough.

I use RxDart when the problem is not simply receiving values. The problem is timing or combining them:

• wait until a user pauses typing;
• ignore repeated connectivity changes;
• combine two live sources;
• keep only the latest request;
• give a new subscriber the most recent value.

Using RxDart for those cases keeps the intent short and testable. Using it everywhere can make ordinary async code harder to follow.

Problem one: search fires on every keystroke#

A search field can dispatch values quickly:

TextField(
  onChanged: (query) {
    context.read<SearchBloc>().add(
          SearchQueryChanged(query),
        );
  },
)

Without control, typing “flutter” can start seven requests. Most responses are useless before they arrive.

RxDart adds debounceTime to Dart streams:

Stream<String> searchQueries(Stream<String> input) {
  return input
      .map((query) => query.trim())
      .debounceTime(const Duration(milliseconds: 400))
      .distinct();
}

This does two things:

1. waits until no new value arrives for 400 milliseconds;
2. ignores a value equal to the previous query.

It does not cancel an older network request by itself. If search events run in a BLoC, pair the debounce with restartable event handling so stale results cannot win.

Problem two: connectivity is noisy#

Connectivity sources can emit several changes close together while a device moves between Wi-Fi and mobile data. Immediately showing and hiding an offline banner makes the interface flicker.

Stream<bool> stableConnection(Stream<bool> connection) {
  return connection
      .distinct()
      .debounceTime(const Duration(seconds: 2));
}

The stream emits only after the value has remained quiet for the selected period.

Be careful with the user experience. Delaying an “online” status may be fine; delaying a critical “offline” action by two seconds may not be. When the two directions need different behavior, model them separately instead of applying one operator chain blindly.

Problem three: the screen depends on two streams#

Suppose a page needs the signed-in user and their active subscription. Either one can change:

Stream<AccountViewData> accountViewData({
  required Stream<User?> user,
  required Stream<Subscription?> subscription,
}) {
  return Rx.combineLatest2(
    user,
    subscription,
    (currentUser, currentSubscription) {
      return AccountViewData(
        user: currentUser,
        subscription: currentSubscription,
      );
    },
  );
}

combineLatest2 emits after both sources have produced a value, then emits again whenever either changes.

This is useful when the values are truly independent sources. If subscription always belongs to the current user and must be fetched after the user changes, a dependent switchMap may express the relationship better.

Problem four: only the latest selection matters#

Imagine a category selector that loads products:

Stream<List<Product>> productsForSelection(
  Stream<String> categoryIds,
  ProductsRepository repository,
) {
  return categoryIds.switchMap(
    repository.watchProducts,
  );
}

When a new category arrives, switchMap unsubscribes from the previous inner stream and listens to the new one. Old category updates no longer reach the UI.

Use this for stream-producing operations. For Future-based BLoC handlers, bloc_concurrency restartable is often clearer because it keeps the policy beside the event registration.

Problem five: a new listener needs the current value#

A normal broadcast stream sends future events. A new subscriber does not automatically receive the previous value.

BehaviorSubject keeps the latest value:

class SessionStatusStore {
  final _status = BehaviorSubject<SessionStatus>.seeded(
    SessionStatus.unknown,
  );

  ValueStream<SessionStatus> get stream => _status.stream;
  SessionStatus get current => _status.value;

  void update(SessionStatus status) {
    _status.add(status);
  }

  Future<void> dispose() => _status.close();
}

This is useful for a small service that must expose both the current session status and future updates.

It also introduces ownership. The subject must be closed, errors must be considered, and unrestricted callers should not receive the subject itself. Expose a ValueStream and keep mutation private.

Before adding a BehaviorSubject, check whether a BLoC, ValueNotifier, or an existing repository stream already owns this state. Two sources of truth are worse than one missing convenience method.

Operators should explain the requirement#

These operators solve different timing rules:

Choose the operator from the sentence describing the requirement.

“Search after the user pauses” suggests debounce.

“Do not load another page more than once per second” suggests throttle.

“Show a view based on the latest user and subscription” suggests combineLatest.

Long chains are a warning sign. Extract and name parts of the transformation:

final cleanedQueries = input
    .map((value) => value.trim())
    .where((value) => value.length >= 2)
    .distinct();

return cleanedQueries
    .debounceTime(const Duration(milliseconds: 400))
    .switchMap(repository.watchSearch);

Errors and completion still matter#

Reactive operators do not remove normal stream responsibilities.

Decide:

• Should one failed request close the whole stream?
• Should an error become a state value?
• Who closes the subject?
• Does a retry resubscribe to the source?
• What happens when the owning BLoC or service is disposed?

For UI features, I often convert expected errors into explicit states before they reach the widget. A stream that terminates permanently after a temporary network failure can leave the screen stuck.

Test with time under your control#

Timing code needs timing tests:

test('emits only the final debounced query', () {
  fakeAsync((async) {
    final input = StreamController<String>();
    final values = <String>[];

    searchQueries(input.stream).listen(values.add);

    input
      ..add('flu')
      ..add('flut')
      ..add('flutter');

    async.elapse(const Duration(milliseconds: 399));
    expect(values, isEmpty);

    async.elapse(const Duration(milliseconds: 1));
    expect(values, ['flutter']);
  });
});

Also test stream errors, disposal, and what a late subscriber receives when using a subject.

When plain Dart is better#

I do not use RxDart for:

• one HTTP request;
• a straightforward async function;
• a single stream with no timing transformation;
• local widget state;
• code where a loop or await is clearer.

RxDart is most valuable at boundaries where streams become noisy, dependent, or time-sensitive. I keep those transformations small and named. The goal is to make the timing rule obvious, not to turn the whole application into an operator chain.