RxDart & Streams Crash Course

Jan 31, 2021 9 min read
RxDart & Streams Crash Course

A crash course to learn streams and RxDart in detail with curated diagrams, examples and videos.

RxDart adds additional capabilities to Dart Streams and StreamControllers.

RxDart does not provide its own Observable class as a replacement for Dart Streams. Rather, it provides a number of additional Stream classes, operators (extension methods on the Stream class), and Subjects.

How To Use RxDart


Adding RxDart dependency in pubspec.yaml

rxdart: ^0.25.0
rxdart | Dart Package
rxdart - RxDart is an implementation of the popular reactiveX api for asynchronous programming, leveraging the native Dart Streams api.

What is a Stream?

A Stream provides a way to receive a sequence of events. Each event is either a data event, also called an element of the stream, or an error event, which is a notification that something has failed. When a stream has emitted all its event, a single "done" event will notify the listener that the end has been reached.

You listen on a stream to make it start generating events, and to set up listeners that receive the events. When you listen, you receive a StreamSubscription object which is the active object providing the events, and which can be used to stop listening again, or to temporarily pause events from the subscription.

Types of Stream

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.

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.

On either kind of stream, stream transformations, such as where and skip, return the same type of stream as the one the method was called on, unless otherwise noted.

Let's understand it by an Example

var subject = StreamController();

subject.stream.listen((value) {
    print("Added $value to 1st Subsciber");
});

//This will produce an error
subject.stream.listen((value) {
    print("Added $value to 2nd Subscriber");
});

If we will try to run this code, it will give us an error because we have already listened to the stream once.

Solution - Replace StreamController() to StreamController.broadcast()

var subject = StreamController.broadcast();

Rx PublishSubject

Exactly like a normal broadcast StreamController with one exception: this class is both a Stream and Sink.

final subject = PublishSubject<int>();

// observer1 will receive all data and done events
subject.stream.listen(observer1);
subject.add(1);
subject.add(2);

// observer2 will only receive 3 and done event
subject.stream.listen(observe2);
subject.add(3);
subject.close();
Great! Next, complete checkout for full access to Codepur.
Welcome back! You've successfully signed in.
You've successfully subscribed to Codepur.
Success! Your account is fully activated, you now have access to all content.
Success! Your billing info has been updated.
Your billing was not updated.