Understanding Streams in Dart: A Complete Guide

Streams are one of the foundations of Dart asynchronous programming, and In its simplest form, a stream is an asynchronous event that can be in the form of single element or collection. Streams are widely used for managing events in Dart applications (eg Flutter), such as user interactions, file I/O events, and API responses.

What is a Stream?

A Stream is an asynchronous event sequence. These events can be:

Single subscription: Only one listener is allowed at a time

Broadcast: It is permissible with many listeners at the same time.

Key Concepts

• Stream: The source of asynchronous data.

• StreamController: Manages the stream and its sink.

• StreamSubscription: Represents the listening process to a stream.

Creating a Stream

a. Using StreamController

StreamController StreamController is often used to create custom streams.

final StreamController<int> controller = StreamController<int>();

void main() {
  final stream = controller.stream;

  stream.listen((data) {
    print('Data received: $data');
  });

  controller.sink.add(1); // Emit data
  controller.sink.add(2);

  controller.close(); // Close the stream
}

b. Using Stream.fromIterable

Creates a stream from a collection.

final Stream<int> stream = Stream.fromIterable([1, 2, 3, 4, 5]);

stream.listen((event) {
  print('Event: $event');
});

c. Using Stream.periodic

Generates events periodically.

final Stream<int> stream = Stream.periodic(
  Duration(seconds: 1),
  (count) => count,
);

stream.take(5).listen((event) {
  print('Periodic event: $event');
});

Listening to Streams

You can listen to streams using the listen method, which returns a StreamSubscription.

Listening Example

final Stream<int> stream = Stream.fromIterable([1, 2, 3]);

final subscription = stream.listen((data) {
  print('Data: $data');
});

subscription.onDone(() {
  print('Stream closed');
});

Transforming Streams

Streams support powerful transformation operations like mapping, filtering, and reducing.

a. Mapping

stream.map((event) => event * 2).listen((data) {
  print('Mapped Data: $data');
});

b. Filtering

stream.where((event) => event % 2 == 0).listen((data) {
  print('Even Data: $data');
});

c. Reducing

stream.reduce((acc, curr) => acc + curr).then((sum) {
  print('Sum: $sum');
});

Types of Streams

a. Single Subscription Stream

• Default type.

• Allows only one listener at a time.

• Used for one-time tasks like API calls.

final stream = Stream.fromIterable([1, 2, 3]);

b. Broadcast Stream

• Allows multiple listeners.

• Use .asBroadcastStream() or StreamController.broadcast.

final controller = StreamController<int>.broadcast();

controller.stream.listen((data) {
  print('Listener 1: $data');
});

controller.stream.listen((data) {
  print('Listener 2: $data');
});

controller.add(1);
controller.add(2);

Handling Errors

Streams can emit errors, which you can handle using the onError callback.

stream.listen(
  (data) {
    print('Data: $data');
  },
  onError: (error) {
    print('Error: $error');
  },
  onDone: () {
    print('Stream completed');
  },
);

Combining Streams

You can merge multiple streams or zip them together.

a. Merging Streams

Stream<int> stream1 = Stream.fromIterable([1, 2, 3]);
Stream<int> stream2 = Stream.fromIterable([4, 5, 6]);

Stream<int> mergedStream = Stream.fromFutures([
  stream1.toList(),
  stream2.toList(),
]).expand((element) => element);

mergedStream.listen(print);

b. Zipping Streams

Use external packages like rxdart for advanced combinations.

StreamController Types

a. Regular StreamController

Single-enterprise streams are for

b. Broadcast StreamController

For a broadcast stream with multiple listeners.

final controller = StreamController<int>.broadcast();

Asynchronous Generators

Dart provides a convenient way to create streams using the async* keyword.

Stream<int> generateNumbers(int max) async* {
  for (int i = 1; i <= max; i++) {
    yield i;
    await Future.delayed(Duration(seconds: 1));
  }
}

generateNumbers(5).listen(print);

Best Practices

1. Close Controllers: Always close StreamController to release resources.

2. Broadcast Streams: Use broadcast streams for shared data.

3. Error Handling: Provide robust error handling.

4. Use Extensions: Utilize rxdart or stream_transform for advanced operations.