import uid2 = require ( "uid2" ) ; import msgpack = require ( "notepack.io" ) ; import { Adapter , BroadcastOptions , Room , SocketId } from "socket.io-adapter" ; const debug = require ( "debug" ) ( "socket.io-redis" ) ; module . exports = exports = createAdapter ; /**
 * 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 } interface Request { type : RequestType ; resolve : Function ; timeout : NodeJS . Timeout ; numSub ? : number ; msgCount ? : number ; [ other ] : any ; } const isNumeric = ( str ) => ! isNaN ( str ) && ! isNaN ( parseFloat ( str ) ) ; export interface RedisAdapterOptions { default key : string ; default requestsTimeout : number ; default publishOnSpecificResponseChannel : boolean ; } param pubClient param subClient param opts public export function createAdapter ( pubClient , subClient , opts ? ) { return function ( nsp ) { return new RedisAdapter ( nsp , pubClient , subClient , opts ) ; } ; } export class RedisAdapter extends Adapter { public readonly uid ; public readonly requestsTimeout : number ; public readonly publishOnSpecificResponseChannel : boolean ; private readonly channel : string ; private readonly requestChannel : string ; private readonly responseChannel : string ; private requests : Map < string , Request > = new Map ( ) ; param nsp param pubClient param subClient param opts public constructor ( nsp , readonly pubClient , readonly subClient , opts = { } ) { super ( nsp ) ; this . uid = uid2 ( 6 ) ; this . requestsTimeout = opts . requestsTimeout || 5000 ; this . publishOnSpecificResponseChannel = ! ! opts . publishOnSpecificResponseChannel ; const prefix = opts . key || "socket.io" ; this . channel = prefix + "#" + nsp . name + "#" ; this . requestChannel = prefix + "-request#" + this . nsp . name + "#" ; this . responseChannel = prefix + "-response#" + this . nsp . name + "#" ; const specificResponseChannel = this . responseChannel + this . uid + "#" ; const isRedisV4 = typeof this . pubClient . pSubscribe === "function" ; if ( isRedisV4 ) { this . subClient . pSubscribe ( this . channel + "*" , ( msg , channel ) => { this . onmessage ( null , channel , msg ) ; } , true ) ; this . subClient . subscribe ( [ this . requestChannel , this . responseChannel , specificResponseChannel ] , ( msg , channel ) => { this . onrequest ( channel , msg ) ; } ) ; } else { this . subClient . psubscribe ( this . channel + "*" ) ; this . subClient . on ( "pmessageBuffer" , this . onmessage . bind ( this ) ) ; this . subClient . subscribe ( [ this . requestChannel , this . responseChannel , specificResponseChannel , ] ) ; this . subClient . on ( "messageBuffer" , this . onrequest . bind ( this ) ) ; } const registerFriendlyErrorHandler = ( redisClient ) => { redisClient . on ( "error" , ( ) => { if ( redisClient . listenerCount ( "error" ) === 1 ) { console . warn ( "missing 'error' handler on this Redis client" ) ; } } ) ; } ; registerFriendlyErrorHandler ( this . pubClient ) ; registerFriendlyErrorHandler ( this . subClient ) ; } private private onmessage ( pattern , channel , msg ) { channel = channel . toString ( ) ; const channelMatches = channel . startsWith ( this . channel ) ; if ( ! channelMatches ) { return debug ( "ignore different channel" ) ; } const room = channel . slice ( this . channel . length , - 1 ) ; if ( room !== "" && ! this . hasRoom ( room ) ) { return debug ( "ignore unknown room %s" , room ) ; } const args = msgpack . decode ( msg ) ; const [ uid , packet , opts ] = args ; if ( this . uid === uid ) return debug ( "ignore same uid" ) ; if ( packet && packet . nsp === undefined ) { packet . nsp = "/" ; } if ( ! packet || packet . nsp !== this . nsp . name ) { return debug ( "ignore different namespace" ) ; } opts . rooms = new Set ( opts . rooms ) ; opts . except = new Set ( opts . except ) ; super . broadcast ( packet , opts ) ; } private hasRoom ( room ) { const hasNumericRoom = isNumeric ( room ) && this . rooms . has ( parseFloat ( room ) ) ; return hasNumericRoom || this . rooms . has ( room ) ; } private private async onrequest ( channel , msg ) { channel = channel . toString ( ) ; if ( channel . startsWith ( this . responseChannel ) ) { return this . onresponse ( channel , msg ) ; } else if ( ! channel . startsWith ( this . requestChannel ) ) { return debug ( "ignore different channel" ) ; } let request ; try { request = JSON . parse ( msg ) ; } catch ( err ) { debug ( "ignoring malformed request" ) ; return ; } debug ( "received request %j" , request ) ; let response , socket ; switch ( request . type ) { case RequestType . SOCKETS : if ( this . requests . has ( request . requestId ) ) { return ; } const sockets = await super . sockets ( new Set ( request . rooms ) ) ; response = JSON . stringify ( { requestId : request . requestId , sockets : [ ... sockets ] , } ) ; this . publishResponse ( request , response ) ; break ; case RequestType . ALL_ROOMS : if ( this . requests . has ( request . requestId ) ) { return ; } response = JSON . stringify ( { requestId : request . requestId , rooms : [ ... this . rooms . keys ( ) ] , } ) ; this . publishResponse ( request , response ) ; break ; case RequestType . REMOTE_JOIN : if ( request . opts ) { const opts = { rooms : new Set < Room > ( request . opts . rooms ) , except : new Set < Room > ( request . opts . except ) , } ; return super . addSockets ( opts , request . rooms ) ; } socket = this . nsp . sockets . get ( request . sid ) ; if ( ! socket ) { return ; } socket . join ( request . room ) ; response = JSON . stringify ( { requestId : request . requestId , } ) ; this . publishResponse ( request , response ) ; break ; case RequestType . REMOTE_LEAVE : if ( request . opts ) { const opts = { rooms : new Set < Room > ( request . opts . rooms ) , except : new Set < Room > ( request . opts . except ) , } ; return super . delSockets ( opts , request . rooms ) ; } socket = this . nsp . sockets . get ( request . sid ) ; if ( ! socket ) { return ; } socket . leave ( request . room ) ; response = JSON . stringify ( { requestId : request . requestId , } ) ; this . publishResponse ( request , response ) ; break ; case RequestType . REMOTE_DISCONNECT : if ( request . opts ) { const opts = { rooms : new Set < Room > ( request . opts . rooms ) , except : new Set < Room > ( request . opts . except ) , } ; return super . disconnectSockets ( opts , request . close ) ; } socket = this . nsp . sockets . get ( request . sid ) ; if ( ! socket ) { return ; } socket . disconnect ( request . close ) ; response = JSON . stringify ( { requestId : request . requestId , } ) ; this . publishResponse ( request , response ) ; break ; case RequestType . REMOTE_FETCH : if ( this . requests . has ( request . requestId ) ) { return ; } const opts = { rooms : new Set < Room > ( request . opts . rooms ) , except : new Set < Room > ( request . opts . except ) , } ; const localSockets = await super . fetchSockets ( opts ) ; response = JSON . stringify ( { requestId : request . requestId , sockets : localSockets . map ( ( socket ) => { const { sessionStore , ... handshake } = socket . handshake ; return { id : socket . id , handshake , rooms : [ ... socket . rooms ] , data : socket . data , } ; } ) , } ) ; this . publishResponse ( request , response ) ; break ; case RequestType . SERVER_SIDE_EMIT : if ( request . uid === this . uid ) { debug ( "ignore same uid" ) ; return ; } const withAck = request . requestId !== undefined ; if ( ! withAck ) { this . nsp . _onServerSideEmit ( request . data ) ; return ; } let called = false ; const callback = ( arg ) => { if ( called ) { return ; } called = true ; debug ( "calling acknowledgement with %j" , arg ) ; this . pubClient . publish ( this . responseChannel , JSON . stringify ( { type : RequestType . SERVER_SIDE_EMIT , requestId : request . requestId , data : arg , } ) ) ; } ; request . data . push ( callback ) ; this . nsp . _onServerSideEmit ( request . data ) ; break ; default : debug ( "ignoring unknown request type: %s" , request . type ) ; } } param request param response private private publishResponse ( request , response ) { const responseChannel = this . publishOnSpecificResponseChannel ? `${ this . responseChannel }${ request . uid }#` : this . responseChannel ; debug ( "publishing response to channel %s" , responseChannel ) ; this . pubClient . publish ( responseChannel , response ) ; } private private onresponse ( channel , msg ) { let response ; try { response = JSON . parse ( msg ) ; } catch ( err ) { debug ( "ignoring malformed response" ) ; return ; } const requestId = response . requestId ; if ( ! requestId || ! this . requests . has ( requestId ) ) { debug ( "ignoring unknown request" ) ; return ; } debug ( "received response %j" , response ) ; const request = this . requests . get ( requestId ) ; switch ( request . type ) { case RequestType . SOCKETS : case RequestType . REMOTE_FETCH : request . msgCount ++ ; if ( ! response . sockets || ! Array . isArray ( response . sockets ) ) return ; if ( request . type === RequestType . SOCKETS ) { response . sockets . forEach ( ( s ) => request . sockets . add ( s ) ) ; } else { response . sockets . forEach ( ( s ) => request . sockets . push ( s ) ) ; } if ( request . msgCount === request . numSub ) { clearTimeout ( request . timeout ) ; if ( request . resolve ) { request . resolve ( request . sockets ) ; } this . requests . delete ( requestId ) ; } break ; case RequestType . ALL_ROOMS : request . msgCount ++ ; if ( ! response . rooms || ! Array . isArray ( response . rooms ) ) return ; response . rooms . forEach ( ( s ) => request . rooms . add ( s ) ) ; if ( request . msgCount === request . numSub ) { clearTimeout ( request . timeout ) ; if ( request . resolve ) { request . resolve ( request . rooms ) ; } this . requests . delete ( requestId ) ; } break ; case RequestType . REMOTE_JOIN : case RequestType . REMOTE_LEAVE : case RequestType . REMOTE_DISCONNECT : clearTimeout ( request . timeout ) ; if ( request . resolve ) { request . resolve ( ) ; } this . requests . delete ( requestId ) ; break ; case RequestType . SERVER_SIDE_EMIT : request . responses . push ( response . data ) ; debug ( "serverSideEmit: got %d responses out of %d" , request . responses . length , request . numSub ) ; if ( request . responses . length === request . numSub ) { clearTimeout ( request . timeout ) ; if ( request . resolve ) { request . resolve ( null , request . responses ) ; } this . requests . delete ( requestId ) ; } break ; default : debug ( "ignoring unknown request type: %s" , request . type ) ; } } param { Object } packet param { Object } opts public public broadcast ( packet , opts ) { packet . nsp = this . nsp . name ; const onlyLocal = opts && opts . flags && opts . flags . local ; if ( ! onlyLocal ) { const rawOpts = { rooms : [ ... opts . rooms ] , except : [ ... new Set ( opts . except ) ] , flags : opts . flags , } ; const msg = msgpack . encode ( [ this . uid , packet , rawOpts ] ) ; let channel = this . channel ; if ( opts . rooms && opts . rooms . size === 1 ) { channel += opts . rooms . keys ( ) . next ( ) . value + "#" ; } debug ( "publishing message to channel %s" , channel ) ; this . pubClient . publish ( channel , msg ) ; } super . broadcast ( packet , opts ) ; } param { Set < Room > } rooms public async sockets ( rooms ) { const localSockets = await super . sockets ( rooms ) ; const numSub = await this . getNumSub ( ) ; debug ( 'waiting for %d responses to "sockets" request' , numSub ) ; if ( numSub <= 1 ) { return Promise . resolve ( localSockets ) ; } const requestId = uid2 ( 6 ) ; const request = JSON . stringify ( { uid : this . uid , requestId , type : RequestType . SOCKETS , rooms : [ ... rooms ] , } ) ; return new Promise ( ( resolve , reject ) => { const timeout = setTimeout ( ( ) => { if ( this . requests . has ( requestId ) ) { reject ( new Error ( "timeout reached while waiting for sockets response" ) ) ; this . requests . delete ( requestId ) ; } } , this . requestsTimeout ) ; this . requests . set ( requestId , { type : RequestType . SOCKETS , numSub , resolve , timeout , msgCount : 1 , sockets : localSockets , } ) ; this . pubClient . publish ( this . requestChannel , request ) ; } ) ; } public public async allRooms ( ) { const localRooms = new Set ( this . rooms . keys ( ) ) ; const numSub = await this . getNumSub ( ) ; debug ( 'waiting for %d responses to "allRooms" request' , numSub ) ; if ( numSub <= 1 ) { return localRooms ; } const requestId = uid2 ( 6 ) ; const request = JSON . stringify ( { uid : this . uid , requestId , type : RequestType . ALL_ROOMS , } ) ; return new Promise ( ( resolve , reject ) => { const timeout = setTimeout ( ( ) => { if ( this . requests . has ( requestId ) ) { reject ( new Error ( "timeout reached while waiting for allRooms response" ) ) ; this . requests . delete ( requestId ) ; } } , this . requestsTimeout ) ; this . requests . set ( requestId , { type : RequestType . ALL_ROOMS , numSub , resolve , timeout , msgCount : 1 , rooms : localRooms , } ) ; this . pubClient . publish ( this . requestChannel , request ) ; } ) ; } param { String } id param { String } room public public remoteJoin ( id , room ) { const requestId = uid2 ( 6 ) ; const socket = this . nsp . sockets . get ( id ) ; if ( socket ) { socket . join ( room ) ; return Promise . resolve ( ) ; } const request = JSON . stringify ( { uid : this . uid , requestId , type : RequestType . REMOTE_JOIN , sid : id , room , } ) ; return new Promise ( ( resolve , reject ) => { const timeout = setTimeout ( ( ) => { if ( this . requests . has ( requestId ) ) { reject ( new Error ( "timeout reached while waiting for remoteJoin response" ) ) ; this . requests . delete ( requestId ) ; } } , this . requestsTimeout ) ; this . requests . set ( requestId , { type : RequestType . REMOTE_JOIN , resolve , timeout , } ) ; this . pubClient . publish ( this . requestChannel , request ) ; } ) ; } param { String } id param { String } room public public remoteLeave ( id , room ) { const requestId = uid2 ( 6 ) ; const socket = this . nsp . sockets . get ( id ) ; if ( socket ) { socket . leave ( room ) ; return Promise . resolve ( ) ; } const request = JSON . stringify ( { uid : this . uid , requestId , type : RequestType . REMOTE_LEAVE , sid : id , room , } ) ; return new Promise ( ( resolve , reject ) => { const timeout = setTimeout ( ( ) => { if ( this . requests . has ( requestId ) ) { reject ( new Error ( "timeout reached while waiting for remoteLeave response" ) ) ; this . requests . delete ( requestId ) ; } } , this . requestsTimeout ) ; this . requests . set ( requestId , { type : RequestType . REMOTE_LEAVE , resolve , timeout , } ) ; this . pubClient . publish ( this . requestChannel , request ) ; } ) ; } param { String } id param { Boolean } close public public remoteDisconnect ( id , close ? ) { const requestId = uid2 ( 6 ) ; const socket = this . nsp . sockets . get ( id ) ; if ( socket ) { socket . disconnect ( close ) ; return Promise . resolve ( ) ; } const request = JSON . stringify ( { uid : this . uid , requestId , type : RequestType . REMOTE_DISCONNECT , sid : id , close , } ) ; return new Promise ( ( resolve , reject ) => { const timeout = setTimeout ( ( ) => { if ( this . requests . has ( requestId ) ) { reject ( new Error ( "timeout reached while waiting for remoteDisconnect response" ) ) ; this . requests . delete ( requestId ) ; } } , this . requestsTimeout ) ; this . requests . set ( requestId , { type : RequestType . REMOTE_DISCONNECT , resolve , timeout , } ) ; this . pubClient . publish ( this . requestChannel , request ) ; } ) ; } public async fetchSockets ( opts ) { const localSockets = await super . fetchSockets ( opts ) ; if ( opts . flags ?. local ) { return localSockets ; } const numSub = await this . getNumSub ( ) ; debug ( 'waiting for %d responses to "fetchSockets" request' , numSub ) ; if ( numSub <= 1 ) { return localSockets ; } const requestId = uid2 ( 6 ) ; const request = JSON . stringify ( { uid : this . uid , requestId , type : RequestType . REMOTE_FETCH , opts : { rooms : [ ... opts . rooms ] , except : [ ... opts . except ] , } , } ) ; return new Promise ( ( resolve , reject ) => { const timeout = setTimeout ( ( ) => { if ( this . requests . has ( requestId ) ) { reject ( new Error ( "timeout reached while waiting for fetchSockets response" ) ) ; this . requests . delete ( requestId ) ; } } , this . requestsTimeout ) ; this . requests . set ( requestId , { type : RequestType . REMOTE_FETCH , numSub , resolve , timeout , msgCount : 1 , sockets : localSockets , } ) ; this . pubClient . publish ( this . requestChannel , request ) ; } ) ; } public addSockets ( opts , rooms ) { if ( opts . flags ?. local ) { return super . addSockets ( opts , rooms ) ; } const request = JSON . stringify ( { uid : this . uid , type : RequestType . REMOTE_JOIN , opts : { rooms : [ ... opts . rooms ] , except : [ ... opts . except ] , } , rooms : [ ... rooms ] , } ) ; this . pubClient . publish ( this . requestChannel , request ) ; } public delSockets ( opts , rooms ) { if ( opts . flags ?. local ) { return super . delSockets ( opts , rooms ) ; } const request = JSON . stringify ( { uid : this . uid , type : RequestType . REMOTE_LEAVE , opts : { rooms : [ ... opts . rooms ] , except : [ ... opts . except ] , } , rooms : [ ... rooms ] , } ) ; this . pubClient . publish ( this . requestChannel , request ) ; } public disconnectSockets ( opts , close ) { if ( opts . flags ?. local ) { return super . disconnectSockets ( opts , close ) ; } const request = JSON . stringify ( { uid : this . uid , type : RequestType . REMOTE_DISCONNECT , opts : { rooms : [ ... opts . rooms ] , except : [ ... opts . except ] , } , close , } ) ; this . pubClient . publish ( this . requestChannel , request ) ; } public serverSideEmit ( packet ) { const withAck = typeof packet [ packet . length - 1 ] === "function" ; if ( withAck ) { this . serverSideEmitWithAck ( packet ) . catch ( ( ) => { } ) ; return ; } const request = JSON . stringify ( { uid : this . uid , type : RequestType . SERVER_SIDE_EMIT , data : packet , } ) ; this . pubClient . publish ( this . requestChannel , request ) ; } private async serverSideEmitWithAck ( packet ) { const ack = packet . pop ( ) ; const numSub = ( await this . getNumSub ( ) ) - 1 ; debug ( 'waiting for %d responses to "serverSideEmit" request' , numSub ) ; if ( numSub <= 0 ) { return ack ( null , [ ] ) ; } const requestId = uid2 ( 6 ) ; const request = JSON . stringify ( { uid : this . uid , requestId , type : RequestType . SERVER_SIDE_EMIT , data : packet , } ) ; const timeout = setTimeout ( ( ) => { const storedRequest = this . requests . get ( requestId ) ; if ( storedRequest ) { ack ( new Error ( `timeout reached: only ${ storedRequest . responses . length } responses received out of ${ storedRequest . numSub }` ) , storedRequest . responses ) ; this . requests . delete ( requestId ) ; } } , this . requestsTimeout ) ; this . requests . set ( requestId , { type : RequestType . SERVER_SIDE_EMIT , numSub , timeout , resolve : ack , responses : [ ] , } ) ; this . pubClient . publish ( this . requestChannel , request ) ; } private private getNumSub ( ) { if ( this . pubClient . constructor . name === "Cluster" ) { const nodes = this . pubClient . nodes ( ) ; return Promise . all ( nodes . map ( ( node ) => node . send_command ( "pubsub" , [ "numsub" , this . requestChannel ] ) ) ) . then ( ( values ) => { let numSub = 0 ; values . map ( ( value ) => { numSub += parseInt ( value [ 1 ] , 10 ) ; } ) ; return numSub ; } ) ; } else if ( typeof this . pubClient . pSubscribe === "function" ) { return this . pubClient . sendCommand ( [ "pubsub" , "numsub" , this . requestChannel ] ) . then ( ( res ) => parseInt ( res [ 1 ] , 10 ) ) ; } else { return new Promise ( ( resolve , reject ) => { this . pubClient . send_command ( "pubsub" , [ "numsub" , this . requestChannel ] , ( err , numSub ) => { if ( err ) return reject ( err ) ; resolve ( parseInt ( numSub [ 1 ] , 10 ) ) ; } ) ; } ) ; } } }