public void receive ( IncomingT element ) { assert adminExecutor . inEventLoop ( ) ; if ( stopped ) { return ; } if ( window . isZero ( ) || maxEvents == 1 ) { LOG . debug ( "Received {}, flushing immediately (window = {}, maxEvents = {})" , element , window , maxEvents ) ; onFlush . accept ( coalescer . apply ( ImmutableList . of ( element ) ) ) ; } else { currentBatch . add ( element ) ; if ( currentBatch . size ( ) == maxEvents ) { LOG . debug ( "Received {}, flushing immediately (because {} accumulated events)" , element , maxEvents ) ; flushNow ( ) ; } else { LOG . debug ( "Received {}, scheduling next flush in {}" , element , window ) ; scheduleFlush ( ) ; } } }
This must be called on eventExecutor too .