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);
        }
    }
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

@param shouldrun  operations to run
@param shouldskip operations to skip
@param inputData  input data for the currently runnable operations