Different Pipe Based on Filter RxJS

Question

I have the following problem. I want to subscribe to an observable using the async pipe in Angular. But the data in the pipe can originate from different data sources and the pipe can look different depending on the data source.

The way it is implemented now does not work, because I override the first value of this.data$ and thus source A never gets subscribed to.

Is it possible to 'split' the pipe depending on a filter?

problem.component.ts:

// Data source A and logic A 
this.data$ = this.service.context.pipe(
  filter(context => context.flag === true),
  switchMap(context => this.dataSourceA(context.id))
  map(data => this.processDataA(data))
);

// Data source B and logic B 
this.data$ = this.service.context.pipe(
  filter(context => context.flag === false),
  switchMap(context => this.dataSourceB(context.id))
  map(data => this.processDataB(data))
);

problem.component.html

<pro-table [data]="data$ | async"></pro-table>

Answer 1

Move the logic to your switchMap()

this.data$ = this.service.context.pipe(
   switchMap(context => context.flag
      ? this.dataSourceA(context.id).pipe(map(d => this.processDataA(d)))
      : this.dataSourceB(context.id).pipe(map(d => this.processDataB(d)))
   )
);

Answer 2

That's just wrong way to achieve the result you desire. You have to take over to something called "higher order observable": Observable<Observable<T>>. Like a "higher order function" (that able to produce another function based on it's arguments), this guy is able to produce another Observable based on whatever parametrization you rely on.

this.service.context.pipe(
  switchMap(context => context.flag ? this.dataSourceA(context.id).pipe(...) : this.dataSourceB(context.id).pipe(...))
);

...it might seem a littble bit esoteric at the beginging, but actually it isn't. \

---

P.S. Important point to remember: higher order observables let you parametrize your streams if a fancy way: by dealing with streams of streams.

Answer 3

You can add a Subject which will catch many data sources and subscribe to that:

    this.data$ = new Subject();
    observer = {
     next(data) { this.data$.next(data) },
     error(msg) { /*handle error*/ }
   };
    // Data source A and logic A 
   this.service.context.pipe(
     filter(context => context.flag === true),
     switchMap(context => this.dataSourceA(context.id))
     map(data => this.processDataA(data))
   ).subscribe(observer);

   this.service.context.pipe(
     filter(context => context.flag === false),
     switchMap(context => this.dataSourceB(context.id))
     map(data => this.processDataB(data))
   ).subscribe(observer);

In this situation though it seems like you can use the flag to keep it in one chunk:

this.service.context.pipe(
     switchMap(context => context.flag ?
         this.dataSourceA(context.id) : this.dataSourceB(context.id)
    )
     map(data => this.processData(data)) // handle both cases
   ).subscribe(observer);