feat(websocket): request scoped gateways

packages/core/nest-application.ts

@@ -159,7 +159,12 @@ export class NestApplication
     if (!this.socketModule) {
       return;
     }
-    this.socketModule.register(this.container, this.config, this.httpServer);
+    this.socketModule.register(
+      this.container,
+      this.config,
+      this.injector,
+      this.httpServer,
+    );
   }
 
   public async init(): Promise<this> {

packages/microservices/nest-microservice.ts

@@ -12,6 +12,7 @@ import { ApplicationConfig } from '@nestjs/core/application-config';
 import { MESSAGES } from '@nestjs/core/constants';
 import { optionalRequire } from '@nestjs/core/helpers/optional-require';
 import { NestContainer } from '@nestjs/core/injector/container';
+import { Injector } from '@nestjs/core/injector/injector';
 import { NestApplicationContext } from '@nestjs/core/nest-application-context';
 import { Transport } from './enums/transport.enum';
 import { CustomTransportStrategy } from './interfaces/custom-transport-strategy.interface';
@@ -68,8 +69,13 @@ export class NestMicroservice
   }
 
   public async registerModules(): Promise<any> {
+    const injector = new Injector();
     this.socketModule &&
-      this.socketModule.register(this.container, this.applicationConfig);
+      this.socketModule.register(
+        this.container,
+        this.applicationConfig,
+        injector,
+      );
     this.microservicesModule.setupClients(this.container);
 
     this.registerListeners();

packages/websockets/decorators/socket-gateway.decorator.ts

@@ -1,28 +1,33 @@
 import { GATEWAY_METADATA, GATEWAY_OPTIONS, PORT_METADATA } from '../constants';
 import { GatewayMetadata } from '../interfaces';
+import { SCOPE_OPTIONS_METADATA } from '@nestjs/common/constants';
+import { ScopeOptions } from '@nestjs/common';
 
 /**
  * Decorator that marks a class as a Nest gateway that enables real-time, bidirectional
  * and event-based communication between the browser and the server.
  */
 export function WebSocketGateway(port?: number): ClassDecorator;
 export function WebSocketGateway<
-  T extends Record<string, any> = GatewayMetadata,
+  T extends Record<string, any> & ScopeOptions = GatewayMetadata,
 >(options?: T): ClassDecorator;
 export function WebSocketGateway<
-  T extends Record<string, any> = GatewayMetadata,
+  T extends Record<string, any> & ScopeOptions = GatewayMetadata,
 >(port?: number, options?: T): ClassDecorator;
 export function WebSocketGateway<
-  T extends Record<string, any> = GatewayMetadata,
+  T extends Record<string, any> & ScopeOptions = GatewayMetadata,
 >(portOrOptions?: number | T, options?: T): ClassDecorator {
   const isPortInt = Number.isInteger(portOrOptions as number);
   // eslint-disable-next-line prefer-const
-  let [port, opt] = isPortInt ? [portOrOptions, options] : [0, portOrOptions];
+  let [port, opt] = isPortInt
+    ? [portOrOptions, options]
+    : [0, portOrOptions as T];
+  const { scope, durable, ...gatewayOptions } = opt || ({} as T);
 
-  opt = opt || ({} as T);
   return (target: object) => {
     Reflect.defineMetadata(GATEWAY_METADATA, true, target);
     Reflect.defineMetadata(PORT_METADATA, port, target);
-    Reflect.defineMetadata(GATEWAY_OPTIONS, opt, target);
+    Reflect.defineMetadata(GATEWAY_OPTIONS, gatewayOptions, target);
+    Reflect.defineMetadata(SCOPE_OPTIONS_METADATA, { scope, durable }, target);
   };
 }

packages/websockets/factories/server-and-event-streams-factory.ts

@@ -1,18 +1,59 @@
 import { ReplaySubject, Subject } from 'rxjs';
 import { ServerAndEventStreamsHost } from '../interfaces/server-and-event-streams-host.interface';
+import { WebSocketAdapter } from '@nestjs/common';
+import { ClientAndEventStreamsHost } from '../interfaces/client-and-event-streams-host.interface';
 
 export class ServerAndEventStreamsFactory {
-  public static create<T = any>(server: T): ServerAndEventStreamsHost<T> {
+  public static create<T = any>(
+    adapter: WebSocketAdapter,
+    server: T,
+  ): ServerAndEventStreamsHost<T> {
     const init = new ReplaySubject<T>();
     init.next(server);
 
-    const connection = new Subject();
-    const disconnect = new Subject();
+    const connection = new Subject<ClientAndEventStreamsHost<T>>();
+    const handler = this.getConnectionHandler(adapter, connection);
+    adapter.bindClientConnect(server, handler);
+
     return {
+      server,
       init,
       connection,
+    };
+  }
+
+  private static getConnectionHandler(
+    adapter: WebSocketAdapter,
+    connection: Subject<ClientAndEventStreamsHost>,
+  ) {
+    return (client: any, request?: any) => {
+      const observableClient = this.createObservableClient(
+        adapter,
+        client,
+        request,
+      );
+
+      connection.next(observableClient);
+    };
+  }
+
+  public static createObservableClient<TClient = any, TRequest = any>(
+    adapter: WebSocketAdapter,
+    client: TClient,
+    request?: TRequest,
+  ): ClientAndEventStreamsHost<TClient, TRequest> {
+    const disconnect = new Subject<TClient>();
+
+    const observableClient = {
+      client,
+      request,
       disconnect,
-      server,
     };
+
+    const disconnectHook = adapter.bindClientDisconnect;
+    disconnectHook &&
+      disconnectHook.call(adapter, client, () => disconnect.next(client));
+
+    return observableClient;
   }
 }

packages/websockets/interfaces/client-and-event-streams-host.interface.ts

@@ -0,0 +1,7 @@
+import { Observable } from 'rxjs';
+
+export interface ClientAndEventStreamsHost<TClient = any, TRequest = any> {
+  client: TClient;
+  request?: TRequest;
+  disconnect: Observable<TClient>;
+}

packages/websockets/interfaces/gateway-metadata.interface.ts

@@ -3,8 +3,9 @@
  */
 
 import { CorsOptions } from '@nestjs/common/interfaces/external/cors-options.interface';
+import { ScopeOptions } from '@nestjs/common';
 
-export interface GatewayMetadata {
+export interface GatewayMetadata extends ScopeOptions {
   /**
    * The name of a namespace
    */

packages/websockets/interfaces/nest-gateway.interface.ts

@@ -1,5 +1,5 @@
 export interface NestGateway {
   afterInit?: (server: any) => void;
-  handleConnection?: (...args: any[]) => void;
+  handleConnection?: (client: any, request?: any) => void;
   handleDisconnect?: (client: any) => void;
 }

packages/websockets/interfaces/server-and-event-streams-host.interface.ts

@@ -1,8 +1,8 @@
-import { ReplaySubject, Subject } from 'rxjs';
+import { Observable } from 'rxjs';
+import { ClientAndEventStreamsHost } from './client-and-event-streams-host.interface';
 
 export interface ServerAndEventStreamsHost<T = any> {
   server: T;
-  init: ReplaySubject<T>;
-  connection: Subject<any>;
-  disconnect: Subject<any>;
+  init: Observable<T>;
+  connection: Observable<ClientAndEventStreamsHost<T>>;
 }

packages/websockets/socket-module.ts

@@ -16,10 +16,10 @@ import { GATEWAY_METADATA } from './constants';
 import { ExceptionFiltersContext } from './context/exception-filters-context';
 import { WsContextCreator } from './context/ws-context-creator';
 import { WsProxy } from './context/ws-proxy';
-import { NestGateway } from './interfaces/nest-gateway.interface';
 import { SocketServerProvider } from './socket-server-provider';
 import { SocketsContainer } from './sockets-container';
 import { WebSocketsController } from './web-sockets-controller';
+import { Injector } from '@nestjs/core/injector/injector';
 
 export class SocketModule<HttpServer = any> {
   private readonly socketsContainer = new SocketsContainer();
@@ -31,6 +31,7 @@ export class SocketModule<HttpServer = any> {
   public register(
     container: NestContainer,
     config: ApplicationConfig,
+    injector: Injector,
     httpServer?: HttpServer,
   ) {
     this.applicationConfig = config;
@@ -42,6 +43,8 @@ export class SocketModule<HttpServer = any> {
       config,
     );
     this.webSocketsController = new WebSocketsController(
+      container,
+      injector,
       serverProvider,
       config,
       contextCreator,
@@ -65,7 +68,7 @@ export class SocketModule<HttpServer = any> {
     wrapper: InstanceWrapper<Injectable>,
     moduleName: string,
   ) {
-    const { instance, metatype } = wrapper;
+    const { metatype } = wrapper;
     const metadataKeys = Reflect.getMetadataKeys(metatype);
     if (!metadataKeys.includes(GATEWAY_METADATA)) {
       return;
@@ -74,7 +77,7 @@ export class SocketModule<HttpServer = any> {
       this.initializeAdapter();
     }
     this.webSocketsController.connectGatewayToServer(
-      instance as NestGateway,
+      wrapper,
       metatype,
       moduleName,
     );

packages/websockets/socket-server-provider.ts

@@ -41,8 +41,10 @@ export class SocketServerProvider {
       unknown
     >;
     const ioServer = adapter.create(port, partialOptions);
-    const serverAndEventStreamsHost =
-      ServerAndEventStreamsFactory.create(ioServer);
+    const serverAndEventStreamsHost = ServerAndEventStreamsFactory.create(
+      adapter,
+      ioServer,
+    );
 
     this.socketsContainer.addOne(
       { port, path: options.path },
@@ -59,13 +61,17 @@ export class SocketServerProvider {
     port: number,
     targetServer: unknown,
   ): ServerAndEventStreamsHost {
+    const adapter = this.applicationConfig.getIoAdapter();
+
     const namespaceServer = this.getServerOfNamespace(
       options,
       port,
       targetServer,
     );
-    const serverAndEventStreamsHost =
-      ServerAndEventStreamsFactory.create(namespaceServer);
+    const serverAndEventStreamsHost = ServerAndEventStreamsFactory.create(
+      adapter,
+      namespaceServer,
+    );
     this.socketsContainer.addOne(
       { port, path: options.path, namespace: options.namespace },
       serverAndEventStreamsHost,

packages/websockets/test/factories/server-and-event-streams-factory.spec.ts

@@ -1,17 +1,25 @@
 import { expect } from 'chai';
 import { ReplaySubject, Subject } from 'rxjs';
 import { ServerAndEventStreamsFactory } from '../../factories/server-and-event-streams-factory';
+import { AbstractWsAdapter } from '@nestjs/websockets';
+
+class NoopAdapter extends AbstractWsAdapter {
+  public create(port: number, options?: any) {}
+  public bindMessageHandlers(client: any, handlers) {}
+}
 
 describe('ServerAndEventStreamsFactory', () => {
   describe('create', () => {
     it(`should return expected observable socket object`, () => {
       const server = { test: 'test' };
-      const result = ServerAndEventStreamsFactory.create(server);
+      const result = ServerAndEventStreamsFactory.create(
+        new NoopAdapter(),
+        server,
+      );
 
       expect(result).to.have.keys('init', 'connection', 'disconnect', 'server');
       expect(result.init instanceof ReplaySubject).to.be.true;
       expect(result.connection instanceof Subject).to.be.true;
-      expect(result.disconnect instanceof Subject).to.be.true;
       expect(result.server).to.be.eql(server);
     });
   });