import { EventEmitter } from "events" ; import { sample , Debug , noop , defaults } from "../utils" ; import { IRedisOptions , getNodeKey , NodeKey , NodeRole } from "./util" ; import Redis from "../redis" ; const debug = Debug ( "cluster:connectionPool" ) ; type NODE_TYPE = "all" | "master" | "slave" ; export default class ConnectionPool extends EventEmitter { private nodes : { [ key in NODE_TYPE ] : { [ key ] : any ; } ; } = { all : { } , master : { } , slave : { } , } ; private specifiedOptions : { [ key ] : any ; } = { } ; constructor ( private redisOptions ) { super ( ) ; } public getNodes ( role = "all" ) { const nodes = this . nodes [ role ] ; return Object . keys ( nodes ) . map ( ( key ) => nodes [ key ] ) ; } public getInstanceByKey ( key ) { return this . nodes . all [ key ] ; } public getSampleInstance ( role ) { const keys = Object . keys ( this . nodes [ role ] ) ; const sampleKey = sample ( keys ) ; return this . nodes [ role ] [ sampleKey ] ; } param { IRedisOptions } node param { boolean } readOnly returns { * } memberof public findOrCreate ( node , readOnly = false ) { const key = getNodeKey ( node ) ; readOnly = Boolean ( readOnly ) ; if ( this . specifiedOptions [ key ] ) { Object . assign ( node , this . specifiedOptions [ key ] ) ; } else { this . specifiedOptions [ key ] = node ; } let redis ; if ( this . nodes . all [ key ] ) { redis = this . nodes . all [ key ] ; if ( redis . options . readOnly !== readOnly ) { redis . options . readOnly = readOnly ; debug ( "Change role of %s to %s" , key , readOnly ? "slave" : "master" ) ; redis [ readOnly ? "readonly" : "readwrite" ] ( ) . catch ( noop ) ; if ( readOnly ) { delete this . nodes . master [ key ] ; this . nodes . slave [ key ] = redis ; } else { delete this . nodes . slave [ key ] ; this . nodes . master [ key ] = redis ; } } } else { debug ( "Connecting to %s as %s" , key , readOnly ? "slave" : "master" ) ; redis = new Redis ( defaults ( { retryStrategy : null , enableOfflineQueue : true , readOnly : readOnly , } , node , this . redisOptions , { lazyConnect : true } ) ) ; this . nodes . all [ key ] = redis ; this . nodes [ readOnly ? "slave" : "master" ] [ key ] = redis ; redis . once ( "end" , ( ) => { this . removeNode ( key ) ; this . emit ( "-node" , redis , key ) ; if ( ! Object . keys ( this . nodes . all ) . length ) { this . emit ( "drain" ) ; } } ) ; this . emit ( "+node" , redis , key ) ; redis . on ( "error" , function ( error ) { this . emit ( "nodeError" , error , key ) ; } ) ; } return redis ; } /**
     * Remove a node from the pool.
     */ private removeNode ( key ) { const { nodes } = this ; if ( nodes . all [ key ] ) { debug ( "Remove %s from the pool" , key ) ; delete nodes . all [ key ] ; } delete nodes . master [ key ] ; delete nodes . slave [ key ] ; } param { ( Array < string | number | object > ) } nodes memberof public reset ( nodes ) { debug ( "Reset with %O" , nodes ) ; const newNodes = { } ; nodes . forEach ( ( node ) => { const key = getNodeKey ( node ) ; if ( ! ( node . readOnly && newNodes [ key ] ) ) { newNodes [ key ] = node ; } } ) ; Object . keys ( this . nodes . all ) . forEach ( ( key ) => { if ( ! newNodes [ key ] ) { debug ( "Disconnect %s because the node does not hold any slot" , key ) ; this . nodes . all [ key ] . disconnect ( ) ; this . removeNode ( key ) ; } } ) ; Object . keys ( newNodes ) . forEach ( ( key ) => { const node = newNodes [ key ] ; this . findOrCreate ( node , node . readOnly ) ; } ) ; } }