import { EventEmitter } from "events" ; import ConnectionPool from "./ConnectionPool" ; import { getConnectionName , getNodeKey } from "./util" ; import { sample , noop , Debug } from "../utils" ; import Redis from "../redis" ; const debug = Debug ( "cluster:subscriber" ) ; export default class ClusterSubscriber { private started = false ; private subscriber : any = null ; private lastActiveSubscriber : any ; constructor ( private connectionPool , private emitter ) { this . connectionPool . on ( "-node" , ( _ , key ) => { if ( ! this . started || ! this . subscriber ) { return ; } if ( getNodeKey ( this . subscriber . options ) === key ) { debug ( "subscriber has left, selecting a new one..." ) ; this . selectSubscriber ( ) ; } } ) ; this . connectionPool . on ( "+node" , ( ) => { if ( ! this . started || this . subscriber ) { return ; } debug ( "a new node is discovered and there is no subscriber, selecting a new one..." ) ; this . selectSubscriber ( ) ; } ) ; } getInstance ( ) { return this . subscriber ; } private selectSubscriber ( ) { const lastActiveSubscriber = this . lastActiveSubscriber ; if ( lastActiveSubscriber ) { lastActiveSubscriber . disconnect ( ) ; } if ( this . subscriber ) { this . subscriber . disconnect ( ) ; } const sampleNode = sample ( this . connectionPool . getNodes ( ) ) ; if ( ! sampleNode ) { debug ( "selecting subscriber failed since there is no node discovered in the cluster yet" ) ; this . subscriber = null ; return ; } const { options } = sampleNode ; debug ( "selected a subscriber %s:%s" , options . host , options . port ) ; this . subscriber = new Redis ( { port : options . port , host : options . host , username : options . username , password : options . password , enableReadyCheck : true , connectionName : getConnectionName ( "subscriber" , options . connectionName ) , lazyConnect : true , tls : options . tls , } ) ; this . subscriber . on ( "error" , noop ) ; const previousChannels = { subscribe : [ ] , psubscribe : [ ] } ; if ( lastActiveSubscriber ) { const condition = lastActiveSubscriber . condition || lastActiveSubscriber . prevCondition ; if ( condition && condition . subscriber ) { previousChannels . subscribe = condition . subscriber . channels ( "subscribe" ) ; previousChannels . psubscribe = condition . subscriber . channels ( "psubscribe" ) ; } } if ( previousChannels . subscribe . length || previousChannels . psubscribe . length ) { let pending = 0 ; for ( const type of [ "subscribe" , "psubscribe" ] ) { const channels = previousChannels [ type ] ; if ( channels . length ) { pending += 1 ; debug ( "%s %d channels" , type , channels . length ) ; this . subscriber [ type ] ( channels ) . then ( ( ) => { if ( ! -- pending ) { this . lastActiveSubscriber = this . subscriber ; } } ) . catch ( ( ) => { debug ( "failed to %s %d channels" , type , channels . length ) ; } ) ; } } } else { this . lastActiveSubscriber = this . subscriber ; } for ( const event of [ "message" , "messageBuffer" ] ) { this . subscriber . on ( event , ( arg1 , arg2 ) => { this . emitter . emit ( event , arg1 , arg2 ) ; } ) ; } for ( const event of [ "pmessage" , "pmessageBuffer" ] ) { this . subscriber . on ( event , ( arg1 , arg2 , arg3 ) => { this . emitter . emit ( event , arg1 , arg2 , arg3 ) ; } ) ; } } start ( ) { this . started = true ; this . selectSubscriber ( ) ; debug ( "started" ) ; } stop ( ) { this . started = false ; if ( this . subscriber ) { this . subscriber . disconnect ( ) ; this . subscriber = null ; } debug ( "stopped" ) ; } }