feat: use isObervable from rxjs instead of our own version

Relying on `isObservable` type guard from RxJS is better as it has less
false-positives.

Author: micalevisk

packages/core/helpers/external-context-creator.ts

@@ -6,7 +6,7 @@ import {
   PipeTransform,
 } from '@nestjs/common/interfaces';
 import { isEmpty, isFunction } from '@nestjs/common/utils/shared.utils';
-import { lastValueFrom } from 'rxjs';
+import { lastValueFrom, isObservable } from 'rxjs';
 import { ExternalExceptionFilterContext } from '../exceptions/external-exception-filter-context';
 import { FORBIDDEN_MESSAGE } from '../guards/constants';
 import { GuardsConsumer } from '../guards/guards-consumer';
@@ -329,7 +329,7 @@ export class ExternalContextCreator {
   }
 
   public async transformToResult(resultOrDeferred: any) {
-    if (resultOrDeferred && isFunction(resultOrDeferred.subscribe)) {
+    if (isObservable(resultOrDeferred)) {
       return lastValueFrom(resultOrDeferred);
     }
     return resultOrDeferred;

packages/microservices/context/rpc-proxy.ts

@@ -1,6 +1,5 @@
-import { isFunction } from '@nestjs/common/utils/shared.utils';
 import { ExecutionContextHost } from '@nestjs/core/helpers/execution-context-host';
-import { Observable } from 'rxjs';
+import { Observable, isObservable } from 'rxjs';
 import { catchError } from 'rxjs/operators';
 import { RpcExceptionsHandler } from '../exceptions/rpc-exceptions-handler';
 
@@ -12,7 +11,7 @@ export class RpcProxy {
     return async (...args: unknown[]) => {
       try {
         const result = await targetCallback(...args);
-        return !this.isObservable(result)
+        return !isObservable(result)
           ? result
           : result.pipe(
               catchError(error =>
@@ -34,8 +33,4 @@ export class RpcProxy {
     host.setType('rpc');
     return exceptionsHandler.handle(error, host);
   }
-
-  isObservable(result: any): boolean {
-    return result && isFunction(result.subscribe);
-  }
 }

packages/microservices/server/server.ts

@@ -1,10 +1,10 @@
 import { Logger, LoggerService } from '@nestjs/common/services/logger.service';
 import { loadPackage } from '@nestjs/common/utils/load-package.util';
-import { isFunction } from '@nestjs/common/utils/shared.utils';
 import {
   connectable,
   EMPTY as empty,
   from as fromPromise,
+  isObservable,
   Observable,
   of,
   Subject,
@@ -113,7 +113,7 @@ export abstract class Server {
       );
     }
     const resultOrStream = await handler(packet.data, context);
-    if (this.isObservable(resultOrStream)) {
+    if (isObservable(resultOrStream)) {
       const connectableSource = connectable(resultOrStream, {
         connector: () => new Subject(),
         resetOnDisconnect: false,
@@ -125,10 +125,13 @@ export abstract class Server {
   public transformToObservable<T = any>(resultOrDeferred: any): Observable<T> {
     if (resultOrDeferred instanceof Promise) {
       return fromPromise(resultOrDeferred);
-    } else if (!this.isObservable(resultOrDeferred)) {
-      return of(resultOrDeferred);
     }
-    return resultOrDeferred;
+
+    if (isObservable(resultOrDeferred)) {
+      return resultOrDeferred as Observable<T>;
+    }
+
+    return of(resultOrDeferred);
   }
 
   public getOptionsProp<
@@ -180,10 +183,6 @@ export abstract class Server {
       new IncomingRequestDeserializer();
   }
 
-  private isObservable(input: unknown): input is Observable<any> {
-    return input && isFunction((input as Observable<any>).subscribe);
-  }
-
   /**
    * Transforms the server Pattern to valid type and returns a route for him.
    *

packages/microservices/test/context/rpc-proxy.spec.ts

@@ -39,17 +39,4 @@ describe('RpcProxy', () => {
       (await proxy(null, null)).subscribe(null, () => expectation.verify());
     });
   });
-
-  describe('isObservable', () => {
-    describe('when observable', () => {
-      it('should return true', () => {
-        expect(routerProxy.isObservable(of('test'))).to.be.true;
-      });
-    });
-    describe('when not observable', () => {
-      it('should return false', () => {
-        expect(routerProxy.isObservable({})).to.be.false;
-      });
-    });
-  });
 });

packages/websockets/context/ws-proxy.ts

@@ -1,6 +1,5 @@
-import { isFunction } from '@nestjs/common/utils/shared.utils';
 import { ExecutionContextHost } from '@nestjs/core/helpers/execution-context-host';
-import { empty } from 'rxjs';
+import { empty, isObservable } from 'rxjs';
 import { catchError } from 'rxjs/operators';
 import { WsExceptionsHandler } from '../exceptions/ws-exceptions-handler';
 
@@ -12,7 +11,7 @@ export class WsProxy {
     return async (...args: unknown[]) => {
       try {
         const result = await targetCallback(...args);
-        return !this.isObservable(result)
+        return !isObservable(result)
           ? result
           : result.pipe(
               catchError(error => {
@@ -35,8 +34,4 @@ export class WsProxy {
     host.setType('ws');
     exceptionsHandler.handle(error, host);
   }
-
-  isObservable(result: any): boolean {
-    return result && isFunction(result.subscribe);
-  }
 }

packages/websockets/test/context/ws-proxy.spec.ts

@@ -1,5 +1,5 @@
 import { expect } from 'chai';
-import { of, throwError } from 'rxjs';
+import { throwError } from 'rxjs';
 import * as sinon from 'sinon';
 import { WsProxy } from '../../context/ws-proxy';
 import { WsException } from '../../errors/ws-exception';
@@ -40,17 +40,4 @@ describe('WsProxy', () => {
       (await proxy(null, null)).subscribe(null, () => expectation.verify());
     });
   });
-
-  describe('isObservable', () => {
-    describe('when observable', () => {
-      it('should return true', () => {
-        expect(routerProxy.isObservable(of('test'))).to.be.true;
-      });
-    });
-    describe('when not observable', () => {
-      it('should return false', () => {
-        expect(routerProxy.isObservable({})).to.be.false;
-      });
-    });
-  });
 });