import * as PromiseContainer from "./promiseContainer" ; import { flatten , isArguments , noop } from "./utils/lodash" ; import * as calculateSlot from "cluster-key-slot" ; import asCallback from "standard-as-callback" ; export const kExec = Symbol ( "exec" ) ; export const kCallbacks = Symbol ( "callbacks" ) ; export const notAllowedAutoPipelineCommands = [ "auth" , "info" , "script" , "quit" , "cluster" , "pipeline" , "multi" , "subscribe" , "psubscribe" , "unsubscribe" , "unpsubscribe" , ] ; function executeAutoPipeline ( client , slotKey ) { if ( client . _runningAutoPipelines . has ( slotKey ) ) { return ; } if ( ! client . _autoPipelines . has ( slotKey ) ) { return ; } client . _runningAutoPipelines . add ( slotKey ) ; const pipeline = client . _autoPipelines . get ( slotKey ) ; client . _autoPipelines . delete ( slotKey ) ; const callbacks = pipeline [ kCallbacks ] ; pipeline [ kCallbacks ] = null ; pipeline . exec ( function ( err , results ) { client . _runningAutoPipelines . delete ( slotKey ) ; if ( err ) { for ( let i = 0 ; i < callbacks . length ; i ++ ) { process . nextTick ( callbacks [ i ] , err ) ; } } else { for ( let i = 0 ; i < callbacks . length ; i ++ ) { process . nextTick ( callbacks [ i ] , ... results [ i ] ) ; } } if ( client . _autoPipelines . has ( slotKey ) ) { executeAutoPipeline ( client , slotKey ) ; } } ) ; } export function shouldUseAutoPipelining ( client , functionName , commandName ) { return ( functionName && client . options . enableAutoPipelining && ! client . isPipeline && ! notAllowedAutoPipelineCommands . includes ( commandName ) && ! client . options . autoPipeliningIgnoredCommands . includes ( commandName ) ) ; } private export function getFirstValueInFlattenedArray ( args ) { for ( let i = 0 ; i < args . length ; i ++ ) { const arg = args [ i ] ; if ( typeof arg === "string" ) { return arg ; } else if ( Array . isArray ( arg ) || isArguments ( arg ) ) { if ( arg . length === 0 ) { continue ; } return arg [ 0 ] ; } const flattened = flatten ( [ arg ] ) ; if ( flattened . length > 0 ) { return flattened [ 0 ] ; } } return undefined ; } export function executeWithAutoPipelining ( client , functionName , commandName , args , callback ) { const CustomPromise = PromiseContainer . get ( ) ; if ( client . isCluster && ! client . slots . length ) { if ( client . status === "wait" ) client . connect ( ) . catch ( noop ) ; return asCallback ( new CustomPromise ( function ( resolve , reject ) { client . delayUntilReady ( ( err ) => { if ( err ) { reject ( err ) ; return ; } executeWithAutoPipelining ( client , functionName , commandName , args , null ) . then ( resolve , reject ) ; } ) ; } ) , callback ) ; } const prefix = client . options . keyPrefix || "" ; const slotKey = client . isCluster ? client . slots [ calculateSlot ( `${ prefix }${ getFirstValueInFlattenedArray ( args ) }` ) ] . join ( "," ) : "main" ; if ( ! client . _autoPipelines . has ( slotKey ) ) { const pipeline = client . pipeline ( ) ; pipeline [ kExec ] = false ; pipeline [ kCallbacks ] = [ ] ; client . _autoPipelines . set ( slotKey , pipeline ) ; } const pipeline = client . _autoPipelines . get ( slotKey ) ; if ( ! pipeline [ kExec ] ) { pipeline [ kExec ] = true ; setImmediate ( executeAutoPipeline , client , slotKey ) ; } const autoPipelinePromise = new CustomPromise ( function ( resolve , reject ) { pipeline [ kCallbacks ] . push ( function ( err , value ) { if ( err ) { reject ( err ) ; return ; } resolve ( value ) ; } ) ; pipeline [ functionName ] ( ... args ) ; } ) ; return asCallback ( autoPipelinePromise , callback ) ; }