How to maintain the rxdart broadcast stream listner when using bloc with stateless widget and navigating between screens?

Question

I am learning flutter and flutter_bloc:8.1.6. I am developing a simple app where a user can add or edit their transactions and see a list of transactions when a new transaction is added or updated.

Under the hood, my repository exposes a stream of data using rxDart Behaviour subject like below:

class TransactionRepository {
  final LocalDB _database;
  final BehaviorSubject<List<Transaction>> _transactionsController =
      BehaviorSubject.seeded([]);

  TransactionRepository({required LocalDB database}) : _database = database {
    _init();
  }

  Future<void> _init() async {
    try {
      final transactions = await _getAllTransactions();
      _transactionsController.add(transactions);
    } catch (e) {
      rethrow;
    }
  }

  Stream<List<Transaction>> streamTransactions() {
    return _transactionsController.asBroadcastStream();
  }

  /// Save a transaction
  Future<void> createTransaction(NewTransaction transaction) async {
    try {
      await _database
            .into(_database.transactionTable)
            .insert(TransactionTableCompanion(
              //... fields
            ));

      final transactions = await _getAllTransactions();
      _transactionsController.add(transactions);
    } catch (e) {
      rethrow;
    }
  }
}

With bloc, I am absorbing this stream like below:

  ...
class TransactionsBloc extends Bloc<TransactionsEvent, TransactionsState> {
  final TransactionRepository _transactionRepository;

  TransactionsBloc({required TransactionRepository transactionRepository})
      : _transactionRepository = transactionRepository,
        super(const TransactionsState(
            transactionStatus: TransactionsStatus.initial)) {
    on<TransactionsEvent>(
      (event, emit) => event.map(
        loadAll: (event) => _loadTransactions(event, emit),
        error: (event) => _errorLoadingTransactions(event, emit),
      ),
      transformer: null,
    );
  }

  Future<void> _loadTransactions(
      TransactionsEvent event, Emitter<TransactionsState> emit) async {
    try {
      emit(state.copyWith(transactionStatus: TransactionsStatus.fetching));

      await emit.forEach(
        _transactionRepository.streamTransactions(),
        onData: (transactions) {
          if (transactions.isEmpty) {
            return state.copyWith(
              transactionStatus: TransactionsStatus.initial,
              transactions: [],
            );
          }

          return state.copyWith(
            transactionStatus: TransactionsStatus.fetchedSuccessfully,
            transactions: transactions,
          );
        },
        onError: (e, s) {
          return state.copyWith(
            transactionStatus: TransactionsStatus.fetchingFailed,
            message: 'Error loading transaction(s)',
          );
        },
      );
    } catch (e) {
      add(const TransactionsEvent.error('Error loading transaction(s)'));
    }
  }
}
...

And this bloc is hooked to my main.dart StatelessWidget like:

...
return CupertinoApp(
      title: appTitle,
      home: SafeArea(
        child: RepositoryProvider(
          create: (BuildContext context) {
            final database = context.read<DatabaseCubit>();
            return TransactionRepository(
              database: database.state,
            );
          },
          child: BlocProvider(
            create: (BuildContext context) => TransactionsBloc(
              transactionRepository: context.read<TransactionRepository>(),
            )..add(const TransactionsEvent.loadAll()),  // stream is subscribed here
            child: const HomeScreen(title: appTitle),
          ),
        ),
      ),
    );
...

My HomeScreen StatelessWidget looks like:

@override
  Widget build(BuildContext context) {
    return CupertinoPageScaffold(
      navigationBar: CupertinoNavigationBar(
        leading: const Icon(CupertinoIcons.line_horizontal_3),
        trailing: IconButton(
          onPressed: () {
            Navigator.push(
              context,
              CupertinoPageRoute(
                builder: (BuildContext context) => const AddNewTransaction(),
              ),
            );
          },
          icon: const Icon(CupertinoIcons.add),
        ),
      ),
      child: const TransactionsList(),
    );
  }

TransactionList StatelessWidget where the list is finally being displayed:

BlocBuilder<TransactionsBloc, TransactionsState>(
        builder: (BuildContext context, TransactionsState state) {
          switch (state.transactionStatus) {
            case TransactionsStatus.initial:
              return TransactionListItem(
                  transaction: Transaction.nullTransaction);
            case TransactionsStatus.fetching:
              return const Center(
                child: CupertinoActivityIndicator(),
              );
            case TransactionsStatus.fetchedSuccessfully:
              return GroupedTransactionsListView(
                transactions:
                    state.transactions ?? [Transaction.nullTransaction],
                // transactions: state.transactions ?? [],
              );
            case TransactionsStatus.fetchingFailed:
              return Center(
                child: Text(state.message ?? 'Error loading transaction(s)'),
              );
            default:
              return const Center(
                child: Text('Something went wrong! :('),
              );
          }
        },
      ),

What is happening now is that when the app starts, the stream gets subscribed successfully and I can see all the data in my transaction list.

The user navigates to AddTransaction page (see HomeScreen -> navigation) and adds a new transaction. When they click "Save", a new entry is added to the database and upon save success, the user is navigated back (using Navigation.pop).

Now, when the user comes back to this TransactionList page, the list is not updated.

I debugged the flow and to my understanding, the stream listner (_loadTransactions) is closed.

So I want to understand and learn if and how can I maintain the stream from getting closed?

From what I have gathered so far, one way is to switch to StatefulWidget, and that's alright but I want to explore new ideas around StatelessWidget first.

If that's not possible, what is the recommended way to trigger re-subscribing the stream when a user navigated back to TransactionList page?

Thanks in advance!

Edit: Working prototype of above: https://github.com/pkimtani/oma-dhan/tree/feature/new-transaction

Answer 1

So, method _loadTransactions(event, emit) is triggered in bloc on TransactionsEvent. In the main.dart you do:

... add(const TransactionsEvent.loadAll()).

And your bloc handling this event. But then, when you do the transaction, in the repo you put event to the stream(not to the bloc) and this is correct. But your bloc is not listening stream in the repo all time. It trying to listen repo stream after new TransactionsEvent only. As there are no new TransactionsEvent, method _loadTransactions not called and your UI is not updated.

What you could do here:

class TransactionsBloc extends Bloc< TransactionsEvent, TransactionsState> {
  final TransactionRepository _repository
  final StreamSubscription _subscription;

      Bloc1(TransactionRepository repository) : 
        _repository = repository,
        super(TransactionsState()) {
        ...
        _subscription = repository.streamTransactions().listen(
          onData: (transactions) {
             if (transactions.isEmpty) {
               return state.copyWith(
                   transactionStatus: TransactionsStatus.initial,
                   transactions: [],
               );
             }

             return state.copyWith(
               transactionStatus: TransactionsStatus.fetchedSuccessfully,
               transactions: transactions,
             );
          },
          onError: (e, s) {
            return state.copyWith(
              transactionStatus: TransactionsStatus.fetchingFailed,
              message: 'Error loading transaction(s)',
            );
          },
       );
      }

      @override
      Future<void> close() {
        _subscription?.cancel();
        return super.close();
      }
    }
...

You should listen stream from repo in bloc and react by emiting state on every event from stream.