docs: 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)