Reads all requests for a flow that can be processed by the worker.

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