Rxjs: Observable.empty() or Observable.of() stops a chain of observables

[giovannicandido]

I'm transforming a number of unknown observables to be executed in parallel before some other observable. For that I use forkJoin and concat. If some of the observables return Observable.empty() or Observable.of(), it stops one of the streams.

I think that Observable.empty could be used to sinalize something that should return immediately

RxJS version: 5.0.0-rc.5

Code to reproduce:

var Rx = require('rxjs/Rx');

let first = Rx.Observable.timer(10, 1000).map(r => {
    console.info("Source " + 1)
    return { source: 1, value: r };
}).take(4);
let second = Rx.Observable.timer(10, 1000).map(r => {
    console.info("Source " + 2)
    return { source: 2, value: r };
}).take(4);

let third = Rx.Observable.timer(10, 500).map(r => {
    console.info("Source " + 3)
    return { source: 3, value: r };
}).take(4);

let fourth = Rx.Observable.empty() // Change to Rx.Observable.of("")

let observables = Rx.Observable.forkJoin([first, second, fourth])
let started = new Date()
observables.concat(third).skip(1)
    .do(res => console.info(res)).subscribe(res => {
        let end = new Date()
        let seconds = end.getSeconds() - started.getSeconds()
        console.info("Take: " + seconds)
        console.log(res)
    });

Expected behavior:

first and second streams are executed before the execution of the third

Actual behavior:

none are executed before third

Additional information:

Changing the code of fourth to return Observable.of("") provide the expected behavior

For completeness, the real use is on a Interceptor like the following:

interface Interceptor {
   before(request: Request): Observable<any>
   after(response: Response): void
}

The user provides the interceptors to an Http class. The http class joins all before methods and wait then to execute before doing the http request and calling after in the end

But how the user could make a observable that return immediately, without doing anything? In other words, the user want to override aftermethod and leave before untouched. Without returning null

[paulpdaniels]

That is by design. forkJoin short-circuits if one of the streams completes without emitting a value, since conceptually you probably don't want to join values that don't exist.

I would suggest that if you have control over the interceptor join code that you just append defaultIfEmpty('NO_EFFECT') onto the end of streams, it will be transparent to all non-empty streams and it will give an event that can be easily filtered out later on for empty streams.

[giovannicandido]

Thank you @paulpdaniels
I will use your tip about defaultIfEmpty to protect the forkJoin, because in this use case the short-circuits is not desired. It prevents the http call to execute after all the before methods.

It works like a charm :-)

[samal-rasmussen]

Just a note: undefined works as a default value too. Forkjoin will not short-circuit if some observable produces an undefined value before completing.

[lock[bot]]

This thread has been automatically locked since there has not been any recent activity after it was closed. Please open a new issue for related bugs.