RxJS: making a buffer emit after certain event

Question

Problem: I would like to create Rxjs composition ( function chain ) that would result in buffering values from one Observable until certain event occurs then emmiting all buffered values synchroniously and then buffering until next event.

I use this function to gather all http requests that have to wait until my application make a call for authorization. And then run all those requests. ( Its implemented inside Angular4 HttpClient Interceptor ), thats my usecase but I generally seek solution on how to create such rx chain.

Why Rxjs buffer is not enoyugh. From what I read and tested buffer either requires exact time frames, or in case of getting a scheduler instead of time as parameter, its resubscribes to scheduler after detecting last scheduler`s "event" propagation. And I would like it work like this: When there appears first request I start to buffer and then subscribe to scheduler, after scheduler emits, a stop buffering, reemit all buffered values, and wait until next new request is made to start buffering again and to launch scheduler again.

Right now my solution is using helper object thats either undefined or my observable, with code roughly as follows:

private observable: Observable<boolean>;

makeRequest(): Observable<boolean> {
    if (this.observable !== void 0) {
        return this.observable;
    } else {
        this.observable = this.authenticationReuqest()
            .share()
            .finally(() => this.observable = void 0);

        return this.observable;
    }
}

In that way I kind of buffer my requests, by maiking them .delay() until same multicasted observable emits, and after it emits I just clean it up (though no need for unsubscribing as its cleans up on finally so after completing or erroring).

If anyone has an idea or patter how to replace this solution with pure Rxjs I am intrested. I have some feeling that some combination of Buffer and Zip could make it happen though I cant get the exact solution.

Thanks Tomasz

Answer 1

You should probably use bufferWhen: it buffers value from an observable until a second one emits.

Answer 2

I ran into exactly the same problem, I have events coming in like this

I kick off loading events early so that we don't need to wait. But the problem is, only after one point (the "rdy", stands for ready) I want to process these events. So the ready event needs to come first. I need to hold the values until ready comes and reemit them. Like this

Here's what I do, I use multicast to share the same subscription in downstream. Create a ready observable in the selector function filtering the first ready event. Then merge the ready observable, and buffer of the multicasted events, take one, and switchMap back from array to values. Finally concat the multicasted events. Here's the code

const events = [
  'ev1',
  'ev2',
  'ready',
  'ev3',
  'ev4',
  'ev5'
]
Rx.Observable
  .interval(500)
  .take(events.length)
  .map(i => events[i])
  .multicast(
      () => new Rx.Subject(),
      events$ => {
          const ready$ = events$
            .filter(event => event === 'ready')
            .take(1)
          return Rx.Observable
            .merge(
                ready$,
                events$
                  .buffer(ready$)
                  .take(1)
                  .switchMap(events => Rx.Observable.from(events))
            )
                .concat(events$)
      }
  )

Or you can find it and run it on RxViz here https://rxviz.com/v/j8nNpe8N It's not pretty, but it works. I am still looking for a better way to do it, any idea?

Answer 3

I also spent time looking into this exact issue, and I couldn't figure it out. In the end, I got the following to work. I feel confident there must be a more idiomatic / elegant approach, but I am able to understand this approach as a new RxJS user.

import { fromEvent } from "rxjs";
import { tap, scan, filter, map } from "rxjs/operators";

const clicks = fromEvent(document, "click").pipe(
  scan(acc => acc + 1, 0),
  tap((v) => console.log("tapping", v))
);

const out = clicks.pipe(
  // tap(v => console.log('before scan', v)),
  scan((acc, count) => {
    if (count % 3 === 0) {
      return {acc: [], out: acc.acc.concat(count)}
    }
    return {out: [], acc: acc.acc.concat(count)}
  }, {acc: [] as number[], out: [] as number[]}),
  // tap(v => console.log('after scan', v)),
  filter(obj => obj.out.length > 0),
  map(obj => obj.out)
)

out.subscribe(v => console.log('out', v))

The key is that I used scan() as a running reducer and then filter out values. I build my own "buffer" of values from each subsequent value in the .scan() and then remove any events which don't have the "ready" state in the next filter.

Seems ugly, but allows for my use case which is a bit more complex than ready events. I'm parsing git-remote-helper commands, some which end with a double newline and some which don't.