public < ReturnType > CompletableFuture < ReturnType > addWithFilter ( Predicate < KeyType > keyFilter , Supplier < CompletableFuture < ? extends ReturnType > > toRun ) { CompletableFuture < ReturnType > result = new CompletableFuture <> ( ) ; ArrayList < CompletableFuture < ? > > existingTasks = new ArrayList <> ( ) ; synchronized ( this . queue ) { Exceptions . checkNotClosed ( this . closed , this ) ; // Collect all currently executing tasks for the given keys. for ( val e : this . queue . entrySet ( ) ) { if ( keyFilter . test ( e . getKey ( ) ) ) { existingTasks . add ( e . getValue ( ) ) ; } } executeAfterIfNeeded ( existingTasks , toRun , result ) ; // Record the action. this . filterQueue . put ( keyFilter , result ) ; } executeNowIfNeeded ( existingTasks , toRun , result ) ; // Cleanup: if this was the last task in the queue, then clean up the queue. result . whenComplete ( ( r , ex ) -> cleanupFilter ( keyFilter ) ) ; return result ; }
Queues up a new task to execute subject to the dependency keys that match the given filter .