Dart Streams Basics

This is the second part of my Flutter Architecture series:

• Introduction
• Basics of Dart Streams (this post)
• RxDart: magical stream transformations
• RxVMS Basics: RxCommand and GetIt
• RxVMS: Services and Managers
• RxVMS: self-contained widgets
• User Authentication with RxVMS

The streams are the main building block of RxVMS , their understanding is absolutely necessary for working with this library, so we’ll dwell on them in this post.

It turned out that the inclusion of Rx in this post would make it too long, so I divided it into two parts.

Let it flow

I read a lot of comments that, say, streams, and especially Rx, are too difficult to understand and, as a result, to use.

I would like you to know that I do not consider myself a guru of Rx. Mastering all his power is not easy, and I admit that I continue to learn. But let me fix one misconception right from the start: you don’t need to be an Rx wizard to start getting a lot of benefits from using threads and this technology . I will make every effort to explain to you the flow in the most accessible way.

What are threads?

In my opinion, the best analogy of threads is the conveyor belt. You can put something on one end of it and this “something” is automatically transferred to the other. Unlike the physical pipeline, streams manipulate data objects, transferring them automatically from the beginning - but where? As in the real pipeline, if there is nothing that captures the data at the other end, it will simply “fall” and disappear (this is certainly not quite true for Dart Streams, but it’s best to treat the streams as if they were) .

In order to avoid data loss, you can set a "trap" on the output stream. So you can capture data and make the necessary manipulations with them whenever data objects reach the end of the stream.

Remember:

• If the trap is not set, the data will simply disappear forever and there will be no way to get them again (again, not quite so with Dart Streams, but you better pretend it is)
• After sending the data to the stream, you do not need to suspend the execution of the program and wait until they reach the end, it all happens in the background.
• The trap can receive data at any time, not necessarily immediately after sending (but you should not worry, the streams are actually very fast). Imagine that you don’t know how fast or how long a conveyor belt is moving. This means that placing something in the flow is completely separated from the reaction to the element at the other end. Your trap will work and catch the item when it gets there. (Some of you may already understand that this fits well with the reactive way Flutter updates its widgets)
• You can set a trap long before work begins and the first element appears.
• The stream works on the FIFO principle. Data always comes in the order they are placed in the stream.

What is Rx?

Rx, abbreviated from Reactive Extensions (reactive extensions), are streams "on steroids". This is a concept very similar to Streams, which was invented for the .Net framework by the Microsoft team. Since .Net already had the Stream type, which is used for file I / O, they called the Rx streams Observables and created many functions to manipulate the data passing through them. Dart has Streams embedded in its language specification, which already offer most of this functionality, but not all. That is why the RxDart package was developed; It is based on Dart Streams, but extends their functionality. I will look at Rx and RxDart in the next part of this series.

Some terms

Dart Streams and Rx use some terminology that may look scary, so here's the translation. First comes the term Dart, then Rx.

• Stream / Observable . This is the "conveyor" described earlier. Stream can be converted to Observable and wherever a Stream is expected, you can assign an Observable. So do not be confused if I will mix these terms in the process of explaining
• listen / subscribe - setting the trap- listener
• StreamController / Subject . The "left" side of the conveyor belt where you put the data in the stream. They are slightly different in their properties and characteristics, but serve the same purpose.
• Emitting an item / data . The moment when data appears at the exit of the "pipeline"

Stream creation

If you are going to continue studying the topic, please clone this project with examples. I will use the Dart / Flutter test system.

To create a stream, you create a StreamController

var controller = new StreamController<String>(); controller.add("Item1"); //      

The template type (in this case, the String) passed when the StreamController is created determines the type of objects that we can send to the stream. It can be ANY type! If you wish, you can create a StreamController<List<MyObject>>() and the stream will transfer the entire sheet instead of a single object.

Trap set

If you run the specified test, you could not see anything, because nothing caught our string at the output of the stream. Now we set the trap:

 var controller = new StreamController<String>(); controller.stream.listen((item) => print(item)); //  controller.add("Item1"); controller.add("Item2"); controller.add("Item3"); 

The hook is now set using the .listen() method. The record looks like controller.stream.listen , but if you scroll it backwards, like an album from the 60s, the true meaning of what was written: “listen to the stream of a given controller”

You need to pass a non-function to the .listen() method in order to somehow manipulate the incoming data. The function must accept a parameter of the type specified when creating a StreamController, in this case, a String.

If you run the above code, you will see

 Item1 Item2 Item3 

In my opinion, the biggest problem for newbies in Streams is that you can define a reaction for an emitted element long before the first element is placed in the stream, triggering a call to this reaction.

End listening

The code above missed a small but important part. listen() returns a StreamSubscription - a stream subscription object. Calling its method .cancel() terminates the subscription, freeing resources, and warning you to call your listening function after it has become unnecessary.

 var controller = new StreamController<String>(); StreamSubscription subscription = controller.stream.listen((item) => print(item)); // This is the Trap controller.add("Item1"); controller.add("Item2"); controller.add("Item3"); //    ,        //  ,     Stream   await Future.delayed(Duration(milliseconds: 500)); subscription.cancel; 

