β Be wary of backpressure when the source emits at a faster pace than inner observables complete!
π‘ In many cases you can use concatMap as a single operator instead!
ββββ
( example tests )
Example 1: concatAll with observable
( StackBlitz | jsBin | jsFiddle )
// RxJS v6+import { map, concatAll } from 'rxjs/operators';import { of, interval } from 'rxjs';β//emit a value every 2 secondsconst source = interval(2000);const example = source.pipe(//for demonstration, add 10 to and return as observablemap(val => of(val + 10)),//merge values from inner observableconcatAll());//output: 'Example with Basic Observable 10', 'Example with Basic Observable 11'...const subscribe = example.subscribe(val =>console.log('Example with Basic Observable:', val));
Example 2: concatAll with promise
( StackBlitz | jsBin | jsFiddle )
// RxJS v6+import { map, concatAll } from 'rxjs/operators';import { interval } from 'rxjs';β//create and resolve basic promiseconst samplePromise = val => new Promise(resolve => resolve(val));//emit a value every 2 secondsconst source = interval(2000);βconst example = source.pipe(map(val => samplePromise(val)),//merge values from resolved promiseconcatAll());//output: 'Example with Promise 0', 'Example with Promise 1'...const subscribe = example.subscribe(val =>console.log('Example with Promise:', val));
Example 3: Delay while inner observables complete
( StackBlitz | jsBin | jsFiddle )
// RxJS v6+import { take, concatAll } from 'rxjs/operators';import { interval, of } from 'rxjs';βconst obs1 = interval(1000).pipe(take(5));const obs2 = interval(500).pipe(take(2));const obs3 = interval(2000).pipe(take(1));//emit three observablesconst source = of(obs1, obs2, obs3);//subscribe to each inner observable in order when previous completesconst example = source.pipe(concatAll());/*output: 0,1,2,3,4,0,1,0How it works...Subscribes to each inner observable and emit values, when complete subscribe to nextobs1: 0,1,2,3,4 (complete)obs2: 0,1 (complete)obs3: 0 (complete)*/βconst subscribe = example.subscribe(val => console.log(val));
βProgress Barβ
βconcatAll π° - Official docs
βFlatten a higher order observable with concatAll in RxJSβ
π₯ π΅ - AndrΓ© Staltz
π Source Code: https://github.com/ReactiveX/rxjs/blob/master/src/internal/operators/concatAll.tsβ