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

def ReadFlowRequestsReadyForProcessing(self,
                                         client_id,
                                         flow_id,
                                         next_needed_request,
                                         cursor=None):
    """Reads all requests for a flow that can be processed by the worker."""
    query = ("SELECT request, needs_processing, responses_expected, "
             "UNIX_TIMESTAMP(timestamp) "
             "FROM flow_requests "
             "WHERE client_id=%s AND flow_id=%s")
    args = [db_utils.ClientIDToInt(client_id), db_utils.FlowIDToInt(flow_id)]
    cursor.execute(query, args)

    requests = {}
    for req, needs_processing, responses_expected, ts in cursor.fetchall():
      if not needs_processing:
        continue

      request = rdf_flow_objects.FlowRequest.FromSerializedString(req)
      request.needs_processing = needs_processing
      request.nr_responses_expected = responses_expected
      request.timestamp = mysql_utils.TimestampToRDFDatetime(ts)
      requests[request.request_id] = request

    query = ("SELECT response, status, iterator, UNIX_TIMESTAMP(timestamp) "
             "FROM flow_responses "
             "WHERE client_id=%s AND flow_id=%s")
    cursor.execute(query, args)

    responses = {}
    for res, status, iterator, ts in cursor.fetchall():
      if status:
        response = rdf_flow_objects.FlowStatus.FromSerializedString(status)
      elif iterator:
        response = rdf_flow_objects.FlowIterator.FromSerializedString(iterator)
      else:
        response = rdf_flow_objects.FlowResponse.FromSerializedString(res)
      response.timestamp = mysql_utils.TimestampToRDFDatetime(ts)
      responses.setdefault(response.request_id, []).append(response)

    res = {}
    while next_needed_request in requests:
      req = requests[next_needed_request]
      sorted_responses = sorted(
          responses.get(next_needed_request, []), key=lambda r: r.response_id)
      res[req.request_id] = (req, sorted_responses)
      next_needed_request += 1

    return res