@ VisibleForTesting protected void executionComplete ( Throwable exception ) { Collection < QueueItem > toFail = null ; Throwable failEx = null ; synchronized ( this . stateLock ) { // Release the spot occupied by this item's execution. this . activeCount -- ; if ( exception != null && ! this . closed ) { // Need to fail all future items and close to prevent new items from being processed. failEx = new ProcessingException ( "A previous item failed to commit. Cannot process new items." , exception ) ; toFail = new ArrayList <> ( this . pendingItems ) ; this . pendingItems . clear ( ) ; this . closed = true ; } if ( this . emptyNotifier != null && this . activeCount == 0 && this . pendingItems . isEmpty ( ) ) { // We were asked to notify when we were empty. this . emptyNotifier . release ( ) ; this . emptyNotifier = null ; } } if ( toFail != null ) { for ( QueueItem q : toFail ) { q . result . completeExceptionally ( failEx ) ; } return ; } // We need to ensure the items are still executed in order. Once out of the main sync block, it is possible that // this callback may be invoked concurrently after various completions. As such, we need a special handler for // these, using its own synchronization, different from the main one, so that we don't hold that stateLock for too long. synchronized ( this . processingLock ) { while ( true ) { QueueItem toProcess ; synchronized ( this . stateLock ) { if ( hasCapacity ( ) && ! this . pendingItems . isEmpty ( ) ) { // We have spare capacity and we have something to process. Dequeue it and reserve the spot. toProcess = this . pendingItems . pollFirst ( ) ; this . activeCount ++ ; } else { // No capacity left or no more pending items. break ; } } Futures . completeAfter ( ( ) -> processInternal ( toProcess . data ) , toProcess . result ) ; } } }
Callback that is invoked when an item has completed execution .