Buffer until with redux-observable and rxjs

Question

Is there a pattern for making an redux-observable epic buffer until a value in store is true?

const uploadsCompleteEpic = (action$, store) => {
  return action$
    .ofType(actionTypes.UPLOAD_SUCCESS)
    .bufferWhen(store => store.getState().remaining -1 === 0)
    .do(action => console.log(action))
    .ignoreElements();
};

The above doesn't work because the function in bufferWhen is not an Observable. I've also tried wrapping the buffer function in Observable.single with no success.

jayphelps jayphelps · Accepted Answer · 2018-02-27T06:52:25

You can use the regular buffer operator for this.

public buffer(closingNotifier: Observable<any>): Observable<T[]>

Collects values from the past as an array, and emits that array only when the closingNotifier Observable emits.

const uploadsCompleteEpic = (action$, store) => {
  // just used as a notifier, so we don't actually
  // care about what the value is, just when.
  // Since the state is only ever updated when an
  // action is received we can use that time to check.
  const notification$ = action$.filter(() => {
    return store.getState().remaining - 1 === 0;
  );

  return action$
    .ofType(actionTypes.UPLOAD_SUCCESS)
    .buffer(notification$)
    // Use this if you want to exclude empty buffers
    // .filter(buffer => buffer.length > 0) 
    .do(action => console.log(action))
    .ignoreElements();
};

I don't know under what exact circumstances the remaining value changes or is zero, so it's definitely possible you'll need to do additional checking or waiting so that things don't race. e.g. exclude empty buffers, if that ends up being possible.

Buffer until with redux-observable and rxjs

1 Answers