RXJS implement blocking for observable

Question

I am new on RxJS and angular2, my goal is develop a shared function to prevent racing between model update and saving data to database without using blockui

My rough idea is:

initial state is "unfreeze"

1. Parent send "freeze" to children
2. Child components "block" all their internal dom event queues
3. Parent wait for all outstanding async operations (including 
    children) to complete
4. Parent save model data into database
5. Parent send "unfreeze" to children, and child component unblock their queues

but do not know how it can be done in RxJS world

(observable) signal -> "freeze"/"unfreeze" from parent component

(observable) target queue ->

When "freeze", either (choice of children component)

    a. buffer all events (e.g. keystroke)
    b. drop all events (e.g. mouse click)
    c. keep last event (e.g. windows resize)

    until "unfreeze"

When "unfreeze", no buffering or filtering, forward event ASAP, until "freeze"

Thank you very much

You need to be more specific on what exactly you need or what you are trying to achieve. Based on what you asked, there are many different operators you could use. (buffer, debounce, filter, takeWhile, skipWhile, switchMap, exhaustMap, ...and more...) — olsn, May 17 '17 at 09:02
see http://stackoverflow.com/questions/43912849/conditional-emission-delays-with-rxjs/43913932#43913932 — Ishan Soni, May 17 '17 at 12:24
You can use [bufferWhen](http://reactivex.io/rxjs/class/es6/Observable.js~Observable.html#instance-method-bufferWhen) or bufferToggle. — thatseeyou, May 17 '17 at 12:38

Lukas Neicelis · Answer 1 · 2017-05-17T17:54:28.370

1

If you want to skip values while signal is "freeze"

const signal$ = Rx.Observable.interval(1000).map(i => i % 2);
const target$ = Rx.Observable.interval(500);
const unfreeze$ = signal$.filter(even => !!even);
const freeeze$ = signal$.filter(even => !even);

target$
  .windowToggle(unfreeze$, () => freeeze$)
  .switch() 
  .subscribe(val => console.log(val));

<script src="https://cdnjs.cloudflare.com/ajax/libs/rxjs/5.4.0/Rx.min.js"></script>

If you want to buffer values while it's freezed

const signal$ = Rx.Observable.interval(1000).map(i => i % 2);
const target$ = Rx.Observable.interval(500).publish();
const unfreeze$ = signal$.filter(odd => odd);

target$.connect();

signal$
  .concatMap(freeze => {
    return freeze
      ? target$.takeUntil(unfreeze$).toArray()
      : target$.takeUntil(signal$);
  })
  .subscribe(val => console.log(val));

<script src="https://cdnjs.cloudflare.com/ajax/libs/rxjs/5.4.0/Rx.min.js"></script>

edited May 17 '17 at 17:54

answered May 17 '17 at 14:18

Lukas Neicelis

438
3
6

instead of .switchMap(branch$ => branch$) you can just do .switch() – Julia Passynkova May 17 '17 at 17:37
Thanks! I updated the snippet :) ant more thoughts on how to optimize it? ;) – Lukas Neicelis May 17 '17 at 17:55
I agree with windowToggle. Why not to use would say use bufferToggle to buffer? – Julia Passynkova May 17 '17 at 19:22
If the signal queue do not send any more "freeze" signal after "unfreeze", then the target queue will be "buffered" forever? – user8024165 May 18 '17 at 00:33

score 0 · Answer 2 · answered May 17 '17 at 19:24

You can use bufferToggle

http://reactivex.io/rxjs/class/es6/Observable.js~Observable.html#instance-method-bufferToggle

Collects values from the past as an array. Starts collecting only when opening emits, and calls the closingSelector function to get an Observable that tells when to close the buffer.

const start$ = Rx.Observable.fromEvent(document.getElementById('start'), 'click');

const stop$ = Rx.Observable.fromEvent(document.getElementById('stop'),  'click');

const clock$ = Rx.Observable.interval(1000);

clock$
  .bufferToggle(start$, ()=>stop$)
  .mergeAll()
  .subscribe(x=>console.log(x))

it seems that "clock" signal cannot pass through until clicking "stop"? — user8024165, May 18 '17 at 00:41

RXJS implement blocking for observable

2 Answers2