Flutter - State management with RxDart Streams

What is State Management and Why do we need it? In Flutter we have widgets that define our screens. A widget is a Flutter component that can be built out of 0, 1 or multiple widgets which all together can become a screen. These screens display some information to us and most of the time we need this information to change. For example, if we have an application that displays a list of universities we need a screen with a list of widgets where each widget displays some information for a specific university. This screen should also display something when it loads the data (e.g. a loading spinner) and it also has to show something when it failed to load the data. We need to find a way to manage all 3 phases (Loading, Success, Error) within the app life-cycle in an easy-to-test and easy-to-use way. These 3 phases, the moment when the screen is loading the data, the moment when the data is successfully loaded and the moment when there was something wrong and an error is displayed, build the screen state. What here is using Dart streams and library. we will achieve to manage the screen state RxDart codebase for the examples within this article What is RxDart? stands for and it comes from reactive programming. In Dart, comes with a bunch of extensions over and and they introduce a lot of other specific Rx components such as , , , etc. Rx library is based on functional programming style, meaning we can have a chain of transformation functions applied to a stream: Rx ReactiveX RxDart Streams StreamControllers BehaviourSubject ReplaySubjects MergeStreams CombineStreams return _universityEndpoint .getUniversitiesByCountry(country) .safeApiConvert((p0) => p0.map((e) => e.toDomain()).toList()) .map((event) => transformToSomethingElse(event)) .flatMap((data) => tranformIntoAStream(data)) .mergeWith([stream2, stream3]); “Functional programming consists in building software by composing pure functions while avoiding shared state and mutable data. Reactive programming is an asynchronous programming paradigm concerned with data streams and the propagation of change. Together, functional programming and reactive programming form a combination of functional and reactive techniques that can represent an elegant approach to event-driven programming – with values that change over time and where the consumer reacts to the data as it comes in. This technology brings together different implementations of its core principles, some authors came up with a document that defines the common vocabulary for describing the new type of applications.” - https://www.baeldung.com/rx-java Since uses Streams it follows the Observer pattern. What you when working with streams is that to them. RxDart must always know they never work unless you subscribe // Stream.fromIterable builds a stream which emits the values from the given list Stream.fromIterable([1, 2, 3, 4, 5, 6, 7]) // where function calls its body each time a value is emitted by the stream // and checks if it is an even number. This function acts like a filter. .where((element) { // if the element is even this function returns true // and the element passes the filter // otherwise it returns false and the element is filtered out. return element % 2 == 0; }); You must always subscribe to a stream to get any information from it! The above code because there is no other code that subscribes/listens to it. So what we can do to make it work is to call listen on this stream. does nothing var evenNumbers = Stream.fromIterable([1, 2, 3, 4, 5, 6, 7]) .where((element) { return element % 2 == 0; }); evenNumbers.listen((element){ print(element); }); Now, what function does is to create a subscription to the stream and listen to its events. The moment is called and registered as a subscription for stream, the stream code starts working. Firstly it takes the first element from the list then it checks it in the function and then if it passes the test (in our case is an even number) it is sent in the callback and then it’s printed. So the above code prints .listen evenNumbers .listen evenNumbers .where listen 2 4 6 When we manually listen for streams we have to be really careful because we can easily add memory leaks to our code. When we listen to a stream, a subscription is created. This subscription is stored in memory and therefore if it’s not disposed/deleted it will remain there, causing memory leaks. What we can do to prevent this to happen is store the subscription somewhere and then, we don’t need them anymore we dispose of them. CompositeSubscription subscriptions = CompositeSubscription(); var evenNumbers = Stream.fromIterable([1, 2, 3, 4, 5, 6, 7]).where((element) { return element % 2 == 0; }); subscriptions.add(evenNumbers.listen((element) { print(element); })); // other code .... // ..... // When we are done with our streams subscriptions.dispose(); Dart provides us with an easy way to handle our subscriptions. There is which is similar to a list of subscriptions. Each time we want to listen to a stream, we add the listen inside of our subscription list, and at the end when we don’t need them anymore we call function which closes all the subscriptions from memory. CompositeSubscription dispose More information about can be found on their GitHub page RxDart https://github.com/ReactiveX/rxdart A non-disposed Stream subscription produces memory leaks What is App State and how do we Create it? Before managing the state, we need to create a class that encapsulates our app states ( , , ). To avoid lots of boilerplate code for creating different methods on models ( , , , etc) I’m using . Loading Data Error copyWith equals toString when freezed library @freezed class AppResult with _$AppResult { const AppResult._(); const factory AppResult.data(T value) = Data; const factory AppResult.loading() = Loading; const factory AppResult.appError([String? message]) = AppError; } Now that we have our class, we can use it to handle our states. So what we are going to do is whenever we want to handle these 3 states somewhere in the app we will make functions to return an object of type This object can have one of the 3 states ( , , ). AppResult AppResult Data Loading Error After creating the state, we have to handle it. Since we are using it already builds for us a method that allows us to do an action for each possible state. freezed when appResult.when( data: (data) { // code run when appResult has data stata }, loading: () { // code run when appResult has loading state }, appError: (appError) { // code run when appResult has error state }, ); If we don’t use then we have to manually write a switch to check which state we are currently in. e.g. freezed switch(appResult){ case AppResult.data: // code run when appResult has loading state break; case AppResult.loading: // code run when appResult has loading state break; case AppResult.error: // code run when appResult has error state break; } State Management with RxDart Before jumping into the code I want to describe a little the application built using for state management. There will be an app that provides a list of universities that can be queried by their country. RxDart There we are using: for data streams and state management RxDart with as a design pattern MVVM Clean Architecture with for handling API calls Retrofit Dio for avoiding models boilerplate code Freezed for helping us with serializing and de-serializing JSON data Json Serializer Free API for getting universities’ data [http://universities.hipolabs.com](http://universities.hipolabs.com) This article only focuses on RxDart state management. Since we are following the MVVM with Clean Architecture it means we have the following flow of data: / is opened Wiget Screen When it is initialized, its is initialized with it as well ViewModel The calls the usecase for initializing the data ViewModel The usecase calls the repository to get the data Based on where the data is and the logic behind it, the repository calls an endpoint to an API or queries the local database, or gets the data from shared preferences. The data comes back exactly through the same classes in reverse order using a stream. e.g: is opened UniversitiesScreen initializes the UniversitiesScreen ViewModel subscribes to the stream with universities data from UniversitiesScreen ViewModel calls to get the universities UniversitiesViewModel GetUniversitiesByCountryUseCase The use-case calls to get the universities’ data UniversitiesRepository The repository calls the to get the universities’ data UniversityRemoteDataSource The data source using does an API request to the API to fetch the data UniversityEndpoint When there is a result from the API call the data goes all the way back via the streams. gets the data (error, data, loading) UniversityEndpoint it sends it back to UniversityRemoteDataSource sends it back to UniversitiesRepository sends it back to GetUniversitiesByCountryUseCase sends it back to UniversitiesViewModel then it finally arrives on the where it is handled by its state and displayed UniversitiesScreen Since we are using streams, we can easily send how many values we want on the same stream. Let’s jump into the code! Here we have the : UniversitiesScreen class UniversitiesScreen extends StatefulWidget { const UniversitiesScreen({Key? key}) : super(key: key); @override State createState() => _UniversitiesScreenState(); } class _UniversitiesScreenState extends State { final UniversitiesViewModel _viewModel = UniversitiesViewModel(); @override Widget build(BuildContext context) { return Scaffold( appBar: AppBar( title: const Text("RxDart State"), ), body: Column( children: [ Container( margin: const EdgeInsets.all(10), child: TextField( onChanged: _viewModel.searchByCountry, decoration: const InputDecoration( labelText: 'Search', suffixIcon: Icon(Icons.search)), ), ), Expanded( child: StreamBuilder( stream: _viewModel.universities, builder: (BuildContext context, AsyncSnapshot > snapshot) { return snapshot.data?.when( data: (e) => _buildUniversities(e.universities), loading: () => _buildLoading(), appError: (e) => _buildError(e.toString()), apiError: (e) => _buildError(e.toString())) ?? _buildLoading(); }, ), ), ], ), ); } Widget _buildUniversities(List universities) { return ListView.builder( itemCount: universities.length, itemBuilder: (BuildContext context, int index) { return Card( elevation: 5, margin: const EdgeInsets.all(10), child: Container( padding: const EdgeInsets.all(25), child: Column( children: [ Text("Name: ${universities[index].name}"), Text("Country: ${universities[index].country}"), Text("Website: ${universities[index].website}"), ], ), ), ); }, ); } Widget _buildLoading() { return const Center( child: CircularProgressIndicator(), ); } Widget _buildError(String error) { return Center( child: Text( error, style: Theme.of(context).textTheme.headline3?.copyWith(color: Colors.red), ), ); } } Our screen is built having a with an and a column with an input for searching the universities by country and then the list of universities. Scaffold AppBar TextField For the list of universities, we can see that we have a there: StreamBuilder StreamBuilder( stream: _viewModel.universities, builder: (BuildContext context, AsyncSnapshot > snapshot) { return snapshot.data?.when( data: (e) => _buildUniversities(e.universities), loading: () => _buildLoading(), appError: (e) => _buildError(e.toString()), apiError: (e) => _buildError(e.toString())) ?? _buildLoading(); }, ), We are using the here because all of our data is managed by , remember? The is managing the subscription to the university stream automatically (remember that for a stream to work we have to / to it and then take care of the subscription to prevent memory leaks) and it provides us a callback function where we can build the actual widget that we want to display based on the result provided. StreamBuilder Streams StreamBuilder subscribe listen builder We can also manually subscribe to the stream and handle the state changing using callback. Below is an example of how to listen to a stream and how to dispose of the subscriptions. setState final UniversitiesViewModel _viewModel = UniversitiesViewModel(); final CompositeSubscription _subscriptions = CompositeSubscription(); AppResult _screenState = const AppResult.loading(); @override void initState() { super.initState(); _subscriptions.add(_viewModel.universities.listen((event) { setState(() { _screenState = event; }); })); } @override void dispose() { _subscriptions.dispose(); super.dispose(); } @override Widget build(BuildContext context) { return Scaffold( appBar: AppBar( title: const Text("RxDart State"), ), body: Column( children: [ Container( margin: const EdgeInsets.all(10), child: TextField( onChanged: _viewModel.searchByCountry, decoration: const InputDecoration( labelText: 'Search', suffixIcon: Icon(Icons.search)), ), ), Expanded( child: _screenState.when( data: (e) => _buildUniversities(e.universities), loading: () => _buildLoading(), appError: (e) => _buildError(e.toString()), apiError: (e) => _buildError(e.toString())), ), ], ), ); } } The is initialized exactly as before, but this time we are using a variable to keep the app state. Since we don’t have any data stored locally and we always have to fetch it from the server we want to see a loading indicator when we initialize the screen so we also initialize the variable with the loading state. ViewModel _screenState Then, before the screen is built, in the callback we subscribe to the universities data stream and register the subscription in our subscriptions list . In the subscription callback, you can see that we call and then we set the new value in our variable. Here the magic happens. When is called, Flutter checks what was changed and goes into the widget tree, and rebuilds the widgets depending on the changed variables. So it goes in method and rebuilds the widgets from there using the new initState _subscriptions setState _screenState setState build _screenState To prevent memory leaks we dispose of/close all the subscriptions from the list inside callback. dispose This is all we have to do for managing the screen state on the UI. In the there are only a few lines of code. UniversitiesViewModel class UniversitiesViewModel { final GetUniversitiesByCountryUseCase _getUniversitiesByCountryUseCase; final Subject _searchByCountry = PublishSubject(); late Stream > universities; UniversitiesViewModel( {GetUniversitiesByCountryUseCase? getUniversitiesByCountryUseCase}) : _getUniversitiesByCountryUseCase = getUniversitiesByCountryUseCase ?? GetUniversitiesByCountryUseCase() { universities = _searchByCountry .startWith(null) .flatMap((value) => _getUniversitiesByCountryUseCase.invoke(value)); } void searchByCountry(String country) { _searchByCountry.add(country); } } We can notice that here we have something new which comes from . Here is a which is created using a . These s are some extensions over the , which is just a stream that allows us to send values through it. This called is used for sending the search query to the for providing us the universities from that country. Since we don’t want any other class to be able to send information via this subject, we make it private. RxDart Subject PublishSubject Subject StreamController Subject _searchByCountry useCase Now that we have a way to send our requests we also need a way for the view to be able to get the information that it needs. For this, we create a variable that will send the app state data through it. This variable be initialized in the ’s constructor. There we also set some rules/transformations to it: Stream universities must viewModel // We assign our _searchByCountry Subject to the universities variable universities = _searchByCountry // When the application is open we also want to load some universities and because // there we don't want to choose a country by default we just tell the stream // connection to send null when it's connected for the first time .startWith(null) // Now we want to call our useCase to get us the universities for the specified country // This is done by using flatMap which maps/transforms a stream into another stream .flatMap((value) => _getUniversitiesByCountryUseCase.invoke(value)); Now that we initialized our stream of data, there is nothing else for us to do. The view has to subscribe to streams and it will get the needed data. universities If you noticed on the screen, there was the used for getting user’s input for a specific country Search TextField TextField( onChanged: _viewModel.searchByCountry, decoration: const InputDecoration( labelText: 'Search', suffixIcon: Icon(Icons.search)), ) There we have to set a callback function for action so the moment the text is changed, it will call the callback function we specify. In our case, we will use the function from our . What this function does, is that it only sends the string received into the internal subject and then everything happens automatically (the string is sent to → the calls → and then it sends the data back to the listener) onChanged searchByCountry viewModel flatMap flatMap _getUniversitiesByCountryUseCase void searchByCountry(String country) { _searchByCountry.add(country); } Now we just cover everything related to the actual state management for our Flutter widgets. For this article, I haven’t covered the details of the architectural approach or how the data is fetched. In our case, if you don’t want to use MVVM with Clean Architecture, you can do directly the API call in the or even in the widget and then transform it into a stream of and everything will work. As long as you use Streams for getting the data you can the above steps can be applied for any architecture you choose. viewModel AppResult Why Streams and not Async and Await? The main difference between and is that a is used to send multiple data asynchronous whereas is used to only send one element asynchronous. streams async-await Stream async-await For lots of the projects which only interact with Rest APIs, there will be no need for streams, a simple await on an async function can do the job. But in that case, we have to handle the loading state manually, maybe in the or in the widget. viewmodel Moreover, if an application will need to connect to a then an async-await can’t help us and we must use a stream that is connected to the and then continuously sends us the data. WebSocket WebSocket Another use case that is way easier to handle with Streams is when we need to get the data from the local storage, display it on the screen, and then update it from the server, and after it is updated display the updated version. Benefits of streams Most of the time in real-world applications there are screens that needs data from multiple sources. An example can be a screen with user info and a list of users’ books. In this case, we have to combine data from two sources (from two streams in this case) and merge their data into a object. e.g ScreenState class GetMainScreenUseCase { final BooksRepository _booksRepository; final UserRepository _userRepository; GetMainScreenUseCase({ BooksRepository? booksRepository, UserRepository? userRepository, }) : _booksRepository = booksRepository ?? BooksRepository(), _userRepository = userRepository ?? UserRepository(); Stream > invoke() { return CombineLatestStream.combine2( _userRepository.getUserData(), _booksRepository.getBooks(), (userDataResponse, booksResponse) { // Here we need to check if we got success on both streams if (userDataResponse is AppResult.data && booksResponse is AppResult.data) { // Since we got data on both streams now we create our ScreenState // with the received data. return AppResult.data( ScreenState.from( userDataResponse, booksResponse, ), ); }else if(....){ ..... }.... else{ ..... } }, ); } } There can be other cases where you need to display some data from different sources, for that case you can use . MergeStream Other useful streams are which are streams which keep the last emitted value (saves state) and directly send it when someone subscribes to it. This behavior can be added to simple streams as well, by using extension. These two are useful when we want to keep the screen’s state e.g. User is on the main page, see all the data there, then they go on the details page, and when they come back to the main screen we need to display the same data as before without triggering any DB/API/SharedPreferences call. BehaviourSubjects .reply You can find out more about what these streams provide by reading and the one from . RxDart documentation RxJava https://github.com/ReactiveX/rxdart https://reactivex.io/documentation/observable.html Other articles about and its integration with Flutter: RxDart https://medium.com/flutter-community/why-use-rxdart-and-how-we-can-use-with-bloc-pattern-in-flutter-a64ca2c7c52d https://medium.com/globant/intro-to-rxdart-7e1429fabce0