The callback for the flow processing queue.

def ProcessFlow(self, flow_processing_request):
    """The callback for the flow processing queue."""

    client_id = flow_processing_request.client_id
    flow_id = flow_processing_request.flow_id

    data_store.REL_DB.AckFlowProcessingRequests([flow_processing_request])

    try:
      rdf_flow = data_store.REL_DB.LeaseFlowForProcessing(
          client_id, flow_id, processing_time=rdfvalue.Duration("6h"))
    except db.ParentHuntIsNotRunningError:
      flow_base.TerminateFlow(client_id, flow_id, "Parent hunt stopped.")
      return

    logging.info("Processing Flow %s/%s (%s).", client_id, flow_id,
                 rdf_flow.flow_class_name)

    flow_cls = registry.FlowRegistry.FlowClassByName(rdf_flow.flow_class_name)
    flow_obj = flow_cls(rdf_flow)

    if not flow_obj.IsRunning():
      raise ValueError(
          "Received a request to process flow %s on client %s that is not "
          "running." % (flow_id, client_id))

    processed = flow_obj.ProcessAllReadyRequests()
    if processed == 0:
      raise ValueError(
          "Unable to process any requests for flow %s on client %s." %
          (flow_id, client_id))

    while not self._ReleaseProcessedFlow(flow_obj):
      processed = flow_obj.ProcessAllReadyRequests()
      if processed == 0:
        raise ValueError(
            "%s/%s: ReleaseProcessedFlow returned false but no "
            "request could be processed (next req: %d)." %
            (client_id, flow_id, flow_obj.rdf_flow.next_request_to_process))