protected void copyDataOutBound ( int sourceTaskId , boolean isLocalSpout , TopologyAPI . StreamId streamId , HeronTuples . HeronDataTuple tuple , List < Integer > outTasks ) { boolean firstIteration = true ; boolean isAnchored = tuple . getRootsCount ( ) > 0 ; for ( Integer outTask : outTasks ) { long tupleKey = tupleCache . addDataTuple ( sourceTaskId , outTask , streamId , tuple , isAnchored ) ; if ( isAnchored ) { // Anchored tuple if ( isLocalSpout ) { // This is from a local spout. We need to maintain xors if ( firstIteration ) { xorManager . create ( sourceTaskId , tuple . getRoots ( 0 ) . getKey ( ) , tupleKey ) ; } else { xorManager . anchor ( sourceTaskId , tuple . getRoots ( 0 ) . getKey ( ) , tupleKey ) ; } } else { // Anchored emits from local bolt for ( HeronTuples . RootId rootId : tuple . getRootsList ( ) ) { HeronTuples . AckTuple t = HeronTuples . AckTuple . newBuilder ( ) . addRoots ( rootId ) . setAckedtuple ( tupleKey ) . build ( ) ; tupleCache . addEmitTuple ( sourceTaskId , rootId . getTaskid ( ) , t ) ; } } } firstIteration = false ; } }
Process HeronDataTuple and insert it into cache