How to use an Angular service inside an RXJS Custom Operator?

Question

How do I use an Angular service in my custom rxjs operator?

Is it possible to do this?

function myOperator() {
    return function <T>(source: Observable<T>): Observable<T> {
        return new Observable(subscriber => {
            const subscription = source.subscribe({
                next(value) {
                    //access an Angular service HERE
                    subscriber.next(value);
                },
                error(error) {
                    subscriber.error(error);
                },
                complete() {
                    subscriber.complete();
                }
            });
            return () => subscription.unsubscribe();
        });
    };
}

I'd like to use it in an observable pipe:

observable
.pipe(
    myOperator()
)
.subscribe(result => {

});

Answer 1

Creating and registering an injector, bootstrapping it and using it in the custom operator seems to work well, without having to use a service.

export class RootInjector {
  private static rootInjector: Injector;
  private static readonly $injectorReady = new BehaviorSubject(false);
  readonly injectorReady$ = RootInjector.$injectorReady.asObservable();

  static setInjector(injector: Injector) {
    if (this.rootInjector) {
      return;
    }

    this.rootInjector = injector;
    this.$injectorReady.next(true);
  }

  static get<T>(
    token: Type<T> | InjectionToken<T>,
    notFoundValue?: T,
    flags?: InjectFlags
  ): T {
    try {
      return this.rootInjector.get(token, notFoundValue, flags);
    } catch (e) {
      console.error(
        `Error getting ${token} from RootInjector. This is likely due to RootInjector is undefined. Please check RootInjector.rootInjector value.`
      );
      return null;
    }
  }
}

This can be registered during the bootstrapping:

platformBrowserDynamic.bootstrapModule(AppModule).then((ngModuleRef) => {
  RootInjector.setInjector(ngModuleRef.injector);
});

And then be used in the custom operator:

function myOperator() {
    const myAngularService = RootInjector.get(MyAngularService);
    return function <T>(source: Observable<T>): Observable<T> {
        return new Observable(subscriber => {
            const subscription = source.subscribe({
                next(value) {
                    myAngularService.doMyThing();
                    subscriber.next(value);
                },
                error(error) {
                    subscriber.error(error);
                },
                complete() {
                    subscriber.complete();
                }
            });
            return () => subscription.unsubscribe();
        });
    };
}

This causes tests to crash because RootInjector is not set up. But placing this in root-injector.mock.ts file:

import { TestBed } from '@angular/core/testing';
import { RootInjector } from 'src/app/core/injectors/root-injector';

RootInjector.setInjector({
    get: (token) => {
        return TestBed.inject(token);
    }
});

..and then importing it into the jasmine test file did the trick:

import 'src/mocks/root-injector.mock';

describe('MyComponent', () => {
    ...
}

Note that this only works for services providedIn: 'root'

Thanks to this post!: https://nartc.netlify.app/blogs/root-injector/

Answer 2

Of course it is possible. However, I recommend modifying your operator a bit. Rxjs offers a special interface that defines a custom operator, thanks to which you can freely process your data stream. This interface is called OperatorFunction and consists of two parameters - the first specifies the input type and the second the output type.

In addition, it is worth remembering that the operator is not an observer, so you should probably not define the next, error and complete methods because you are returning a new data stream. The Observer reacts to the values in the stream and is used to observe the stream values, not processing - that's what the operator is for.

Some example:

@Injectable()
export class MyOperator {
  constructor(private readonly myAngularService: MyAngularService) {}

  public myOperator(): OperatorFunction<number, string> {
    return (source: Observable<number>) => {
      return source.pipe(
        map((value: number) => {
          const newValue = this.myAngularService.yourServiceFunction(value);
          return `Your new transformed value: ${newValue}`;
        })
      );
    };
  }
}

Answer 3

Why not simply pass the service as argument to the operator?

function myOperator(myService) {
  return function <T>(source: Observable<T>): Observable<T> {
     ...here you can use "myService"..
}

//and use as:

observable
.pipe(
    myOperator(this.myService)
)
.subscribe(result => {

});