refactor: remove deprecated apis from 3rd-party deps

Replacie old and deprecated APIs from `rxjs`, `body-parser`
and NodeJS core (read this guide https://nodejs.org/en/docs/guides/buffer-constructor-deprecation).
Do notice that the later only touches test files, thus doesn't affect
production code.

integration/microservices/src/grpc-advanced/advanced.grpc.controller.ts

@@ -134,8 +134,8 @@ export class AdvancedGrpcController {
   async streamReq(messages: Observable<any>): Promise<any> {
     const s = new Subject();
     const o = s.asObservable();
-    messages.subscribe(
+    messages.subscribe({
-      msg => {
+      next: () => {
         s.next({
           id: 1,
           itemTypes: [1],
@@ -146,9 +146,8 @@ export class AdvancedGrpcController {
           },
         });
       },
-      null,
+      complete: () => s.complete(),
-      () => s.complete(),
+    });
-    );
     return o;
   }
 

integration/microservices/src/grpc/grpc.controller.ts

@@ -50,16 +50,16 @@ export class GrpcController {
   @GrpcStreamMethod('Math')
   async sumStream(messages: Observable<any>): Promise<any> {
     return new Promise<any>((resolve, reject) => {
-      messages.subscribe(
+      messages.subscribe({
-        msg => {
+        next: msg => {
           resolve({
             result: msg.data.reduce((a, b) => a + b),
           });
         },
-        err => {
+        error: err => {
           reject(err);
         },
-      );
+      });
     });
   }
 

packages/microservices/exceptions/base-rpc-exception-filter.ts

@@ -17,7 +17,7 @@ export class BaseRpcExceptionFilter<T = any, R = any>
     }
     const res = exception.getError();
     const message = isObject(res) ? res : { status, message: res };
-    return _throw(message);
+    return _throw(() => message);
   }
 
   public handleUnknownError(exception: T, status: string) {
@@ -29,7 +29,7 @@ export class BaseRpcExceptionFilter<T = any, R = any>
     const logger = BaseRpcExceptionFilter.logger;
     logger.error.apply(logger, loggerArgs as any);
 
-    return _throw({ status, message: errorMessage });
+    return _throw(() => ({ status, message: errorMessage }));
   }
 
   public isError(exception: any): exception is Error {

packages/microservices/server/server-grpc.ts

@@ -220,10 +220,10 @@ export class ServerGrpc extends Server implements CustomTransportStrategy {
   public createUnaryServiceMethod(methodHandler: Function): Function {
     return async (call: GrpcCall, callback: Function) => {
       const handler = methodHandler(call.request, call.metadata, call);
-      this.transformToObservable(await handler).subscribe(
+      this.transformToObservable(await handler).subscribe({
-        data => callback(null, data),
+        next: data => callback(null, data),
-        (err: any) => callback(err),
+        error: (err: any) => callback(err),
-      );
+      });
     };
   }
 

packages/microservices/server/server.ts

@@ -2,7 +2,7 @@ import { Logger, LoggerService } from '@nestjs/common/services/logger.service';
 import { loadPackage } from '@nestjs/common/utils/load-package.util';
 import {
   connectable,
-  EMPTY as empty,
+  EMPTY,
   from as fromPromise,
   isObservable,
   Observable,
@@ -94,7 +94,7 @@ export abstract class Server {
       .pipe(
         catchError((err: any) => {
           scheduleOnNextTick({ err });
-          return empty;
+          return EMPTY;
         }),
         finalize(() => scheduleOnNextTick({ isDisposed: true })),
       )

packages/microservices/test/client/client-grpc.spec.ts

@@ -129,10 +129,10 @@ describe('ClientGrpcProxy', () => {
 
       it('should call native method', () => {
         const spy = sinon.spy(obj, methodName);
-        stream$.subscribe(
+        stream$.subscribe({
-          () => ({}),
+          next: () => ({}),
-          () => ({}),
+          error: () => ({}),
-        );
+        });
 
         expect(spy.called).to.be.true;
       });
@@ -156,10 +156,10 @@ describe('ClientGrpcProxy', () => {
 
       it('should subscribe to request upstream', () => {
         const upstreamSubscribe = sinon.spy(upstream, 'subscribe');
-        stream$.subscribe(
+        stream$.subscribe({
-          () => ({}),
+          next: () => ({}),
-          () => ({}),
+          error: () => ({}),
-        );
+        });
         upstream.next({ test: true });
 
         expect(writeSpy.called).to.be.true;
@@ -201,7 +201,12 @@ describe('ClientGrpcProxy', () => {
 
       it('propagates server errors', () => {
         const err = new Error('something happened');
-        stream$.subscribe(dataSpy, errorSpy, completeSpy);
+        stream$.subscribe({
+          next: dataSpy,
+          error: errorSpy,
+          complete: completeSpy,
+        });
+
         eventCallbacks.data('a');
         eventCallbacks.data('b');
         callMock.finished = true;
@@ -219,7 +224,11 @@ describe('ClientGrpcProxy', () => {
         const grpcServerCancelErrMock = {
           details: 'Cancelled',
         };
-        const subscription = stream$.subscribe(dataSpy, errorSpy);
+        const subscription = stream$.subscribe({
+          next: dataSpy,
+          error: errorSpy,
+        });
+
         eventCallbacks.data('a');
         eventCallbacks.data('b');
         subscription.unsubscribe();
@@ -258,10 +267,10 @@ describe('ClientGrpcProxy', () => {
 
       it('should call native method', () => {
         const spy = sinon.spy(obj, methodName);
-        stream$.subscribe(
+        stream$.subscribe({
-          () => ({}),
+          next: () => ({}),
-          () => ({}),
+          error: () => ({}),
-        );
+        });
 
         expect(spy.called).to.be.true;
       });
@@ -298,10 +307,10 @@ describe('ClientGrpcProxy', () => {
 
       it('should subscribe to request upstream', () => {
         const upstreamSubscribe = sinon.spy(upstream, 'subscribe');
-        stream$.subscribe(
+        stream$.subscribe({
-          () => ({}),
+          next: () => ({}),
-          () => ({}),
+          error: () => ({}),
-        );
+        });
         upstream.next({ test: true });
 
         expect(writeSpy.called).to.be.true;

packages/microservices/test/client/client-mqtt.spec.ts

@@ -1,5 +1,5 @@
 import { expect } from 'chai';
-import { empty } from 'rxjs';
+import { EMPTY } from 'rxjs';
 import * as sinon from 'sinon';
 import { ClientMqtt } from '../../client/client-mqtt';
 import { ERROR_EVENT } from '../../constants';
@@ -311,9 +311,9 @@ describe('ClientMqtt', () => {
         on: (ev, callback) => callback(error),
         off: () => ({}),
       };
-      client
+      client.mergeCloseEvent(instance as any, EMPTY).subscribe({
-        .mergeCloseEvent(instance as any, empty())
+        error: (err: any) => expect(err).to.be.eql(error),
-        .subscribe(null, (err: any) => expect(err).to.be.eql(error));
+      });
     });
   });
   describe('handleError', () => {

packages/microservices/test/client/client-proxy.spec.ts

@@ -96,12 +96,12 @@ describe('ClientProxy', function () {
           throw new Error();
         });
         const stream$ = client.send({ test: 3 }, 'test');
-        stream$.subscribe(
+        stream$.subscribe({
-          () => {},
+          next: () => {},
-          err => {
+          error: err => {
             expect(err).to.be.instanceof(Error);
           },
-        );
+        });
       });
     });
     describe('when is connected', () => {
@@ -142,12 +142,12 @@ describe('ClientProxy', function () {
           throw new Error();
         });
         const stream$ = client.emit({ test: 3 }, 'test');
-        stream$.subscribe(
+        stream$.subscribe({
-          () => {},
+          next: () => {},
-          err => {
+          error: err => {
             expect(err).to.be.instanceof(Error);
           },
-        );
+        });
       });
     });
     describe('when is connected', () => {

packages/microservices/test/client/client-rmq.spec.ts

@@ -1,6 +1,6 @@
 import { expect } from 'chai';
 import { EventEmitter } from 'events';
-import { empty } from 'rxjs';
+import { EMPTY } from 'rxjs';
 import * as sinon from 'sinon';
 import { ClientRMQ } from '../../client/client-rmq';
 import { ReadPacket } from '../../interfaces';
@@ -164,8 +164,8 @@ describe('ClientRMQ', function () {
         off: () => ({}),
       };
       client
-        .mergeDisconnectEvent(instance as any, empty())
+        .mergeDisconnectEvent(instance as any, EMPTY)
-        .subscribe(null, (err: any) => expect(err).to.be.eql(error));
+        .subscribe({ error: (err: any) => expect(err).to.be.eql(error) });
     });
   });
 

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

@@ -36,7 +36,9 @@ describe('RpcProxy', () => {
       const proxy = routerProxy.create(async (client, data) => {
         return throwError(() => new RpcException('test'));
       }, handler);
-      (await proxy(null, null)).subscribe(null, () => expectation.verify());
+      (await proxy(null, null)).subscribe({
+        error: () => expectation.verify(),
+      });
     });
   });
 });

packages/microservices/test/exceptions/rpc-exceptions-handler.spec.ts

@@ -1,5 +1,5 @@
 import { expect } from 'chai';
-import { EMPTY as empty, of } from 'rxjs';
+import { EMPTY, of } from 'rxjs';
 import { catchError } from 'rxjs/operators';
 import * as sinon from 'sinon';
 import { RpcException } from '../../exceptions/rpc-exception';
@@ -23,7 +23,7 @@ describe('RpcExceptionsHandler', () => {
               message: 'Internal server error',
             });
             done();
-            return empty;
+            return EMPTY;
           }),
         )
         .subscribe(() => ({}));
@@ -39,7 +39,7 @@ describe('RpcExceptionsHandler', () => {
             catchError((err: any) => {
               expect(err).to.be.eql(message);
               done();
-              return empty;
+              return EMPTY;
             }),
           )
           .subscribe(() => ({}));
@@ -53,7 +53,7 @@ describe('RpcExceptionsHandler', () => {
             catchError((err: any) => {
               expect(err).to.be.eql({ message, status: 'error' });
               done();
-              return empty;
+              return EMPTY;
             }),
           )
           .subscribe(() => ({}));

packages/microservices/test/server/server-mqtt.spec.ts

@@ -112,7 +112,7 @@ describe('ServerMqtt', () => {
       const handleEventSpy = sinon.spy(server, 'handleEvent');
       await server.handleMessage(
         channel,
-        new Buffer(JSON.stringify({ pattern: '', data })),
+        Buffer.from(JSON.stringify({ pattern: '', data })),
         null,
       );
       expect(handleEventSpy.called).to.be.true;
@@ -120,7 +120,7 @@ describe('ServerMqtt', () => {
     it(`should publish NO_MESSAGE_HANDLER if pattern not exists in messageHandlers object`, async () => {
       await server.handleMessage(
         channel,
-        new Buffer(JSON.stringify({ id, pattern: '', data })),
+        Buffer.from(JSON.stringify({ id, pattern: '', data })),
         null,
       );
       expect(
@@ -139,7 +139,7 @@ describe('ServerMqtt', () => {
 
       await server.handleMessage(
         channel,
-        new Buffer(JSON.stringify({ pattern: '', data, id: '2' })),
+        Buffer.from(JSON.stringify({ pattern: '', data, id: '2' })),
         null,
       );
       expect(handler.calledWith(data)).to.be.true;

packages/microservices/test/server/server.spec.ts

@@ -116,7 +116,7 @@ describe('Server', () => {
       });
       describe('throws exception', () => {
         beforeEach(() => {
-          server.send(_throw('test') as any, sendSpy);
+          server.send(_throw(() => 'test') as any, sendSpy);
         });
         it('should send error and complete', () => {
           process.nextTick(() => {

packages/platform-express/adapters/express-adapter.ts

@@ -22,7 +22,10 @@ import {
 } from '@nestjs/common/utils/shared.utils';
 import { AbstractHttpAdapter } from '@nestjs/core/adapters/http-adapter';
 import { RouterMethodFactory } from '@nestjs/core/helpers/router-method-factory';
-import * as bodyParser from 'body-parser';
+import {
+  json as bodyParserJson,
+  urlencoded as bodyParserUrlencoded,
+} from 'body-parser';
 import * as cors from 'cors';
 import * as express from 'express';
 import * as http from 'http';
@@ -162,8 +165,8 @@ export class ExpressAdapter extends AbstractHttpAdapter {
 
   public registerParserMiddleware() {
     const parserMiddleware = {
-      jsonParser: bodyParser.json(),
+      jsonParser: bodyParserJson(),
-      urlencodedParser: bodyParser.urlencoded({ extended: true }),
+      urlencodedParser: bodyParserUrlencoded({ extended: true }),
     };
     Object.keys(parserMiddleware)
       .filter(parser => !this.isMiddlewareApplied(parser))

packages/platform-ws/adapters/ws-adapter.ts

@@ -8,7 +8,7 @@ import {
 } from '@nestjs/websockets/constants';
 import { MessageMappingProperties } from '@nestjs/websockets/gateway-metadata-explorer';
 import * as http from 'http';
-import { EMPTY as empty, fromEvent, Observable } from 'rxjs';
+import { EMPTY, fromEvent, Observable } from 'rxjs';
 import { filter, first, mergeMap, share, takeUntil } from 'rxjs/operators';
 
 let wsPackage: any = {};
@@ -133,7 +133,7 @@ export class WsAdapter extends AbstractWsAdapter {
       const { callback } = messageHandler;
       return transform(callback(message.data));
     } catch {
-      return empty;
+      return EMPTY;
     }
   }
 

packages/websockets/context/ws-proxy.ts

@@ -1,5 +1,5 @@
 import { ExecutionContextHost } from '@nestjs/core/helpers/execution-context-host';
-import { empty, isObservable } from 'rxjs';
+import { EMPTY, isObservable } from 'rxjs';
 import { catchError } from 'rxjs/operators';
 import { WsExceptionsHandler } from '../exceptions/ws-exceptions-handler';
 
@@ -16,7 +16,7 @@ export class WsProxy {
           : result.pipe(
               catchError(error => {
                 this.handleError(exceptionsHandler, args, error);
-                return empty();
+                return EMPTY;
               }),
             );
       } catch (error) {