private void processRunnableOperations ( final Consumer < WorkflowSystem . OperationResult < DAT , RES , OP > > resultConsumer , final List < OP > shouldrun , final List < OP > shouldskip , final DAT inputData ) { for ( final OP operation : shouldrun ) { if ( shouldskip . contains ( operation ) ) { continue ; } pending . remove ( operation ) ; workflowEngine . event ( WorkflowSystemEventType . WillRunOperation , String . format ( "operation starting: %s" , operation ) , operation ) ; final ListenableFuture < RES > submit = executorService . submit ( ( ) -> operation . apply ( inputData ) ) ; inProcess . add ( operation ) ; futures . add ( submit ) ; FutureCallback < RES > callback = new FutureCallback < RES > ( ) { @ Override public void onSuccess ( final RES successResult ) { workflowEngine . event ( WorkflowSystemEventType . OperationSuccess , String . format ( "operation succeeded: %s" , successResult ) , successResult ) ; assert successResult != null ; WorkflowSystem . OperationResult < DAT , RES , OP > result = result ( successResult , operation ) ; resultConsumer . accept ( result ) ; stateChangeQueue . add ( successResult ) ; inProcess . remove ( operation ) ; } @ Override public void onFailure ( final Throwable t ) { workflowEngine . event ( WorkflowSystemEventType . OperationFailed , String . format ( "operation failed: %s" , t ) , t ) ; WorkflowSystem . OperationResult < DAT , RES , OP > result = result ( t , operation ) ; resultConsumer . accept ( result ) ; StateObj newFailureState = operation . getFailureState ( t ) ; if ( null != newFailureState && newFailureState . getState ( ) . size ( ) > 0 ) { WorkflowSystem . OperationCompleted < DAT > objectOperationCompleted = WorkflowEngine . dummyResult ( newFailureState ) ; stateChangeQueue . add ( objectOperationCompleted ) ; } inProcess . remove ( operation ) ; } } ; Futures . addCallback ( submit , callback , manager ) ; } }
Process the runnable operations