Dynamically add operator to Rxjs observable in Angular application

Question

In our Angular application we have a loader component that receives an observable through an input setter property.

Inside the setter we first set a boolean isLoading to true which inside the template starts showing a loading spinner. Then we subscribe to the observable, and when data is received the isLoading boolean is set to false again making the spinner disappear:

// loading-component template:

<ng-container *ngIf="data; else errorOrLoading">
  ...
</ng-container>

<ng-template #errorOrLoading>
 ...
</ng-template>

// loading-component class (short version without unsubscribe):

  @Input() set data$(data$: Observable<any>) {
    if (data$) {
      this.data = null;
      data$
        .subscribe(data => {
          this.data = data;
        });
    }
  }

This works great if we only have one event from the Observable. But when the Observable emits multiple events, the isLoading won't be set to true again because the setter is not called.

Is there a way to dynamically add an extra tap operator that will allow to set this.data = null before the given Observable chain starts emitting a new event?

So if for instance the Observable is:

myService.onChanges$.pipe(
  switchMap(() => this.getBackendData())
);

Can we dynamically add a tap operator that will change the pipe to:

myService.onChanges$.pipe(
  tap(() => this.data = null),
  switchMap(_ => this.getBackendData())
);

Update: I have chosen to simplify the loader control and to move all observable related logic to services, which feels way more scalable and flexible.

Answer 1

Update #1

Solution #1. Use shareReplay()

As per @Reactgular answer

All you have to do is use a shareReplay() operator:

class MyService {
    public data$: Observable<any>;
    public loaded$: Observable<boolean>;

    constructor(private dataService: DataService) {
        this.data$ = this.dataService.loadData().pipe(
            startWith(null), // data$ emit a null before data is ready followed by the API data.
            shareReplay(1);
        );
        this.loaded$ = this.data$.pipe(
           mapTo(true),
           startWith(false)
        );
    }
}

You have to call myService.data$.subscribe() to trigger the first reading of the stream to make the data ready. You can do that in the constructor, but keep in mind that Angular doesn't create a service until it is first used. If you want the data to be eagerly loaded, then use a resolver in a route or inject the service into a NgModule constructor and subscribe there.

---

Solution #2. Use dedicated service

The better solution can be to introduce a LoaderService which will handle loading of a component/view data.

Depending on your project need it can be singleton or shared.

Let's assume that our service will handle loading state only for the current view (shared service)

Loader Service:

export class LoaderService {
  private  readonly  _loading: BehaviorSubject<boolean> = new BehaviorSubject<boolean>(false)

  // returns value of `_loading` on the moment of accessing `isLoading`
  get isLoading(): boolean {
    return this._loading.getValue();
  }
  
  // returns value of `_loading` as Observable
  get isLoading$(): Observable<boolean> {
    return this._loading.asObservable();
  }
  
  // pushes `true` as a new value of `_loading` subject and notifies subscribers
  start(): void {
    this._loading.next(true);
  }
  
  // pushes `true` as a new value of `_loading` subject and notifies subscribers
  stop(): void {
    this._loading.next(false);
  }
  
}

Assuming we have two services:

• API - only contains declaration of methods that return pure (non-modified) http streams
• ComponentService - service that prepares data before passing it to representation component

MyFruitService

  constructor(
   ..., 
   private api: MyFruitAPIService,
   private loader: LoaderService
  ) { ... }

  getApples(): Observable<T> {
    this.loader.start();
    
    return this.api.getApples()
     .pipe(
      finalize(() => this.loader.stop()) // finalize - will call a function when observable completes or errors
     );
  }
   

MyAppleFruitComponent

 readonly loading$ = this.loader.isLoading$.pipe(tap((loading) => {
   if (loading) {
    this.data = null;
   }
 }));

 constructor(private loader: LoaderService) { ... }

 <ng-container *ngIf="loading$ | async; else errorOrLoading">

 ...

 </ng-container>

Answer 2

In short, no. This isn't possible. It's also probably not a great idea either as it breaks encapsulation to do so.

You can't dynamically change a function call either. If doADance() dances for you, you can't really dynamically make it add a list of numbers as well. The implementation of a function should remain separate from its invocation. Though to your point, Javascript does actually have people making functions do strange things dynamically by binding different contexts and such.

RxJS keeps implementation separate from invocation (subscription) for streams as well. If a library does 20 transformations and returns that stream to you, you're not really being handed a list of transformations, that's just implementation details that the library writers could alter without introducing breaking changes.

Update 1:

True, encapsulation is important and exists for a reason. However, we of course cán dynamically add list of numbers to doADance() by passing the list as a parameter. Maybe a controlled way of allowing sort of placeholders inside the pipe to be filled with dynamically given operators would kind of be the same?

While placeholders in the pipe don't really make sense since any pipeable operator is easily turned into a static operator and any set of operators is easily turned into a single operator.

What you can do is something extremely close. This doesn't work with a stream being returned from a library, for example, but you can design your streams to allow for customization in how they're processed.

In your service:

function dataOnChange(): Observable<Data>{
  return myService.onChanges$.pipe(
    switchMap(() => this.getBackendData())
  );
}

function dataOnChangePlus(op): Observable<Data>{
  if(op == null) return dataOnChange();
  return myService.onChanges$.pipe(
    op,
    switchMap(() => this.getBackendData())
  );
}

elsewhere:

this.service.dataOnChangePlus(
  tap(_ => this.data = null)
).subscribe(console.log);

somewhere else, doing the same thing but a bit differently:

this.service.dataOnChangePlus(
  st => st.pipe(
    mapTo(null),
    tap(val => this.data = val)
  )
).subscribe(console.log);

Now the consumer of dataOnChangePlus is being returned a stream and also gets to help define how that stream is constructed. It's not dynamically adding an operator, but it does allow you to defer the definition of the operator.

The benefit is that each call can define something different.

If you want, you can narrow what a caller can do by only giving them access to a specific type of operator. For example, only let them define the lambda for a tap operator:

function dataOnChangePlusTap(tapper): Observable<Data>{
  return myService.onChanges$.pipe(
    tap(tapper),
    switchMap(() => this.getBackendData())
  );
}

this.service.dataOnChangePlusTap(
  _ => this.data = null
).subscribe(console.log);