public void scheduleTasklet ( final int taskletId ) { synchronized ( stateLock ) { // If there are tasklets are pending to be executed, then that means that a // timer has already been scheduled for an aggregation. if ( ! outstandingTasklets ( ) ) { timer . schedule ( new Runnable ( ) { @ Override public void run ( ) { aggregateTasklets ( AggregateTriggerType . ALARM ) ; synchronized ( stateLock ) { // On the callback, if there are tasklets pending to be executed, that means that this alarm // was triggered by a previous alarm, so we should continue to trigger more alarms. Otherwise // we are done with tasklets for this aggregation function for now. // If more tasklets for this aggregation function arrive, it will be triggered by the outer // call to timer.schedule. if ( outstandingTasklets ( ) ) { timer . schedule ( this , taskletAggregationRequest . getPolicy ( ) . getPeriodMilliseconds ( ) , TimeUnit . MILLISECONDS ) ; } } } } , taskletAggregationRequest . getPolicy ( ) . getPeriodMilliseconds ( ) , TimeUnit . MILLISECONDS ) ; } // Add to pending tasklets, such that on the callback the timer can be refreshed. if ( ! pendingTasklets . containsKey ( taskletId ) ) { pendingTasklets . put ( taskletId , 0 ) ; } pendingTasklets . put ( taskletId , pendingTasklets . get ( taskletId ) + 1 ) ; } }
Schedule aggregation tasks on a Timer . Creates a new timer schedule for triggering the aggregation function if this is the first time the aggregation function has tasklets scheduled on it . Adds the Tasklet to pending Tasklets .