@ Override public < TInput , TOutput > VortexFuture < TOutput > enqueueTasklet ( final VortexFunction < TInput , TOutput > function , final TInput input , final Optional < FutureCallback < TOutput > > callback ) { // TODO[REEF-500]: Simple duplicate Vortex Tasklet launch. final VortexFuture < TOutput > vortexFuture ; final int id = taskletIdCounter . getAndIncrement ( ) ; if ( callback . isPresent ( ) ) { vortexFuture = new VortexFuture <> ( executor , this , id , callback . get ( ) ) ; } else { vortexFuture = new VortexFuture <> ( executor , this , id ) ; } final Tasklet tasklet = new Tasklet <> ( id , Optional . < Integer > empty ( ) , function , input , vortexFuture ) ; putDelegate ( Collections . singletonList ( tasklet ) , vortexFuture ) ; this . pendingTasklets . addLast ( tasklet ) ; return vortexFuture ; }
Add a new tasklet to pendingTasklets .