Streams in dart

Sup folks, hope you are doing great 🤗, we are going to pass through many in one flutter concept 😂 concept called a stream.

Let's start with a simple case study.
Imagine yourself sitting by a creek, having a wonderful time. While watching the water flow, you see a piece of wood or a leaf floating down the stream and you decide to take it out of the water 😂 Funny right? You could even have someone upstream purposely float things down the creek for you to grab.

So we can imagine Dart in a similar way: as data flowing down a creek, waiting for someone to grab it. A stream can be like a pipe, you put a value on the one end and if there’s a listener on the other end that listener will receive that value. A Stream can have multiple listeners and all of those listeners will receive the same value when it’s put in the pipeline. The way you put values on a stream is by using a StreamController.

What are streams in dart?

A stream is a source of asynchronous data events. They provide a way to receive a sequence of events.

What can you do with a stream ?

• With Dart streams, you can send one data event at a time while other parts of your app listen for those events. Such events can be collections, maps or any other type of data you’ve created.

• Streams can send errors in addition to data thus you can also stop the stream, if you need to.

Important concepts of Streams in flutter

1. Stream Controller : A StreamController simplifies stream management, automatically creating a stream and sink, and providing methods for controlling a stream’s behavior. A StreamController object in Dart does exactly what the name suggests, it controls Dart Streams. The object is used to create streams and send data, errors, and done events on them. Controllers also help check a stream’s properties, such as how many subscribers it has or if it’s paused.

2. Stream Builders : StreamBuilder is a widget that uses stream operations and basically, it rebuilds its UI when it gets the new values that are passed via Stream it listens to.

StreamBuilder requires 2 parameters:

• stream: A method that returns a stream object
• builder: widgets that will be returned during different states of a streambuilder.

3. Sink: In Flutter Streams, a Sink is a point from where we can add data into the stream pipe.

4. Source: In Flutter Stream, a Source is a point from where we can keep listening to stream data or get the data that is been added into the stream.

How to create streams in flutter?

Let's first have a look at generators.

Generators can create sequence of values synchronously (by this we can create Iterable) and asynchronously (by this we can create Stream).
So, you can produce a stream by calling an async* function, which then returns a stream. Consuming that stream will lead the function to emit events until it ends, and the stream closes. You consume a stream either using an await for loop, which is available inside an async or async* function, or by forwarding its events directly using yield* inside an async* function.

Future<double> getRandomValue() async {
  var random = Random(2);
  await Future.delayed(Duration(seconds: 1));
  return random.nextDouble();
}

This code can be used and you’ll get a random value back once off and the function will be finished executing. This means if you want another random value you’ll have to call and await the function again. Like below.

var value1 = await getRandomValue();
var value2 = await getRandomValue();

And this is so tiresome right?

So if you want to call the function once and continuously get random values from that function without stopping it’s execution? That’s where async* and yield comes in. Hmm, interesting then lets make a function that returns a stream and every second will emit a new random value.

Stream<double> getRandomValues() async* {
    var random = Random(2);
    while (true) {
      await Future.delayed(Duration(seconds: 1));
      yield random.nextDouble();
    }
  }

Well if you are confused about the difference between the previous function and this one don't worry, let's clearly look at the differences:

• The first thing to notice is that we now return a Streamand not a Future. That means instead of awaiting on the value we’ll have to subscribe to the stream.

• The second difference is that async* instead of async. This tells the runtime that this code should be run asynchronously but execution will continue even after "returning" a value.

• The last difference is the replacement of return with yield. This is basically a return function but it doesn't exit the function. Instead it continues executing the rest of the code after yield.

Welp, that was one way of creating streams 🤗 Tadaaa !!!!

Usually when you are creating streams, you use StreamController which holds both stream and StreamSink like below.

final streamController = StreamController<DateTime>();
    Timer.periodic(Duration(seconds: 2), (timer) {
      streamController.add(DateTime.now());
    });

Before we continue notice that we emitted a new value over a stream using add

How to use streams?

