Throttled Buffering in RxJS
Capturing every event can get chatty. Batching events with a throttled buffer in RxJS lets you capture all of those events and use them responsibly without overloading a subscriber downstream.
[00:00] Let's talk about how to do a throttled buffer in RxJS. To get started, I've added my RxJS script to my HTML page. I've also added a button so we can get a stream of events, as well as a display area.
[00:16] In my JavaScript, I've set JS Bin to 6 to 5 or Babel so I have access to ES6 features like arrow functions. I've already selected my button from the DOM, and I've set up a stream of clicks from that button with Observable from event.
[00:34] I've also added this send values function. Really, all it does is stringifies whatever you send to it -- makes it into a JSON string -- and then, adds it to a pretag inside of my results div.
[00:51] Let's do something with this. I'm going to take my clicks stream, and I'm going to use scan to have it emit a string of incremented values. Then, I'm going to use for each to actually output that to my send values, the function I mentioned earlier.
[01:15] When I run this, I get one, two, three, four. If I click it really fast, I get a lot. Let's say, we're going to use our button clicks to send some sort of ajax event. Obviously, we would not want to send an ajax event for every button click, especially, if I can click them in rapid succession like this.
[01:35] One thing we can do with RxJS is use a buffer. A buffer works like this. It's an operator. It takes an Observable as the opening. What I mean by this is every time the Observable I pass into buffer emits a value, it will flush the buffer into for each.
[02:00] One thing that people try to do, first off, is just use some sort of interval to do this. In this case, I'll use an interval of about one second. That means that every one second it's actually going to emit an empty array. If I click this a bunch of times, you'll see the clicks got built up in there before it flushed.
[02:28] This isn't great, because we wouldn't want to send a bunch of empty arrays to our ajax end point. I could obviously filter it out. Now, you see there's nothing being sent unless I actually have something to send, but it's still less than ideal, because sometimes, I'm only sending one value.
[02:54] What I can do that's a little bit better than this is instead of using this interval, I can actually use the clicks Observable itself with a throttle. What this is going to do is emit a value. If I click this button multiple times and there's a delay of more than one second, it will finally emit a value.
[03:17] Now, what I've got is I'm buffering all of my clicks, but I'm saying if there's a delay of more than one second in between my clicks, flush the buffer. Let's try this out. Buffer sent, 1,001, buffer sent. That is a throttled buffer in RxJS.