import { OnGatewayConnection , OnGatewayDisconnect , OnGatewayInit , SubscribeMessage , WebSocketGateway , WebSocketServer , WsResponse , } from '@nestjs/websockets' ; import { CommandBus , EventBus } from '@nestjs/cqrs' ; import { Observable , of } from 'rxjs' ; import { Logger , UseGuards } from '@nestjs/common' ; import { delay } from 'rxjs/operators' ; import { ISocket } from './interfaces/socket.interface' ; import { Server } from 'socket.io' ; import { WsAuthGuard } from '../auth' ; import { User } from '@ngx-starter-kit/models' ; import { GenericCommand } from './commands/generic.command' ; import { GenericEvent } from './events/generic.event' ; @ WebSocketGateway ( { namespace : 'eventbus' } ) export class CQRSGateway implements OnGatewayInit , OnGatewayConnection , OnGatewayDisconnect { static EVENTS = 'events' ; static COMMANDS = 'actions' ; private readonly logger = new Logger ( CQRSGateway . name ) ; @ WebSocketServer ( ) server : Server ; clients : ISocket [ ] = [ ] ; constructor ( private readonly eventBus , private readonly commandBus ) { } afterInit ( server ) { } handleConnection ( client ) { this . clients . push ( client ) ; } handleDisconnect ( client ) { this . clients = this . clients . filter ( ( c ) => ( c as any ) . id !== ( client as any ) . id ) ; } @ UseGuards ( WsAuthGuard ) @ SubscribeMessage ( 'auth' ) onAuthenticate ( client , data ) { const event = 'auth' ; return { event , status : 'success' } ; } @ SubscribeMessage ( 'test' ) onTest ( client , data ) { const event = 'test' ; return of ( { event , data } ) . pipe ( delay ( 1000 ) ) ; } @ SubscribeMessage ( CQRSGateway . EVENTS ) onEvent ( client , event ) { this . eventBus . publish ( new GenericEvent ( event . type , event . payload , client . user ) ) ; } @ SubscribeMessage ( CQRSGateway . COMMANDS ) onCommand ( client , command ) { this . commandBus . execute ( new GenericCommand ( command . type , command . payload , client . user ) ) ; } sendCommandToUser < T > ( user , action ) { const clients = this . getSocketsForUser ( user ) ; const type = this . getActionTypeFromInstance ( action ) ; clients . forEach ( ( socket ) => ( socket as any ) . emit ( CQRSGateway . COMMANDS , { ... action , type } ) ) ; } sendCommandToAll < T > ( action ) { const type = this . getActionTypeFromInstance ( action ) ; this . clients . forEach ( ( socket ) => ( socket as any ) . emit ( CQRSGateway . COMMANDS , { ... action , type } ) ) ; } sendEventToUser < T > ( user , event ) { const clients = this . getSocketsForUser ( user ) ; const type = this . getActionTypeFromInstance ( event ) ; clients . forEach ( ( socket ) => ( socket as any ) . emit ( CQRSGateway . EVENTS , { ... event , type } ) ) ; } sendEventToAll < T > ( event ) { const type = this . getActionTypeFromInstance ( event ) ; this . clients . forEach ( ( socket ) => ( socket as any ) . emit ( CQRSGateway . EVENTS , { ... event , type } ) ) ; } private getSocketsForUser ( user ) { return this . clients . filter ( ( c ) => c . user && c . user . username === user . username ) ; } private getActionTypeFromInstance ( action ) { if ( action . constructor && action . constructor . type ) { return action . constructor . type ; } return action . type ; } }