def WriteFlowResponses(self, responses):
    """Writes FlowMessages and updates corresponding requests."""
    status_available = set()
    requests_updated = set()
    task_ids_by_request = {}

    for response in responses:
      flow_key = (response.client_id, response.flow_id)
      if flow_key not in self.flows:
        logging.error("Received response for unknown flow %s, %s.",
                      response.client_id, response.flow_id)
        continue

      request_dict = self.flow_requests.get(flow_key, {})
      if response.request_id not in request_dict:
        logging.error("Received response for unknown request %s, %s, %d.",
                      response.client_id, response.flow_id, response.request_id)
        continue

      response_dict = self.flow_responses.setdefault(flow_key, {})
      clone = response.Copy()
      clone.timestamp = rdfvalue.RDFDatetime.Now()

      response_dict.setdefault(response.request_id,
                               {})[response.response_id] = clone

      if isinstance(response, rdf_flow_objects.FlowStatus):
        status_available.add(response)

      request_key = (response.client_id, response.flow_id, response.request_id)
      requests_updated.add(request_key)
      try:
        task_ids_by_request[request_key] = response.task_id
      except AttributeError:
        pass

    # Every time we get a status we store how many responses are expected.
    for status in status_available:
      request_dict = self.flow_requests[(status.client_id, status.flow_id)]
      request = request_dict[status.request_id]
      request.nr_responses_expected = status.response_id

    # And we check for all updated requests if we need to process them.
    needs_processing = []
    for client_id, flow_id, request_id in requests_updated:
      flow_key = (client_id, flow_id)
      request_dict = self.flow_requests[flow_key]
      request = request_dict[request_id]
      if request.nr_responses_expected and not request.needs_processing:
        response_dict = self.flow_responses.setdefault(flow_key, {})
        responses = response_dict.get(request_id, {})

        if len(responses) == request.nr_responses_expected:
          request.needs_processing = True
          self._DeleteClientActionRequest(client_id, flow_id, request_id)

          flow = self.flows[flow_key]
          if flow.next_request_to_process == request_id:
            needs_processing.append(
                rdf_flows.FlowProcessingRequest(
                    client_id=client_id, flow_id=flow_id))

    if needs_processing:
      self.WriteFlowProcessingRequests(needs_processing)
def WriteFlowResponses ( self , responses ) : status_available = set ( ) requests_updated = set ( ) task_ids_by_request = { } for response in responses : flow_key = ( response . client_id , response . flow_id ) if flow_key not in self . flows : logging . error ( "Received response for unknown flow %s, %s." , response . client_id , response . flow_id ) continue request_dict = self . flow_requests . get ( flow_key , { } ) if response . request_id not in request_dict : logging . error ( "Received response for unknown request %s, %s, %d." , response . client_id , response . flow_id , response . request_id ) continue response_dict = self . flow_responses . setdefault ( flow_key , { } ) clone = response . Copy ( ) clone . timestamp = rdfvalue . RDFDatetime . Now ( ) response_dict . setdefault ( response . request_id , { } ) [ response . response_id ] = clone if isinstance ( response , rdf_flow_objects . FlowStatus ) : status_available . add ( response ) request_key = ( response . client_id , response . flow_id , response . request_id ) requests_updated . add ( request_key ) try : task_ids_by_request [ request_key ] = response . task_id except AttributeError : pass for status in status_available : request_dict = self . flow_requests [ ( status . client_id , status . flow_id ) ] request = request_dict [ status . request_id ] request . nr_responses_expected = status . response_id needs_processing = [ ] for client_id , flow_id , request_id in requests_updated : flow_key = ( client_id , flow_id ) request_dict = self . flow_requests [ flow_key ] request = request_dict [ request_id ] if request . nr_responses_expected and not request . needs_processing : response_dict = self . flow_responses . setdefault ( flow_key , { } ) responses = response_dict . get ( request_id , { } ) if len ( responses ) == request . nr_responses_expected : request . needs_processing = True self . _DeleteClientActionRequest ( client_id , flow_id , request_id ) flow = self . flows [ flow_key ] if flow . next_request_to_process == request_id : needs_processing . append ( rdf_flows . FlowProcessingRequest ( client_id = client_id , flow_id = flow_id ) ) if needs_processing : self . WriteFlowProcessingRequests ( needs_processing )
Writes FlowMessages and updates corresponding requests.