def _ProcessCompletedRequests(self, notification):
    """Does the actual processing of the completed requests."""
    # 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 flow 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)

    # The flow is dead - remove all outstanding requests and responses.
    if not self.IsRunning():
      self.queue_manager.DestroyFlowStates(self.session_id)
      return

    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:
            continue

          if not responses:
            break

          # We are missing a needed request - maybe its not completed yet.
          if request.id > self.context.next_processed_request:
            stats_collector_instance.Get().IncrementCounter(
                "grr_response_out_of_order")
            break

          # Not the request we are looking for - we have seen it before
          # already.
          if request.id < self.context.next_processed_request:
            self.queue_manager.DeleteRequest(request)
            continue

          if 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.ReQueueRequest(request)
            break

          # If we get here its all good - run the flow.
          if self.IsRunning():
            self.flow_obj.HeartBeat()
            self.RunStateMethod(request.next_state, request, responses)

          # Quit early if we are no longer alive.
          else:
            break

          # 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
          self.DecrementOutstandingRequests()

        # Are there any more outstanding requests?
        if not self.OutstandingRequests():
          # Allow the flow to cleanup
          if self.IsRunning() and self.context.current_state != "End":
            self.RunStateMethod("End")

        # Rechecking the OutstandingRequests allows the End state (which was
        # called above) to issue further client requests - hence postpone
        # termination.
        if not self.OutstandingRequests():
          # TODO(user): Deprecate in favor of 'flow_completions' metric.
          stats_collector_instance.Get().IncrementCounter(
              "grr_flow_completed_count")

          stats_collector_instance.Get().IncrementCounter(
              "flow_completions", fields=[self.flow_obj.Name()])
          logging.debug(
              "Destroying session %s(%s) for client %s", self.session_id,
              self.flow_obj.Name(), self.runner_args.client_id)

          self.flow_obj.Terminate()

        # 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.flow_obj.Flush()
        continue

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