code: def ProcessFlow(self, flow_processing_request):
      
  
      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))
docs: The callback for the flow processing queue.