The next thing to do is to be able to get the values from a stream. This is commonly referred to as subscribing or listening to a stream. When you subscribe to a stream you will only get the values that are emitted (put onto the stream) after the subscription. You subscribe to the stream by calling the listen function and supplying it with a Function to call back to when there's a new value available, commonly referred to as a callback function, or just a callback.

streamController.stream.listen((event) {
      print(event);
    });

Managing a stream

The listen call returns a StreamSubscription of the type of your stream. We can use this to manage stream subscription. Notice that we have our subscription to the stream and it’s emitting over time. That’s great. But we’ve got a small bug with this implementation: we never disposed or cleaned up our subscription to the stream. And this means that even if the user goes to another part of our app or does something else, our app will still listen to this stream and process the results.
Basically we need to make sure there are no memory leaks. A subscription to a stream will stay active until its memory allocation is destroyed, usually the entire lifecycle of your app. This is perfectly fine in some cases and not fine in others. Let's look together how to cancel to a stream subscription.

  final streamController = StreamController<DateTime>();
    final unsubscribeAt = DateTime.now().add(Duration(seconds: 10));
    StreamSubscription<DateTime>? subscription;

    Timer.periodic(Duration(seconds: 2), (timer) {
      streamController.add(DateTime.now());
    });

    subscription = streamController.stream.listen((event) async {
      print(event);
      if (event.isAfter(unsubscribeAt)) {
        print("It's after ${unsubscribeAt}, cleaning up the stream");
        await subscription?.cancel();
      }
    });

Handling stream errors.

As we know 😂 when we are programming we often meet errors. So we are going to see how we can manage errors in our streams.

The reasons for these can be vast, but if your stream is connected for real-time updates from a server and the mobile device disconnects from the internet, then the stream disconnects as well and yields an error.

When this happens and we don’t handle the error, Flutter will throw an exception and the app can potentially be left in an unusable state.

Fortunately, it’s fairly easy to handle errors. Let’s make our stream yield an error if our seconds are divisible by three, and, for the sake of completeness, let’s also handle the event when the stream completes:

final streamController = StreamController<DateTime>();
    final unsubscribeAt = DateTime.now().add(Duration(seconds: 10));
    late StreamSubscription<DateTime> subscription;

    final timer = Timer.periodic(Duration(seconds: 2), (timer) {
      streamController.add(DateTime.now());
      if (DateTime.now().second % 3 == 0) {
        streamController.addError(() => Exception('Seconds are divisible by three.'));
      }
    });

    subscription = streamController.stream.listen((event) async {
      print(event);
      if (event.isAfter(unsubscribeAt)) {
        print("It's after ${unsubscribeAt}, cleaning up the stream");
        timer.cancel();
        await streamController.close();
        await subscription.cancel();
      }
    }, onError: (err, stack) {
      print('the stream had an error :(');
    }, onDone: () {
      print('the stream is done :)');
    });

The output will be

Bonus 🎉

What are the types of streams in dart?

There are two kinds of streams: "Single-subscription" streams and "broadcast" streams.

• A single-subscription stream allows only a single listener during the whole lifetime of the stream. It doesn't start generating events until it has a listener, and it stops sending events when the listener is unsubscribed, even if the source of events could still provide more. The stream created by an async* function is a single-subscription stream, but each call to the function creates a new such stream.

Listening twice on a single-subscription stream is not allowed, even after the first subscription has been canceled.

Single-subscription streams are generally used for streaming chunks of larger contiguous data, like file I/O

• A broadcast stream allows any number of listeners, and it fires its events when they are ready, whether there are listeners or not.

Broadcast streams are used for independent events/observers.

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

Hoorraaay 😝 congrants you've read to this far. This was a lot. But hope you understand what a stream is, let's have a short recap about what we have covered.

• How a basic stream works and what purpose they serve
• How to clean up a subscription after we’ve used it
• How to handle basic errors that come from the stream and capture them when a stream completes
• Types of streams

So your task now is examine yourself and see how you can implement this in a flutter app 🔥 just give it a try.
Goodluck 🎉🎉