docs: def ReadFlowRequestsReadyForProcessing(self,
                                           client_id,
                                           flow_id,
                                           next_needed_request=None):
      """Reads all requests for a flow that can be processed by the worker."""
      request_dict = self.flow_requests.get((client_id, flow_id), {})
      response_dict = self.flow_responses.get((client_id, flow_id), {})
  
      res = {}
      for request_id in sorted(request_dict):
        # Ignore outdated requests.
        if request_id < next_needed_request:
          continue
        # The request we are currently looking for is not in yet, we are done.
        if request_id != next_needed_request:
          break
        request = request_dict[request_id]
        if not request.needs_processing:
          break
  
        responses = sorted(
            itervalues(response_dict.get(request_id, {})),
            key=lambda response: response.response_id)
        # Serialize/deserialize responses to better simulate the
        # real DB behavior (where serialization/deserialization is almost
        # guaranteed to be done).
        # TODO(user): change mem-db implementation to do
        # serialization/deserialization everywhere in a generic way.
        responses = [
            r.__class__.FromSerializedString(r.SerializeToString())
            for r in responses
        ]
        res[request_id] = (request, responses)
        next_needed_request += 1
  
      return res