import { PacketType } from "socket.io-parser" ; import msgpack = require ( "notepack.io" ) ; import debugModule from "debug" ; import type { DefaultEventsMap , EventNames , EventParams , EventsMap , TypedEventBroadcaster , } from "./typed-events" ; const debug = debugModule ( "socket.io-emitter" ) ; const UID = "emitter" ; /**
 * Request types, for messages between nodes
 */ enum RequestType { SOCKETS = 0 , ALL_ROOMS = 1 , REMOTE_JOIN = 2 , REMOTE_LEAVE = 3 , REMOTE_DISCONNECT = 4 , REMOTE_FETCH = 5 , SERVER_SIDE_EMIT = 6 } export interface EmitterOptions { default key ? : string ; } interface BroadcastOptions { nsp : string ; broadcastChannel : string ; requestChannel : string ; } interface BroadcastFlags { volatile ? : boolean ; compress ? : boolean ; } export class Emitter < EmitEvents extends EventsMap = DefaultEventsMap > { private readonly opts : EmitterOptions ; private readonly broadcastOptions : BroadcastOptions ; constructor ( readonly redisClient , opts ? , readonly nsp = "/" ) { this . opts = Object . assign ( { key : "socket.io" , } , opts ) ; this . broadcastOptions = { nsp , broadcastChannel : this . opts . key + "#" + nsp + "#" , requestChannel : this . opts . key + "-request#" + nsp + "#" , } ; } param nsp public public of ( nsp ) { return new Emitter ( this . redisClient , this . opts , ( nsp [ 0 ] !== "/" ? "/" : "" ) + nsp ) ; } return public public emit < Ev extends EventNames < EmitEvents > > ( ev , ... args ) { return new BroadcastOperator < EmitEvents > ( this . redisClient , this . broadcastOptions ) . emit ( ev , ... args ) ; } param room return public public to ( room ) { return new BroadcastOperator ( this . redisClient , this . broadcastOptions ) . to ( room ) ; } param room return public public in ( room ) { return new BroadcastOperator ( this . redisClient , this . broadcastOptions ) . in ( room ) ; } param room return public public except ( room ) { return new BroadcastOperator ( this . redisClient , this . broadcastOptions ) . except ( room ) ; } return public public get volatile ( ) : BroadcastOperator < EmitEvents > { return new BroadcastOperator ( this . redisClient , this . broadcastOptions ) . volatile ; } param compress return public public compress ( compress ) { return new BroadcastOperator ( this . redisClient , this . broadcastOptions ) . compress ( compress ) ; } param rooms public public socketsJoin ( rooms ) { return new BroadcastOperator ( this . redisClient , this . broadcastOptions ) . socketsJoin ( rooms ) ; } param rooms public public socketsLeave ( rooms ) { return new BroadcastOperator ( this . redisClient , this . broadcastOptions ) . socketsLeave ( rooms ) ; } param close public public disconnectSockets ( close = false ) { return new BroadcastOperator ( this . redisClient , this . broadcastOptions ) . disconnectSockets ( close ) ; } param args public serverSideEmit ( ... args ) { const withAck = typeof args [ args . length - 1 ] === "function" ; if ( withAck ) { throw new Error ( "Acknowledgements are not supported" ) ; } const request = JSON . stringify ( { uid : UID , type : RequestType . SERVER_SIDE_EMIT , data : args , } ) ; this . redisClient . publish ( this . broadcastOptions . requestChannel , request ) ; } } export const RESERVED_EVENTS = new Set ( < const > [ "connect" , "connect_error" , "disconnect" , "disconnecting" , "newListener" , "removeListener" , ] ) ; export class BroadcastOperator < EmitEvents extends EventsMap > implements TypedEventBroadcaster < EmitEvents > { constructor ( private readonly redisClient , private readonly broadcastOptions , private readonly rooms = new Set < string > ( ) , private readonly exceptRooms = new Set < string > ( ) , private readonly flags = { } ) { } param room return public public to ( room ) { const rooms = new Set ( this . rooms ) ; if ( Array . isArray ( room ) ) { room . forEach ( ( r ) => rooms . add ( r ) ) ; } else { rooms . add ( room ) ; } return new BroadcastOperator ( this . redisClient , this . broadcastOptions , rooms , this . exceptRooms , this . flags ) ; } param room return public public in ( room ) { return this . to ( room ) ; } param room return public public except ( room ) { const exceptRooms = new Set ( this . exceptRooms ) ; if ( Array . isArray ( room ) ) { room . forEach ( ( r ) => exceptRooms . add ( r ) ) ; } else { exceptRooms . add ( room ) ; } return new BroadcastOperator ( this . redisClient , this . broadcastOptions , this . rooms , exceptRooms , this . flags ) ; } param compress return public public compress ( compress ) { const flags = Object . assign ( { } , this . flags , { compress } ) ; return new BroadcastOperator ( this . redisClient , this . broadcastOptions , this . rooms , this . exceptRooms , flags ) ; } return public public get volatile ( ) : BroadcastOperator < EmitEvents > { const flags = Object . assign ( { } , this . flags , { volatile : true } ) ; return new BroadcastOperator ( this . redisClient , this . broadcastOptions , this . rooms , this . exceptRooms , flags ) ; } return public public emit < Ev extends EventNames < EmitEvents > > ( ev , ... args ) { if ( RESERVED_EVENTS . has ( ev ) ) { throw new Error ( `"${ ev }" is a reserved event name` ) ; } const data = [ ev , ... args ] ; const packet = { type : PacketType . EVENT , data : data , nsp : this . broadcastOptions . nsp , } ; const opts = { rooms : [ ... this . rooms ] , flags : this . flags , except : [ ... this . exceptRooms ] , } ; const msg = msgpack . encode ( [ UID , packet , opts ] ) ; let channel = this . broadcastOptions . broadcastChannel ; if ( this . rooms && this . rooms . size === 1 ) { channel += this . rooms . keys ( ) . next ( ) . value + "#" ; } debug ( "publishing message to channel %s" , channel ) ; this . redisClient . publish ( channel , msg ) ; return true ; } param rooms public public socketsJoin ( rooms ) { const request = JSON . stringify ( { type : RequestType . REMOTE_JOIN , opts : { rooms : [ ... this . rooms ] , except : [ ... this . exceptRooms ] , } , rooms : Array . isArray ( rooms ) ? rooms : [ rooms ] , } ) ; this . redisClient . publish ( this . broadcastOptions . requestChannel , request ) ; } param rooms public public socketsLeave ( rooms ) { const request = JSON . stringify ( { type : RequestType . REMOTE_LEAVE , opts : { rooms : [ ... this . rooms ] , except : [ ... this . exceptRooms ] , } , rooms : Array . isArray ( rooms ) ? rooms : [ rooms ] , } ) ; this . redisClient . publish ( this . broadcastOptions . requestChannel , request ) ; } param close public public disconnectSockets ( close = false ) { const request = JSON . stringify ( { type : RequestType . REMOTE_DISCONNECT , opts : { rooms : [ ... this . rooms ] , except : [ ... this . exceptRooms ] , } , close , } ) ; this . redisClient . publish ( this . broadcastOptions . requestChannel , request ) ; } }