public Messages process ( Messages messages , InterpreterListener interpreterListener , State state ) { interpreterListener . startProcessing ( ) ; // message id + stream id final Set < Tuple2 < String , String > > processingBlacklist = Sets . newHashSet ( ) ; final List < Message > toProcess = Lists . newArrayList ( messages ) ; final List < Message > fullyProcessed = Lists . newArrayListWithExpectedSize ( toProcess . size ( ) ) ; while ( ! toProcess . isEmpty ( ) ) { final MessageCollection currentSet = new MessageCollection ( toProcess ) ; // we'll add them back below toProcess . clear ( ) ; for ( Message message : currentSet ) { final String msgId = message . getId ( ) ; // this makes a copy of the list, which is mutated later in updateStreamBlacklist // it serves as a worklist, to keep track of which <msg, stream> tuples need to be re-run again final Set < String > initialStreamIds = message . getStreams ( ) . stream ( ) . map ( Stream :: getId ) . collect ( Collectors . toSet ( ) ) ; final ImmutableSet < Pipeline > pipelinesToRun = selectPipelines ( interpreterListener , processingBlacklist , message , initialStreamIds , state . getStreamPipelineConnections ( ) ) ; toProcess . addAll ( processForResolvedPipelines ( message , msgId , pipelinesToRun , interpreterListener , state ) ) ; // add each processed message-stream combination to the blacklist set and figure out if the processing // has added a stream to the message, in which case we need to cycle and determine whether to process // its pipeline connections, too boolean addedStreams = updateStreamBlacklist ( processingBlacklist , message , initialStreamIds ) ; potentiallyDropFilteredMessage ( message ) ; // go to 1 and iterate over all messages again until no more streams are being assigned if ( ! addedStreams || message . getFilterOut ( ) ) { log . debug ( "[{}] no new streams matches or dropped message, not running again" , msgId ) ; fullyProcessed . add ( message ) ; } else { // process again, we've added a stream log . debug ( "[{}] new streams assigned, running again for those streams" , msgId ) ; toProcess . add ( message ) ; } } } interpreterListener . finishProcessing ( ) ; // 7. return the processed messages return new MessageCollection ( fullyProcessed ) ; }
Evaluates all pipelines that apply to the given messages based on the current stream routing of the messages .