docs: 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);
      }