RxJS buffer and delay until one other observable emit first time

1 General Idea - switching back and forth between buffered and unbuffered

Looks like you need to pause (buffer) historyStream$ in the beginning and then unpause it when deviceVersionModelStream$ emits. A way to pause/unpause streams is.

merge(
  source$.pipe(bufferToggle(pauseOn$, () => pauseOff$)),
  source$.pipe(windowToggle(pauseOff$, () => pauseOn$))
).pipe(mergeAll())

Also see: https://medium.com/@kddsky/pauseable-observables-in-rxjs-58ce2b8c7dfd

Your specific case

For your case this would be:

const versionModel$ = deviceVersionModelStream$.pipe(take(1));
merge(
  historyStream$.pipe(bufferToggle(of(0), v => versionModel$)),
  historyStream$.pipe(windowToggle(versionModel$, v => NEVER))
).pipe(
  mergeAll(),
  bufferTime(2000)
);

If you want to have the output of deviceVersionModelStream$ accessible on every emit you could use combineLatest. If multiple subscriptions to historyStream$ and deviceVersionModelStream$ are a problem you can use share beforehand.

const versionModel$ = deviceVersionModelStream$.pipe(share(), take(1));
const historyStreamShared$ = historyStream$.pipe(share());
combineLatest(
  versionModel$,
  merge(
    historyStreamShared$.pipe(bufferToggle(of(0), v => versionModel$)),
    historyStreamShared$.pipe(windowToggle(versionModel$, v => NEVER))
  ).pipe(
    mergeAll(),
    bufferTime(2000)
  )
).subscribe(console.log);

https://stackblitz.com/edit/rxjs-oyctsg

Edit (if you only need the buffer once at the beginning)

In your case you don't need to be able to use the buffer again once it's turned off, i.e. once you switch from the bufferToggle to the windowToggle stream you won't need to switch back to the bufferToggle stream.

This allows for a slightly simple approach using only buffer instead of bufferToggle and windowToggle.

2 General Idea - buffer only at the beginning

source$ = dataStream.pipe(share()) // make sure this is a hot observable
pauseOff$ = timer(5000) // make sure this observable emits once and completes

concat(
  source$.pipe(buffer(pauseOff$), mergeAll()), // start with a buffered stream
  source$ // switch to the unbuffered stream when pauseOff$ emits and completes the previous stream
)

Your specific case

const historyStream$ = timer(0, 100).pipe(share(), take(200));
const versionModel$ = of("version-model").pipe(delay(5000), take(1));

combineLatest(
  versionModel$,
  concat(
    historyStream$.pipe(buffer(versionModel$), mergeAll()),
    historyStream$
  ).pipe(
    bufferTime(2000)
  )
).subscribe(console.log);

https://stackblitz.com/edit/rxjs-2ptrux

This post is a little old, but I recently stumbled upon it while looking for a "buffer rxjs observables until another event occurred", and I thought I would share what an updated version would look like.

I built this on https://thinkrx.io. Here is the code I used:

const { rxObserver } = require('api/v0.3');
const { timer, of, combineLatest, concat } = require('rxjs');
const { delay, take, share, buffer, mergeAll, bufferTime, filter, takeUntil, skipUntil } = require('rxjs/operators');



const historyStream$ = timer(0, 10).pipe(
  share(), take(10)
);

const versionModel$ = of("A").pipe(
  delay(50), 
  take(1)
);

historyStream$.subscribe(rxObserver('History Stream'));
versionModel$.subscribe(rxObserver('Version Model'));

combineLatest([
  versionModel$,
  concat([
    historyStream$.pipe(buffer(versionModel$)),
    historyStream$.pipe(skipUntil(versionModel$))
  ]).pipe(
    mergeAll()
  )]
).subscribe(rxObserver("Buffer Window"));

Here is the output:

What's going on here is that we are passing two observables into concat: one that represents the buffered set of events, and on that represents the stream of events as they come in. We finally pipe all of that into mergeAll to get the appropriate effect.