🌎 preface

Both Stream and Future are core apis for the Dart: Async library and provide excellent support for asynchrony.

I thought a lot about how I would introduce a Stream to you. Because Stream is very useful, it is built to handle asynchronous events. In the application, there are a lot of scenarios that need to use asynchronous events, such as request network, and user interaction, etc., which cannot be done synchronously. Stream can be a great help in dealing with these issues. 😍

But a stream is abstract enough for the novice that it takes a lot of time to understand it.

So I will do my best to introduce Stream to you.

🏊 🏻 Stream

🤔 What is a Stream

A Stream is very idiosyncratic but not very easy to understand, and I prefer to think of it as a factory or machine rather than as a Stream literally.

Let’s take a look at the features of this machine:

  • It has a portal where you can put things/commands.
  • This machine doesn’t know when it’s gonna let something in
  • The machines in the middle can produce or process, which should take some time
  • He has an outlet. There should be something coming out of there
  • We don’t know exactly when the product will be exported

The whole process, the timing is an uncertain factor, we can put something into the entrance of the machine at any time, and then the machine will be processed, but we don’t know how long it will be processed. So the exit had to be watched, waiting for things to come out of the machine. The whole process is viewed asynchronously.

✈ ️We convert the machine model into a Stream

  • This big machine is the StreamController, which is one of the ways to create streams.
  • StreamController has an entry called sink
  • Sink can use the add method to put things in, and after putting things in, it will no longer care about them.
  • When something comes in from sink, our machine starts to work, empty.
  • StreamController has an outlet called Stream
  • After the machine finishes processing, the product will be thrown out from the exit, but we do not know when it will come out, so we need to use listen method to monitor the exit all the time.
  • And when multiple items are put in, it doesn’t shuffle, it’s first in, first out.

From this example, I believe that you should have a basic impression of convection, then it is not difficult to explain the following things.

🤔 How to use Stream

Method of obtaining Stream:

  • By constructor
  • Using StreamController
  • IO Stream

A stream has three constructors:

  • Stream.fromFuture: Creates a new single-subscription Stream from the Future, raises a data or error when the Future completes, and closes the Stream with the Down event.

  • Stream.fromFutures: Create a single subscription Stream from a set of Futures. Each Future has its own data or error event. If Futures is empty, the stream will be closed immediately.

  • Stream.fromIterable: Creates a single subscription Stream that retrieves data from a collection.


Stream.fromItreable([1.2.3]);
Copy the code

🧐 method to listen on Stream

The most common way to listen to a stream is to listen. The flow notifies the listener when an event is emitted. The Listen method provides these trigger events:

  • OnData (Mandatory) : Triggered when data is received
  • OnError: Triggered when an Error is received
  • OnDone: trigger at the end
  • UnsubscribeOnError: Whether to unsubscribe when the first Error is encountered. The default is false

Process the Stream with await for

In addition to listening, we can also use await for.

Future<int> sumStream(Stream<int> stream) async {
  var sum = 0;
  await for (var value in stream) {
    sum += value;
  }
  return sum;
}
Copy the code

This code will receive a Stream, count all the events, and return the result. Await for can process each event as it comes. We know that a stream is undefined when it receives events. When should we exit the await for loop? The answer is, when the Stream is finished or closed.

You can copy this code into DartPad for a few more tests.

😏 StreamController

If you want to create a new stream, it’s very easy! 😀 use StreamController, it gives you very rich functionality, you can send data on StreamController, handle errors, and get results!

// Stream of any type
StreamController controller = StreamController();
controller.sink.add(123);
controller.sink.add("xyz");
controller.sink.add(Anything);

// Create a stream that handles ints
StreamController<int> numController = StreamController();
numController.sink.add(123);
Copy the code

Generics define what types of data we can push onto a stream. It can be any type!

Let’s see how we get the final result.

StreamController controller = StreamController();

// Listen for the exit of the stream and print the data when it comes out
StreamSubscription subscription =
controller.stream.listen((data)=>print("$data"));

controller.sink.add(123);
Copy the code

Output: 123

You need to pass a method to the Stream’s Listen function. The method input (data) is the result of our StreamController processing. We listen for the exit and get the result (data). Lambda expressions can be used here, or any other function.

Listen =>print(data) =>print(data)

Stream generation via async*

