RxJS Subject/Observable issue

Question

I'm trying to convert an Angular function to the observable pattern since it's current implementation has some asynchronicity to it. For the purposes of this discussion, take this simple example.

aFunction(x: boolean){
    if(x){
        // ... do something asynchronous based on the value of x
    }
}

Converting it to use an Observable could be done in this manner:

anObservableFunction(x: boolean): Observable<any> {
    const result = new Subject();
    if(x){
        // ... do something asynchronous based on the value of x
        // ... where once the actual value you want to return to 
        // the subscribing functions, you can pass in the 
        // result.next(value);
        // result.complete();
    }
    return result.asObservable();
}

The issue I'm facing (from my understanding) is catering for the instance where the inner selection statement isn't accessed.

anObservableFunction(x: boolean): Observable<any> {
    const result = new Subject();
    if(x){
        // ... do something asynchronous based on the value of x
        // ... where once the actual value you want to return to 
        // the subscribing functions, you can pass in the 
        // result.next(value);
        // result.complete();
    } else {
        // here
    }
    return result.asObservable();
}

If a regular Subject is used, surely the subscribing functions will not get any value since the order of events would be:

Function is called
Subject is created
Value is set
Calling function subscribes, thus only getting values after this event occurring

And if a BehaviorSubject or ReplaySubject are used, their initial/constructed value will be retained, causing a subscription event to fire unnecessarily?

ReplaySubject doesn't have an initial value, so it would work. — Ingo Bürk
@IngoBürk ah you're right - I had it in my mind that it had an initial value upon initialization too. I suppose in this scenario there will only be one value ever returned. Would an AsyncSubject not work here too? — jarodsmk
Yes, AsyncSubject would also work, but restricts you to a single emission. By the way, another solution would be to delay the emission by a tick, eg by emitting on the subject within a setTimeout(0). — Ingo Bürk
Lastly, you could also return Observable.defer(() => x ? asyncOperation() : syncObservable()) — Ingo Bürk
Brillant. Would you mind putting these into a solution? Will gladly accept it. Thanks! — jarodsmk

Ingo Bürk Ingo Bürk · Accepted Answer · 2018-05-24T05:49:24

You are correct that using a Subject has an issue if the values are emitted synchronously. The BehaviorSubject has the downside of having an unnecessary initial value, but the ReplaySubject actually doesn't have that and would work, but it would also replay this value if someone subscribes way later on which you may not want.

A simple trick would be to delay the synchronous emission by a tick:

setTimeout(() => result$.next(42), 0);

However, you can also return an observable directly and avoid using the subject:

foo(x) {
  if(x) {
    return callRemoteEndpoint();
  }

  return of(42);
}

RxJS Subject/Observable issue

1 Answers