Reactive State in Flutter with BehaviorSubject and StreamProvider

main.dart

// =========
// main.dart
// =========

import 'package:flutter/material.dart';
import 'package:flutter_riverpod/flutter_riverpod.dart';
import 'package:rxdart/subjects.dart';
import 'package:table_sticky_headers/table_sticky_headers.dart';

void main() => runApp(ProviderScope(child: MainPage()));

class MainPage extends ConsumerWidget {
  final ctl = MainPageController();

  @override
  Widget build(BuildContext context, ScopedReader watch) {
    final rows = watch(ctl.rows.provider);
    final columns = watch(ctl.columns.provider);
    return MaterialApp(
      title: 'Material App',
      home: Scaffold(
        appBar: AppBar(
          title: Text('Material App Bar'),
          actions: [
            FlatButton(onPressed: ctl.addRow, child: Text('add row')),
            FlatButton(onPressed: ctl.removeRow, child: Text('rem row')),
            FlatButton(onPressed: ctl.addColumn, child: Text('add col')),
            FlatButton(onPressed: ctl.removeColumn, child: Text('rem col')),
          ],
        ),
        body: Center(
          child: StickyHeadersTable(
            columnsLength: columns.when(
              data: (d) => d,
              loading: () => 0,
              error: (_, __) => 0,
            ),
            rowsLength: rows.when(
              data: (d) => d,
              loading: () => 0,
              error: (_, __) => 0,
            ),
            columnsTitleBuilder: (i) => Text('$i'),
            rowsTitleBuilder: (j) => Text('$j'),
            contentCellBuilder: (i, j) => Text('$i, $j'),
          ),
        ),
      ),
    );
  }
}

class MainPageController {
  final rows = 10.sbj;
  final columns = 10.sbj;

  void addRow() async => rows.emit(rows.state + 1);
  void addColumn() async => columns.emit(columns.state + 1);
  void removeRow() async => rows.emit(rows.state - 1);
  void removeColumn() async => columns.emit(columns.state - 1);
}

// ============
// subject.dart
// ============

class RxSubject<T> {
  final ctl = BehaviorSubject<T>();
  StreamProvider<T> provider;

  RxSubject([T seed]) {
    provider = StreamProvider<T>((ref) async* {
      ref.onDispose(ctl.close);
      await for (final value in ctl.stream) yield value;
    });
    if (seed != null) ctl.add(seed);
  }

  void emit(T value) => ctl.add(value);

  T get state => ctl.value;
}

extension Subject<T> on T {
  RxSubject<T> get sbj => RxSubject(this);
}