RxDart Tutorial for Flutter: Getting Started

Subjects

Up until this point, you’ve been dealing with streams and the improvements RxDart offers to them. Let’s shift the focus to StreamControllers. The StreamController in Dart manages streams, exposing various metadata information and callbacks associated with streams.

By default, they only allow single-subscription streams. To understand single-subscription streams take a look at lib/engine.dart. You use the following code snippet to initialize the game, player and input stream controllers:

final StreamController<Tetrimino> _playerController = StreamController();
final StreamController<UserInput> _inputController = StreamController.broadcast();
final StreamController<GameData> _gameController =
    StreamController.broadcast();

There are two different observations you can make from the code above:

• _playerController manages a single-subscription stream.
• _inputController and _gameController handle broardcast streams, meaning they handle multiple listeners.

If you didn’t specify the StreamController.broadcast and rather initialized _gameController with StreamController() for instance, it’ll throw a runtime exception with the following stacktrace:

Subjects in RxDart are the equivalent of StreamController in the native Dart API, albeit with more capabilities. They handle broadcast streams by default. Let’s have a look at them in a bit more detail.

BehaviorSubject

BehaviorSubject is one of the many different types of Subjects offered by RxDart. They offer all the functionality of a StreamController but differ in the fact that they manage broadcast streams by default. So, to solve the problem you encountered earlier using RxDart, you’ll simply replace the StreamController construction invocation for _gameController with a BehaviorSuject. Replace //TODO: replace with BehaviorSubject and the assignment below it in lib/engine.dart with the following:

//TODO: replace with seeded BehaviorSubject
final StreamController <GameData>_gameController = BehaviorSubject();

Seeded BehaviorSubjects

Besides handling broadcast streams by default, BehaviorSubjects also support seed values. Seed values are the first events transmitted when you initialize BehaviorSubjects. In the case of the game, you can use them to represent a start screen which is shown when none of the blocks have been set. Now, you’ll leverage the power of RxDart and create a seeded BehaviorSubject.

While still in lib/engine.dart, replace //TODO: replace with seeded BehaviorSubject and the code beneath it with the following statement:

final StreamController<GameData> _gameController =
      BehaviorSubject.seeded(GameData(state: GameState.Start, pieces: []));

Build and run the app. Although this is the same start screen as the one from the start of the tutorial, it was achieved using RxDart.

Awesome!

Introducing Backpressure

The whole reactive programming style is useful for developers since the UI can only subscribe to the streams they want to listen to. This promotes the idea of “Separation of Concerns, ” a crucial idea for developing apps.

Of course, reactive programming has issues as well. For instance, think about what happens when the source is much faster than the sink. That is, the sink finishes processing the events at a much slower rate than the source is creating events.

A common example of this issue is in implementing a suggestive search field. As the user types the query, the app needs to present suggestions. If you use a REST API to back the suggestion functionality that takes the query as a parameter, there’ll be a certain amount of delay until you receive a response.

If this delay proves to be greater than the time it takes for the keyword to change, the response returned may be useless for certain search queries. In other words, you’re making unnecessary network calls.

In terms of streams, the keyword changes are the source and the responses from the API are the sink. The source in this case is a lot faster than the sink. Reactive programming uses the term Backpressure to denote this phenomena and RxDart offers various operators to deal with it. Three notable examples of these operators are:

• debounce: Emits items from the input stream only when a “window” has completed.
• debounceTime: Similar to debounce but the “window” in this case is just a timer.
• exhaustMap: Useful for converting events of an input stream into another stream. Ignores all subsequent events of the input stream until the result stream has completed.

To see how you can use these operators to deal with backpressure, open lib/engine.dart and replace //TODO: add the debounceTime operator with the following:

.debounceTime(const Duration(seconds: 1))

Build and run the app. debounceTime will allow _inputController to now ignore frequent taps for a certain period of time:

Where to Go From Here?

Download the final project by clicking the Download Materials button at the top or bottom of this tutorial. In this tutorial, you learned the basics of Dart streams and how RxDart augments it using the following concepts:

• Rx Streams: Builds upon Dart streams to offer more features and flexibility in working with streams.
• Subjects: Works like StreamControllers, but with additional powers.
• Extension Functions: Extends the Dart streams API with expressive, chainable functions to make developing reactive apps easier.
• Backpressure: Arises when there’s inconsistency between the source and sink. You learned how to ensure there’s consistent data synchronization between stream sources and sinks.

To dive deeper, check out the following resources:

• Visit the Dart docs on Asynchronous programming to learn about its nuances.
• Check out the RxDart package and all the extension functions and subjects it provides.
• To strengthen your understanding of Reactive Programming, you can checkout Rx Marbles. It has interactive diagrams that help you understand how different streams operate. Also as a plus point, almost every stream mentioned in the RxDart docs has a corresponding marble diagram.

If you have any questions, comments or suggestions, feel free to join the discussion below!