Dart Concurrency: Streams

Dart, a powerful and versatile programming language developed by Google, comes equipped with a robust concurrency model, allowing developers to handle asynchronous operations effectively. One key component of Dart's concurrency toolbox is Streams. In this post, we'll delve into Dart Streams, exploring their fundamentals and how they can enhance your asynchronous programming experience.

Understanding Dart Streams

Streams in Dart provide a way to handle sequences of asynchronous data. They are particularly useful when dealing with events, such as user input, network requests, or data from other asynchronous sources. Dart Streams enable developers to process and respond to events as they occur, without blocking the main execution thread.

A Stream can be thought of as a sequence of asynchronous events. These events can be of any type, allowing for flexibility in handling various types of data.

Creating a Stream

Creating a Stream in Dart is straightforward. Let's start with a simple example:

import 'dart:async';

void main() {
  // Create a StreamController to manage the stream
  final StreamController<String> streamController = StreamController<String>();

  // Create a Stream from the StreamController
  final Stream<String> myStream = streamController.stream;

  // Subscribe to the stream and listen for events
  final StreamSubscription<String> subscription = myStream.listen((data) {
    print('Received data: $data');
  });

  // Add data to the stream
  streamController.add('Hello, ');
  streamController.add('Dart Streams!');

  // Close the stream when done
  streamController.close();
}

In this example, we use a StreamController to manage our stream. The StreamController provides methods to add data to the stream (add) and close the stream (close). We then create a Stream from the StreamController and subscribe to the stream using listen. The callback inside listen will be invoked whenever new data is added to the stream.

Stream Transformation

Streams in Dart can be transformed using various methods to modify, filter, or combine the data they emit. Let's look at a simple example of transforming a stream:

import 'dart:async';

void main() {
  final StreamController<int> numberStreamController = StreamController<int>();

  final Stream<int> doubledStream =
      numberStreamController.stream.map((number) => number * 2);

  final StreamSubscription<int> subscription =
      doubledStream.listen((doubledNumber) {
    print('Doubled Number: $doubledNumber');
  });

  numberStreamController.add(1);
  numberStreamController.add(2);
  numberStreamController.add(3);

  numberStreamController.close();
}

In this example, we create a stream of numbers using a StreamController and then transform it using the map method to double each number. The transformed stream (doubledStream) is then subscribed to, and the doubled numbers are printed when they are emitted.

Error Handling and Stream Controllers

When working with Streams, it's essential to handle errors gracefully. Dart Streams provide mechanisms for error handling, such as the onError callback in the listen method.

Additionally, Stream Controllers can be used to broadcast events to multiple subscribers. By using a BroadcastStreamController, you can create a broadcast stream that allows multiple listeners to receive the same events independently.

import 'dart:async';

void main() {
  final StreamController<String> broadcastController =
      StreamController<String>.broadcast();

  final Stream<String> broadcastStream = broadcastController.stream;

  final StreamSubscription<String> subscription1 =
      broadcastStream.listen((data) {
    print('Listener 1 received: $data');
  });

  final StreamSubscription<String> subscription2 =
      broadcastStream.listen((data) {
    print('Listener 2 received: $data');
  });

  broadcastController.add('Broadcasting data');

  subscription1.cancel(); // Canceling one subscription does not affect others

  broadcastController.add('Another broadcast');

  broadcastController.close();
}

Here, we use a BroadcastStreamController to create a broadcast stream. Two listeners (subscription1 and subscription2) are added to the stream, and they both independently receive the broadcasted events. Canceling one subscription does not affect the other.