A crash course to learn streams and RxDart in detail with curated diagrams, examples and videos.
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
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.
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.
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();