private Task < T > wrap ( int attempt ) { Task < T > retryTask = Task . async ( _policy . getName ( ) + ", attempt " + attempt , context -> { final SettablePromise < T > result = Promises . settable ( ) ; Task < T > task = _taskFunction . apply ( attempt ) ; final Task < T > recovery = Task . async ( _name + " recovery" , recoveryContext -> { final SettablePromise < T > recoveryResult = Promises . settable ( ) ; if ( task . isFailed ( ) ) { // Failed task will cause retry to be scheduled. ErrorClassification errorClassification = _policy . getErrorClassifier ( ) . apply ( task . getError ( ) ) ; retry ( attempt + 1 , task . getError ( ) , errorClassification , recoveryContext , recoveryResult ) ; } else { recoveryResult . done ( task . get ( ) ) ; } return recoveryResult ; } ) ; // Recovery task should run immediately after the original task to process its error. recovery . setPriority ( Priority . MAX_PRIORITY ) ; recovery . getShallowTraceBuilder ( ) . setSystemHidden ( true ) ; Promises . propagateResult ( recovery , result ) ; context . after ( task ) . run ( recovery ) ; context . run ( task ) ; return result ; } ) ; retryTask . getShallowTraceBuilder ( ) . setTaskType ( TaskType . RETRY . getName ( ) ) ; return retryTask ; }
Create a wrapped task with associated recovery task that will retry if necessary .