RxDart in a banking app: three problems it actually solved

For a normal Flutter app, plain Stream and StreamBuilder are enough, and I will happily tell you not to add dependencies you do not need. Banking apps are not normal apps. Balances update from several sources at once, the search box talks to a rate-limited backend, and a duplicated transfer is not a bug ticket, it is a very angry phone call.

That is the environment where RxDart stops being a fancy dependency and starts paying rent.

A quick orientation first. RxDart is the ReactiveX API implemented on top of native Dart streams. Nothing gets replaced underneath; everything is still a Stream. What you gain is a set of operators for filtering, merging, delaying, and transforming events, plus Subjects (BehaviorSubject, PublishSubject, ReplaySubject) in place of StreamController, with more control over how events are cached and handed to listeners.

I reach for it when input needs time-based filtering, when the UI depends on several independent streams at once, or when a new request should cancel the one already in flight. All three came up in the same app.

Beneficiary search: debounce and cancel

Users search for a transfer recipient by name or phone number. Call the API on every keystroke and several things go wrong. You spam your own backend and burn through rate limits, which mostly annoys the backend team. The sneakier failure is ordering: an older request can resolve after a newer one and paint stale results over fresh ones, and that one ships to production because it only shows up on slow networks.

debounceTime waits for typing silence, distinctUntilChanged skips repeat queries, and switchMap kills the in-flight request the moment a new query arrives.

import 'package:flutter/material.dart';
import 'package:rxdart/rxdart.dart';

class BeneficiarySearch extends StatefulWidget {
  @override
  _BeneficiarySearchState createState() => _BeneficiarySearchState();
}

class _BeneficiarySearchState extends State<BeneficiarySearch> {
  // BehaviorSubject remembers the latest value
  final _searchSubject = BehaviorSubject<String>();
  late Stream<List<String>> _searchResults;

  @override
  void initState() {
    super.initState();

    _searchResults = _searchSubject.stream
        // Wait for 500ms of typing silence
        .debounceTime(const Duration(milliseconds: 500))
        // Ignore if the text hasn't actually changed
        .distinctUntilChanged()
        // Cancel previous API call if a new one starts
        .switchMap((query) {
          if (query.isEmpty) return Stream.value([]);
          return Stream.fromFuture(_fetchBeneficiariesFromApi(query));
        });
  }

  Future<List<String>> _fetchBeneficiariesFromApi(String query) async {
    print('Fetching API for: $query');
    await Future.delayed(const Duration(seconds: 1)); // Simulate network
    return ['Alex ($query)', 'Anna ($query)'];
  }

  @override
  void dispose() {
    _searchSubject.close();
    super.dispose();
  }

  @override
  Widget build(BuildContext context) {
    return Column(
      children: [
        TextField(
          onChanged: _searchSubject.add, // Push new keystrokes to the stream
          decoration: const InputDecoration(labelText: 'Search phone or name...'),
        ),
        Expanded(
          child: StreamBuilder<List<String>>(
            stream: _searchResults,
            builder: (context, snapshot) {
              if (!snapshot.hasData) return const Text('Type to search...');
              return ListView(
                children: snapshot.data!.map((s) => ListTile(title: Text(s))).toList(),
              );
            },
          ),
        ),
      ],
    );
  }
}

switchMap is the piece people miss. Debouncing alone still lets two requests overlap; switching guarantees only the latest query can deliver results.

One total balance from three streams

The account overview shows a single number: checking plus savings plus credit card. Each of those comes from a different API or WebSocket and updates on its own schedule. The question is how to recompute the total every time any one of them moves.

Rx.combineLatest3 listens to all three at once. Whenever one stream emits, it grabs the latest value from each and recalculates.

import 'package:rxdart/rxdart.dart';

class BalanceBloc {
  // Mock streams representing real-time database connections
  final Stream<double> checkingStream = Stream.periodic(const Duration(seconds: 5), (_) => 2500.0);
  final Stream<double> savingsStream = Stream.periodic(const Duration(seconds: 2), (i) => 10000.0 + (i * 10));
  final Stream<double> creditStream = Stream.periodic(const Duration(seconds: 10), (_) => -500.0);

  // The merged stream
  Stream<double> get totalBalanceStream {
    return Rx.combineLatest3(
      checkingStream,
      savingsStream,
      creditStream,
      (double checking, double savings, double credit) {
        // This recalculates every time ANY of the 3 streams emit a new value
        return checking + savings + credit;
      },
    );
  }
}

Try writing this with three separate listeners and a shared mutable total, and you will understand why the operator exists.

Stopping duplicate transfers

The user is on the transfer confirmation screen. The network is slow, nothing visibly happens, so they tap "Transfer Money" five times. Executing five transfers and draining the account is the kind of mistake nobody lets you forget.

Two operators fit here. throttleTime ignores taps for a fixed duration, which works but the duration is a guess. exhaustMap is the better match for API calls: it drops every incoming event until the current request finishes, however long that takes.

import 'package:flutter/material.dart';
import 'package:rxdart/rxdart.dart';

class TransferButton extends StatefulWidget {
  @override
  _TransferButtonState createState() => _TransferButtonState();
}

class _TransferButtonState extends State<TransferButton> {
  final _transferActionSubject = PublishSubject<void>();

  @override
  void initState() {
    super.initState();

    _transferActionSubject.stream
        // exhaustMap ignores all incoming events until the Future completes
        .exhaustMap((_) => Stream.fromFuture(_executeTransfer()))
        .listen((result) {
          print('Transfer successful: $result');
        });
  }

  Future<String> _executeTransfer() async {
    print('Processing transfer... (ignoring other taps)');
    await Future.delayed(const Duration(seconds: 2)); // Simulate API processing
    return 'Transferred \$100';
  }

  @override
  void dispose() {
    _transferActionSubject.close();
    super.dispose();
  }

  @override
  Widget build(BuildContext context) {
    return ElevatedButton(
      // Push event to stream instead of directly calling the API
      onPressed: () => _transferActionSubject.add(null),
      child: const Text('Transfer Money'),
    );
  }
}

None of this makes RxDart the default answer. A visibility toggle wants a ValueNotifier, and app-wide state wants Riverpod or whatever your team has standardized on. But on screens where several live streams have to agree with each other, debounceTime, switchMap, combineLatest, and exhaustMap replaced the timers and boolean flags I used to hand-roll around StreamController, and I do not miss writing any of it.