def _CallFlowLegacy(self,
                      flow_name=None,
                      next_state=None,
                      request_data=None,
                      client_id=None,
                      base_session_id=None,
                      **kwargs):
    """Creates a new flow and send its responses to a state.

    This creates a new flow. The flow may send back many responses which will be
    queued by the framework until the flow terminates. The final status message
    will cause the entire transaction to be committed to the specified state.

    Args:
       flow_name: The name of the flow to invoke.
       next_state: The state in this flow, that responses to this message should
         go to.
       request_data: Any dict provided here will be available in the
         RequestState protobuf. The Responses object maintains a reference to
         this protobuf for use in the execution of the state method. (so you can
         access this data by responses.request). There is no format mandated on
         this data but it may be a serialized protobuf.
       client_id: If given, the flow is started for this client.
       base_session_id: A URN which will be used to build a URN.
       **kwargs: Arguments for the child flow.

    Returns:
       The URN of the child flow which was created.

    Raises:
       RuntimeError: In case of no cpu quota left to start more clients.
    """
    client_id = client_id or self.runner_args.client_id

    # We prepare a request state, and add it to our queue - any
    # responses from the child flow will return to the request state
    # and the stated next_state. Note however, that there is no
    # client_id or actual request message here because we directly
    # invoke the child flow rather than queue anything for it.
    state = rdf_flow_runner.RequestState(
        id=self.GetNextOutboundId(),
        session_id=utils.SmartUnicode(self.session_id),
        client_id=client_id,
        next_state=next_state,
        response_count=0)

    if request_data:
      state.data = rdf_protodict.Dict().FromDict(request_data)

    # Pass our logs collection urn to the flow object.
    logs_urn = self.hunt_obj.logs_collection_urn

    # If we were called with write_intermediate_results, propagate down to
    # child flows.  This allows write_intermediate_results to be set to True
    # either at the top level parent, or somewhere in the middle of
    # the call chain.
    write_intermediate = kwargs.pop("write_intermediate_results", False)

    # Create the new child flow but do not notify the user about it.
    child_urn = self.hunt_obj.StartAFF4Flow(
        base_session_id=base_session_id or self.session_id,
        client_id=client_id,
        cpu_limit=self._GetSubFlowCPULimit(),
        flow_name=flow_name,
        logs_collection_urn=logs_urn,
        network_bytes_limit=self._GetSubFlowNetworkLimit(),
        notify_to_user=False,
        parent_flow=self.hunt_obj,
        queue=self.runner_args.queue,
        request_state=state,
        sync=False,
        token=self.token,
        write_intermediate_results=write_intermediate,
        **kwargs)

    self.QueueRequest(state)

    return child_urn