Flutter Intermediate Lesson 60 of 135 3 min read

RxDart in Flutter: BehaviorSubject, Operators and Reactive Streams

Learn RxDart for Flutter with runnable examples: BehaviorSubject, debounceTime, combineLatest and switchMap with StreamBuilder and the BLoC pattern.

On this page

RxDart adds extra power to Dart streams: special subjects (streams you can push values into) and many operators that transform streams (debounceTime, combineLatest, switchMap and more). It is not a full state management solution; it is a tool used with StreamBuilder or the Bloc pattern.

If streams are new to you, read StreamController + StreamBuilder first.

Install

flutter pub add rxdart

Subjects

Subject	What a new listener gets
`BehaviorSubject`	The latest value immediately, then new ones
`PublishSubject`	Only new values after it starts listening
`ReplaySubject`	All previous values, then new ones

BehaviorSubject is the one you use for state, because it always has a current value.

The counter

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

class CounterBloc {
  final _count = BehaviorSubject<int>.seeded(0);

  Stream<int> get count => _count.stream;
  void increment() => _count.add(_count.value + 1);
  void dispose() => _count.close();
}

final bloc = CounterBloc();

void main() => runApp(const MaterialApp(home: CounterPage()));

class CounterPage extends StatelessWidget {
  const CounterPage({super.key});

  @override
  Widget build(BuildContext context) {
    return Scaffold(
      appBar: AppBar(title: const Text('RxDart')),
      body: Center(
        child: StreamBuilder<int>(
          stream: bloc.count,
          builder: (context, snapshot) =>
              Text('${snapshot.data ?? 0}', style: const TextStyle(fontSize: 48)),
        ),
      ),
      floatingActionButton: FloatingActionButton(
        onPressed: bloc.increment,
        child: const Icon(Icons.add),
      ),
    );
  }
}

debounceTime waits until the user stops typing before searching. switchMap cancels the old search when a new one starts.

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

const allCities = ['Kathmandu', 'Pokhara', 'Delhi', 'Mumbai', 'Toronto', 'Sydney', 'Lisbon', 'London'];

Future<List<String>> searchApi(String q) async {
  await Future.delayed(const Duration(milliseconds: 300)); // pretend network delay
  return allCities.where((c) => c.toLowerCase().contains(q.toLowerCase())).toList();
}

class SearchBloc {
  final _query = BehaviorSubject<String>.seeded('');

  late final Stream<List<String>> results = _query
      .debounceTime(const Duration(milliseconds: 400))
      .distinct()
      .switchMap((q) => Stream.fromFuture(searchApi(q)));

  void search(String q) => _query.add(q);
  void dispose() => _query.close();
}

final searchBloc = SearchBloc();

void main() => runApp(const MaterialApp(home: SearchPage()));

class SearchPage extends StatelessWidget {
  const SearchPage({super.key});

  @override
  Widget build(BuildContext context) {
    return Scaffold(
      appBar: AppBar(title: const Text('Debounced search')),
      body: Column(
        children: [
          Padding(
            padding: const EdgeInsets.all(12),
            child: TextField(
              onChanged: searchBloc.search,
              decoration: const InputDecoration(prefixIcon: Icon(Icons.search), hintText: 'Search cities'),
            ),
          ),
          Expanded(
            child: StreamBuilder<List<String>>(
              stream: searchBloc.results,
              builder: (context, snapshot) {
                if (!snapshot.hasData) return const Center(child: CircularProgressIndicator());
                final cities = snapshot.data!;
                if (cities.isEmpty) return const Center(child: Text('No matches'));
                return ListView(children: [for (final c in cities) ListTile(title: Text(c))]);
              },
            ),
          ),
        ],
      ),
    );
  }
}

Combining streams

Rx.combineLatest2 produces a value whenever either input changes, which is perfect for form validation:

final email = BehaviorSubject<String>.seeded('');
final password = BehaviorSubject<String>.seeded('');

final canSubmit = Rx.combineLatest2<String, String, bool>(
  email,
  password,
  (e, p) => e.contains('@') && p.length >= 6,
);
// StreamBuilder<bool>(stream: canSubmit, ...) enables the button.

Operators you will use most

Operator	Does
`map`, `where`	Transform and filter (built into Dart)
`debounceTime`	Wait for a pause
`throttleTime`	At most one value per period
`distinct`	Skip repeated values
`switchMap`	Start a new inner stream, cancel the old one
`combineLatest`	Combine the latest values of several streams
`startWith`	Emit a first value immediately

Try it yourself

Add a “minimum 2 letters” rule to the search with .where((q) => q.length >= 2 || q.isEmpty), and show “Type at least 2 letters” when it is too short.