If we have a series of events to process, we might want to convert it to stream. You can use async -yield * to generate a Stream.

Stream<int> countStream(int to) async* {
  for (int i = 1; i <= to; i++) {
    yieldi; }}Copy the code

When the loop exits, the Stream is done. We can combine with await for more profound experience.

You can run my sample code directly here.

🤠 Transforming an existing stream

If you already have a stream, you can convert it to a new stream. Very simple! Flows provide map(), WHERE (), expand(), and take() methods that make it easy to turn existing flows into new ones.

where

If you want to filter out some unwanted events. For example, in a guessing game, the user can type in a number, and when the number is correct, we react. If we have to filter out all the wrong answers, we can use where to filter out unwanted numbers.

stream.where((event){... })Copy the code

The WHERE function receives an event, and whenever something from this stream flows to the WHERE function, this is the event. We may not need this event at all, but it must be passed in as a parameter.

take

If you want to control how many things this stream can pass at most. For example, if we want the user to type a password up to four times, we can use take to limit it.

stream.take(4);
Copy the code

The take function accepts an int representing the maximum number of events that can pass through the take function. When this number of transfers is reached, the stream will be closed and cannot be transmitted again.

transform

If you need more control over the transformation, use the transform() method. It needs to work with StreamTransformer. Let’s look at a guessing game and I’ll explain it to you.

StreamController<int> controller = StreamController<int> ();final transformer = StreamTransformer<int.String>.fromHandlers(
    handleData:(value, sink){
  	if(value==100){
      sink.add("You guessed it.");
    }
	else{ sink.addError('Not yet. Try again.'); }}); controller.stream .transform(transformer) .listen( (data) =>print(data),
                onError:(err) => print(err));
    
    controller.sink.add(23);
    //controller.sink.add(100);
Copy the code

Output: Not right, try again

StreamTransformer<S,T> is the inspector for our stream. It receives the stream as it passes and processes it to return a new stream.

  • S stands for the input type of the previous stream, so in this case we’re entering a number, so int.
  • T stands for the input type of the transformed stream, and what we’re adding here is a String, so String.
  • HandleData receives a value and creates a new stream and exposes sink, where we can convert the stream.
  • We can also use addError to go in and tell you there’s a problem.

Then we listen to the stream after the transform. When the converted event flows out, we print this event, which is the data we added to sink just now. OnError captures the errs we add.

🤨 Stream type

There are two kinds of flow

  • “Single-subscription” Streams
  • “Broadcast” Streams Multiple subscription streams

“Single-subscription” streams

A single subscription stream is allowed to have only one listener during the lifetime of the stream. It doesn’t generate events until there’s a listener, and it stops sending events when you cancel listening, even if you’re still sinking.add more events.

Listening twice on a single subscription stream is not allowed even after the first subscription is cancelled.

Single-subscription flows are often used to stream larger contiguous blocks of data, such as file I/O.

StreamController controller = StreamController();

controller.stream.listen((data)=> print(data));
controller.stream.listen((data)=> print(data));

controller.sink.add(123);
Copy the code

Output: Bad state: Stream has already been listened to. A single subscription stream cannot have multiple listeners.

“Broadcast” streams

A broadcast stream allows any number of listeners and can generate events with or without listeners. So a listener who comes in mid-stream will not receive the previous message.

If multiple listeners want to listen to a single subscription stream, use asBroadcastStream to create a broadcast stream on top of a non-broadcast stream.

If a listener is added to the broadcast stream when an event is fired, the listener will not receive the event that is currently being fired. If you cancel listening, the listener immediately stops receiving events.

A typical stream is a single-subscription stream. Broadcast streams inherited from Stream must override isBroadcast to return true.

StreamController controller = StreamController();
// Convert a single subscription stream to a broadcast stream
Stream stream = controller.stream.asBroadcastStream();

stream.listen((data)=> print(data));
stream.listen((data)=> print(data));

controller.sink.add(123);
Copy the code

Output: 123 123

Refer to the article

  • Asynchronous programming: Use stream

🎆 is at the end

That’s a brief introduction to STREAMS in Dart. If you have any questions or suggestions, let me know in the comments section below or via email! I will contact you as soon as possible within 24 hours 😉

In my next post, I’m going to introduce you to a great state management approach that the Google team is pushing for, and I think it’s really cool!

So, are you ready to meet BLoC 😎