Processing events in RxJS "one at a time"

user_96338 0
Viewed 728

I'm using RxJS to process an event stream. Quite a few events could be dispatched and the end processing code is "relatively expensive", eg. Load a file, do some work, save some stuff to a database.

const rx = require("rxjs")

// simulate a lot of events
const observable = rx.interval(1)

// subscribe to the events
const subscription = observable
    .subscribe(x => {
        // do expensive things here
    });

I want the code in subscribe to operate "one at a time" (or potentially in the future, "N at a time"), so that the event handler code fully completes before the next event starts processing. Right now what seems to happen is the code in subscribe gets called over and over again for each event, so there are many many event being processed at the same time.

I've tried a couple of different things (mainly inside of pipe as an experiment). rx.operators.throttle throws away events, which is not what I want - I just want to hold them up. Using flatMap with a async functions, just delays the start of the stream being processed.

const _worker = x => {
    return new rx.Observable(subscriber => {
        setTimeout(() => {
            subscriber.next(x * 5)
            subscriber.complete()
        }, 5 * 1000);
    })
}

const subscription = observable
    .pipe(
        rxops.flatMap(_worker)
    )
    .subscribe(x => {
    });

So I'm stuck here now. Is this possible, or am I just trying to use RxJS in some way it's not intended to be?

rxjs node.js javascript
2 Answers

You want concatMap (https://www.learnrxjs.io/learn-rxjs/operators/transformation/concatmap).

concatMap will map the value emitted from a source Observable to a value emitted by some inner Observable, and it won't move on to the next value from the source until the inner Observable completes.

const source = rx.interval(1);

source.pipe(
    concatMap(val => {
        // Do expensive (async) work here and return an Observable.
        // Next value from source won't be processed until this inner Observable
        // completes (and no values from source will be skipped). 
    })
).subscribe()
user_110471220
user_100925340
Related