Toggle A Stream On And Off With RxJS
This lesson covers how to toggle an observable on and off from another observable by showing how to use a checkbox as a toggle for a stream of data.
[00:02] Frequently in RSJX, you'll find yourself needing to toggle a stream on and off from another event source. That really means that you are using a stream to toggle on and off another stream.
[00:10] To demonstrate this, I am going to need a source stream. I'm going to use an interval of about a tenth of a second, and I'm going to map that to a series of dots.
[00:23] Now, I'll need a place to display my output. I'm going to select that from the dom, then I'll subscribe to my source, and add at the output to the inner text of my display. When I run this, see a series of dots coming in.
[01:00] But I need a way to toggle this on and off, so let's add an input. There's a checkbox. I'll give it an ID of, "toggle," and then I need to select that toggle, and I'm going to create a stream of, whether or not it's checked, by using observable from events, and it's changed events. Let's make this a little wider. Mapping that to its checked property.
[01:48] Let's make sure this works. We'll subscribe to it and put it in the display. You see true/false, true/false. That's what we expected.
[02:03] So, let's go back to subscribing to our source. I'm going to run this, and you see it automatically subscribes, it automatically starts pumping these dots into a display.
[02:13] We don't want it to start until we check our checkbox. To do that, we're going to have to lead in with a stream of when our checked stream is actually checked. So, I'm going to have to filter that down to only the times when it's actually true. So, this is now a stream of when this has been checked.
[02:33] Then I'll need to flatmap, or flatmap latest, that into my source. Now, if I subscribe to that, you'll notice it didn't start streaming in the dots right away. I have to check this checkbox to get it started, but if I uncheck it, it's not stopping. To get it to stop, that means I have to take my source until the very next checked event. Since the previous checked event was one where it was actually checked, the next time this fires will be a time that it's unchecked.
[03:15] So now when I run this, when I check it, it starts. When I uncheck, it stops. Check it again, it starts again, and again I can stop it. That's how you toggle a stream from another stream. You start with the stream of events that you want to use to toggle your source stream on, then you flatmap or flatmap latest into your source stream, but you only take it until the stream of events that you want to use to toggle it off fires one time.
[03:46] From there, you've created an observable that will toggle on and off based off another observable.
In truth, in this case? Not much difference. The flatMapLatest conveys that there's only one subscription going on at a time inside of the flatMap though.
However, if you were to have a situation where the "toggler" stream (the stream of checks) could return multiple trues in a row like: Observable.of(true, true, false, true, true), then using flatMapLatest will keep the underlying subscriptions to one, where flatMap will allow multiple subscriptions to occur.
Here's an altered version of the same jsbin: https://jsbin.com/cohufemero/edit?js,output
Other things to note, the source stream in this example is a cold observable. If we were dealing with a "hot" observable, we'd need to identify a strategy to deal with values that arrive while toggled "off". (buffer? drop? etc)
This example is restarting the stream every time the checkbox is checked. Is it possible to stop a stream, and continue where it was stopped, when it is toggled on?
Please ignore previous question, I understand it is against the philosophy of Rx to pause streams.
The functionality I like to have is a stream of ascending numbers, that can be paused. I found that that can be achieved by putting a scan function after the flatMapLatest.
how does the source stream know when to stop? Before you flatmap you filter the checkbox's stream to only return results when the checkbox is 'checked', which would mean the checkbox being 'unchecked' is not in the stream?
I was confused by this also, how does the observable get 'past' the filter?
FYI: since Rx 5, flatMapLatest has been renamed to switchMap. flatMap is now mergeMap though the flatMap alias is still around. See https://github.com/ReactiveX/rxjs/blob/master/MIGRATION.md
What would be the difference between using flatMapLatest vs flatMap in this instance?