Details about the listeners

The function for listen() can be either a lambda or a simple function.

 void myPrint(String message) { print(message); } StreamSubscription subscription = controller.stream.listen((item) => print(item)); //  - StreamSubscription subscription2 = controller.stream.listen(myPrint); //    StreamSubscription subscription3 = controller.stream.listen((item) { print(item); print(item.toUpperCase); }); // - 

Important note: most Dart streams allow only a one-time subscription, that is, they cannot be subscribed to again after the subscription is completed - this will cause an exception. This is their difference from other implementations of Rx.

The complete listen() signature looks like this:

  /* excerpt from the API doc * The [onError] callback must be of type `void onError(error)` or * `void onError(error, StackTrace stackTrace)`. If [onError] accepts * two arguments it is called with the error object and the stack trace * (which could be `null` if the stream itself received an error without * stack trace). * Otherwise it is called with just the error object. * If [onError] is omitted, any errors on the stream are considered unhandled, * and will be passed to the current [Zone]'s error handler. * By default unhandled async errors are treated * as if they were uncaught top-level errors. * * If this stream closes and sends a done event, the [onDone] handler is * called. If [onDone] is `null`, nothing happens. * * If [cancelOnError] is true, the subscription is automatically canceled * when the first error event is delivered. The default is `false`. */ StreamSubscription<T> listen(void onData(T event), {Function onError, void onDone(), bool cancelOnError}); 

This means that you can do more than just pass one handler for the data sent. You can also have a handler for errors, and another for closing the flow from the controller ( onDone ). Exceptions that are triggered from within a Stream will cause onError() if you provide it, otherwise they are simply swallowed, and you will never know that something has gone wrong.

Flutter flow example

To facilitate the understanding of the following chapters, I made a separate repository branch.
Please incline her

As a first example, I took the well-known counter application that you get when you create a new Flutter project, and reorganized it a bit. I added a model class to store the state of the application, which is basically a counter value:

 class Model { int _counter = 0; StreamController _streamController = new StreamController<int>(); Stream<int> get counterUpdates => _streamController.stream; void incrementCounter() { _counter++; _streamController.add(_counter); } } 

here you can see a very typical pattern: instead of publishing the entire StreamController, we simply publish its Stream property.

To make the Model available to the UI, I made it a static field in the App object, because I didn’t want to enter InheritedWidget or ServiceLocator. For a simple example, it will get away with it, but I would not do it in a real application!

Add to main.dart :

 class _MyHomePageState extends State<MyHomePage> { int _counter = 0; StreamSubscription streamSubscription; @override void initState() { streamSubscription = MyApp.model.counterUpdates.listen((newVal) => setState(() { _counter = newVal; })); super.initState(); } //   State   ,   , //       @override void dispose() { streamSubscription?.cancel(); super.dispose(); } 

initState() good place to install a listener, and, being good citizens of Darts, we always release the subscription of dispose() , right?

In the tree of widgets, we just need to adapt the onPressed handler of the FAB button (a button with a floating action).

 floatingActionButton: new FloatingActionButton( onPressed: MyApp.model.incrementCounter, tooltip: 'Increment', child: new Icon(Icons.add), ), 

In this way, we created a clean separation between View and Model using Stream.

Apply StreamBuilder

A source

Instead of using initState() and setState() for our needs, Flutter comes with a handy StreamBuilder widget. As you may have guessed, it takes a Stream function and a builder method that is called whenever the Stream issues a new value. And now we do not need explicit initialization and release:

 body: new Center( child: new Column( mainAxisAlignment: MainAxisAlignment.center, children: <Widget>[ new Text( 'You have pushed the button this many times:', ), StreamBuilder<int>( initialData: 0, stream: MyApp.model.counterUpdates, builder: (context, snappShot) { String valueAsString = 'NoData'; if (snappShot != null && snappShot.hasData) { valueAsString = snappShot.data.toString(); } return Text( valueAsString, style: Theme.of(context).textTheme.display1, ); }), ], ), ), 

We are almost done, I promise. Here are three things you need to know:

• The great advantage of using StreamBuilder compared to the first solution is that the call to setState() in listen() always rebuilds the entire page, whereas StreamBuilder will only call its own builder
• The snapShot variable contains the latest data from the Stream. Always verify that it contains valid data before using it.

• Based on the principles of initialization during, StreamBuilder cannot get the value during the very first frame. To get around this, we pass the value for initialData , which is used for the first build, that is, for the first frame of the screen. If we do not pass initialData , our builder will be called for the first time with invalid data. An alternative to using initialData is to return a placeholder widget if snapShot valid, which is displayed until we get valid data, for example:

  
  

   // ,           StreamBuilder<int>( stream: MyApp.model.databaseUpdates, builder: (context, snappShot) { if (snappShot != null && snappShot.hasData) { return Text( snappShot.data.toString(), style: Theme.of(context).textTheme.display1, ); } //      ,   Spinner return CircularProgressIndicator (); }) 

  
  

In the next post, we will look at how to convert the data in our streams and do it on the fly. Many thanks to Scott Stoll for reading the proofreading and important feedback.