def FetchRequestsAndResponses(self, session_id, timestamp=None):
    """Fetches all outstanding requests and responses for this flow.

    We first cache all requests and responses for this flow in memory to
    prevent round trips.

    Args:
      session_id: The session_id to get the requests/responses for.
      timestamp: Tuple (start, end) with a time range. Fetched requests and
        responses will have timestamp in this range.

    Yields:
      an tuple (request protobufs, list of responses messages) in ascending
      order of request ids.

    Raises:
      MoreDataException: When there is more data available than read by the
                         limited query.
    """
    if timestamp is None:
      timestamp = (0, self.frozen_timestamp or rdfvalue.RDFDatetime.Now())

    num_requests = 0
    for request, responses in self.data_store.ReadRequestsAndResponses(
        session_id,
        timestamp=timestamp,
        request_limit=self.request_limit,
        response_limit=self.response_limit):
      yield (request, responses)
      num_requests += 1

    if num_requests >= self.request_limit:
      raise MoreDataException()