private void addSpoutsTasks ( ) { // Register spoutTasks Runnable spoutTasks = new Runnable ( ) { @ Override public void run ( ) { spoutMetrics . updateTaskRunCount ( ) ; // Check whether we should produce more tuples if ( isProduceTuple ( ) ) { spoutMetrics . updateProduceTupleCount ( ) ; produceTuple ( ) ; // Though we may execute MAX_READ tuples, finally we will packet it as // one outgoingPacket and push to out queues // Notice: Tuples are not necessary emitted during nextTuple methods. We could emit // tuples as long as we invoke collector.emit(...) collector . sendOutTuples ( ) ; } if ( ! collector . isOutQueuesAvailable ( ) ) { spoutMetrics . updateOutQueueFullCount ( ) ; } // Check if we have any message to process anyway readTuplesAndExecute ( streamInQueue ) ; if ( ackEnabled ) { // Update the pending-to-be-acked tuples counts spoutMetrics . updatePendingTuplesCount ( collector . numInFlight ( ) ) ; } else { doImmediateAcks ( ) ; } // If we have more work to do if ( isContinueWork ( ) ) { spoutMetrics . updateContinueWorkCount ( ) ; looper . wakeUp ( ) ; } } } ; looper . addTasksOnWakeup ( spoutTasks ) ; // Look for the timeout's tuples if ( enableMessageTimeouts ) { lookForTimeouts ( ) ; } InstanceUtils . prepareTimerEvents ( looper , helper ) ; }
Tasks happen in every time looper is waken up