Capacitor plugins with Angular and Rxjs - Stream capacitor plugin in NgZone context

Question

I’m using the CapacitorPurchases plugin (but any plugin can have this situation) and methods return promises with results. Normally I use something like

const result = await CapacitorPurchases.getPurchaserInfo();
this.ngZone.run(()=>{
   // do something with result
})

now I have a situation where I prefer to use rxjs and “complex” stream pipes with switchMap and other things. If I write

from( from(CapacitorPurchases.getPurchaserInfo()))
 .pipe(switchMap(value=>{
  this.a = value;
  return this.httpService.call()
})

the content of switchMap is run outside of the ngZone context but if I write

from( from(CapacitorPurchases.getPurchaserInfo()))
 .pipe(switchMap(value=>{
  this.ngZone.run(()=>{
   this.a = value;
  })
 return this.httpService.call()
})

or other combination, I can lose data.
Do you know if there is a way to create a zone context with a stream pipe?

Answer 1

After seeing and trying the answer postred @Džan Operta the best solution in to implement a pipe that re-enter in the ngZone.

I quote the code from this answer:

import { Observable, OperatorFunction } from 'rxjs';
import { NgZone } from '@angular/core';

export function runInZone<T>(zone: NgZone): OperatorFunction<T, T> {
  return (source) => {
    return new Observable(observer => {
      const onNext = (value: T) => zone.run(() => observer.next(value));
      const onError = (e: any) => zone.run(() => observer.error(e));
      const onComplete = () => zone.run(() => observer.complete());
      return source.subscribe(onNext, onError, onComplete);
    });
  };
}

The other solution in the answer (rxjs-zone-operators plugin) seems loosing the type of the data in the pipe chain.