Go through the list of requests and process the completed ones.

    We take a snapshot in time of all requests and responses for this hunt. We
    then process as many completed requests as possible. If responses are not
    quite here we leave it for next time.

    Args:
      notification: The notification object that triggered this processing.
      thread_pool: The thread pool to process the responses on.

def ProcessCompletedRequests(self, notification, thread_pool):
    """Go through the list of requests and process the completed ones.

    We take a snapshot in time of all requests and responses for this hunt. We
    then process as many completed requests as possible. If responses are not
    quite here we leave it for next time.

    Args:
      notification: The notification object that triggered this processing.
      thread_pool: The thread pool to process the responses on.
    """
    # First ensure that client messages are all removed. NOTE: We make a new
    # queue manager here because we want only the client messages to be removed
    # ASAP. This must happen before we actually run the hunt to ensure the
    # client requests are removed from the client queues.
    with queue_manager.QueueManager(token=self.token) as manager:
      for request, _ in manager.FetchCompletedRequests(
          self.session_id, timestamp=(0, notification.timestamp)):
        # Requests which are not destined to clients have no embedded request
        # message.
        if request.HasField("request"):
          manager.DeQueueClientRequest(request.request)

    processing = []
    while True:
      try:
        # Here we only care about completed requests - i.e. those requests with
        # responses followed by a status message.
        for request, responses in self.queue_manager.FetchCompletedResponses(
            self.session_id, timestamp=(0, notification.timestamp)):

          if request.id == 0 or not responses:
            continue

          # Do we have all the responses here? This can happen if some of the
          # responses were lost.
          if len(responses) != responses[-1].response_id:
            # If we can retransmit do so. Note, this is different from the
            # automatic retransmission facilitated by the task scheduler (the
            # Task.task_ttl field) which would happen regardless of these.
            if request.transmission_count < 5:
              stats_collector_instance.Get().IncrementCounter(
                  "grr_request_retransmission_count")
              request.transmission_count += 1
              self.QueueRequest(request)
            break

          # If we get here its all good - run the hunt.
          self.hunt_obj.HeartBeat()
          self._Process(
              request, responses, thread_pool=thread_pool, events=processing)

          # At this point we have processed this request - we can remove it and
          # its responses from the queue.
          self.queue_manager.DeleteRequest(request)
          self.context.next_processed_request += 1

        # We are done here.
        return

      except queue_manager.MoreDataException:
        # Join any threads.
        for event in processing:
          event.wait()

        # We did not read all the requests/responses in this run in order to
        # keep a low memory footprint and have to make another pass.
        self.FlushMessages()
        self.hunt_obj.Flush()
        continue

      finally:
        # Join any threads.
        for event in processing:
          event.wait()