Aggregating Streams With Reduce And Scan using RxJS
What is the RxJS equivalent of Array reduce? What if I want to emit my reduced or aggregated value at each event? This brief tutorial covers Observable operators reduce() and scan(), their differences and gotchas.
[00:00] When dealing with sets of data, it's often desirable to reduce that set of data to a single composite value. To demonstrate this, I'm going to use the Array's reduce method. Array.reduce simply takes an argument of a function that is my reduced value, along with an item from the Array. You return a new reduced value. Finally, you can initialize the reduced value with the last argument.
[00:35] What this is going to do is it's going to pass zero and the first value from my Array, which is also zero, add them together, and then pass whatever it's added together in as the next argument, along with the next value in the Array, and add those together.
[00:53] We can see what the result is. I'm going to output the results in console. We can see the result is 10, 10 being 01 + 2 + 3 + 4. Let's try the same thing with Observable.
[01:12] Observable actually also has the reduce method. If I take my Array and turn it into an Observable, we can do the same thing, but since Observables can be asynchronous, we can get our result this way. We need to get our result by subscribing.
[01:48] Clear and run it again. We get exactly the same answer.
[01:52] As I stated before, Observables can be asynchronous. Let's try this another way. Let's take an interval. We'll only take the first five values.
[02:05] An interval is going to output zero, one, two, but it's going to do it once every tenth of a second. I'm going to run this. There's a slight delay, and then it outputs the number 10.
[02:18] I want to show you something. If I make this one second and I run this -- 1,001, 1,002, 1,003, 1,004. There's my response. You'll see it actually waited for the entire Observable to complete before the result of the reduce was emitted. This is important to remember.
[02:39] Reduce on an Observable waits for completion of the Observable before it gives you the result. That means if I was to get rid of this take and I have an Observable that will run forever, like this interval, I will never get a response back from reduce.
[02:57] What if I wanted to see the result of my reduction at each tick from my source? To do that, we use scan. Scan works exactly the same way as reduce, with the exception that its initializer comes first.
[03:19] Now, when I run this, you'll see each time it ticks in, I get the next value, the next reduced value. One nice difference with scan though is, since it doesn't wait for completion, if I were to run this again, it's actually going to give me a result every time.