private void continueProcessing ( ) { try { while ( ! Thread . currentThread ( ) . isInterrupted ( ) ) { HashMap < String , String > changes = new HashMap <> ( ) ; //load all changes already in the queue getAvailableChanges ( changes ) ; if ( changes . isEmpty ( ) ) { if ( inProcess . isEmpty ( ) ) { //no pending operations, signalling no new state changes will occur workflowEngine . event ( WorkflowSystemEventType . EndOfChanges , "No more state changes expected, finishing workflow." ) ; return ; } if ( Thread . currentThread ( ) . isInterrupted ( ) ) { break ; } waitForChanges ( changes ) ; } if ( changes . isEmpty ( ) ) { //no changes within sleep time, try again continue ; } getContextGlobalData ( changes ) ; //handle state changes processStateChanges ( changes ) ; if ( workflowEngine . isWorkflowEndState ( workflowEngine . getState ( ) ) ) { workflowEngine . event ( WorkflowSystemEventType . WorkflowEndState , "Workflow end state reached." ) ; return ; } processOperations ( results :: add ) ; } } catch ( InterruptedException e ) { Thread . currentThread ( ) . interrupt ( ) ; } if ( Thread . currentThread ( ) . isInterrupted ( ) ) { workflowEngine . event ( WorkflowSystemEventType . Interrupted , "Engine interrupted, stopping engine..." ) ; cancelFutures ( ) ; interrupted = Thread . interrupted ( ) ; } awaitFutures ( ) ; }
Continue processing from current state