import { HorizontalAdapter , PubsubBroadcastedMessage } from './horizontal-adapter' ; import { Log } from '../log' ; import { Server } from '../server' ; const Redis = require ( 'ioredis' ) ; export class RedisAdapter extends HorizontalAdapter { /**
     * The channel to broadcast the information.
     */ protected channel = 'redis-adapter' ; /**
     * The subscription client.
     */ protected subClient : typeof Redis ; /**
     * The publishing client.
     */ protected pubClient : typeof Redis ; /**
     * Initialize the adapter.
     */ constructor ( server ) { super ( server ) ; if ( server . options . adapter . redis . prefix ) { this . channel = server . options . adapter . redis . prefix + '#' + this . channel ; } this . requestChannel = `${ this . channel }#comms#req` ; this . responseChannel = `${ this . channel }#comms#res` ; let redisDefaultOptions = { maxRetriesPerRequest : 2 , retryStrategy : times => times * 2 , } ; this . subClient = new Redis ( { ... redisDefaultOptions , ... server . options . database . redis } ) ; this . pubClient = new Redis ( { ... redisDefaultOptions , ... server . options . database . redis } ) ; const onError = err => { if ( err ) { Log . warning ( err ) ; } } ; this . subClient . psubscribe ( `${ this . channel }*` , onError ) ; this . subClient . on ( 'pmessageBuffer' , this . onMessage . bind ( this ) ) ; this . subClient . on ( 'messageBuffer' , this . processMessage . bind ( this ) ) ; this . subClient . subscribe ( [ this . requestChannel , this . responseChannel ] , onError ) ; this . pubClient . on ( 'error' , onError ) ; this . subClient . on ( 'error' , onError ) ; } /**
     * Broadcast data to a given channel.
     */ protected broadcastToChannel ( channel , data ) { this . pubClient . publish ( channel , data ) ; } /**
     * Process the incoming message and redirect it to the right processor.
     */ protected processMessage ( redisChannel , msg ) { redisChannel = redisChannel . toString ( ) ; msg = msg . toString ( ) ; if ( redisChannel . startsWith ( this . responseChannel ) ) { this . onResponse ( redisChannel , msg ) ; } else if ( redisChannel . startsWith ( this . requestChannel ) ) { this . onRequest ( redisChannel , msg ) ; } } /**
     * Listen for message coming from other nodes to broadcast
     * a specific message to the local sockets.
     */ protected onMessage ( pattern , redisChannel , msg ) { redisChannel = redisChannel . toString ( ) ; msg = msg . toString ( ) ; if ( ! redisChannel . startsWith ( this . channel ) ) { return ; } let decodedMessage = JSON . parse ( msg ) ; if ( typeof decodedMessage !== 'object' ) { return ; } const { uuid , appId , channel , data , exceptingId } = decodedMessage ; if ( uuid === this . uuid || ! appId || ! channel || ! data ) { return ; } super . sendLocally ( appId , channel , data , exceptingId ) ; } /**
     * Get the number of Redis subscribers.
     */ protected 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 ) => { return values . reduce ( ( numSub , value ) => { return numSub += parseInt ( value [ 1 ] , 10 ) ; } , 0 ) ; } ) ; } 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 ) ) ; } ) ; } ) ; } } }