'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.
sync: If True start the flow inline on the calling thread, else schedule
a worker to actually start the child flow.
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.
Raises:
FlowRunnerError: If next_state is not one of the allowed next states.
Returns:
The URN of the child flow which was created.'

def CallFlow(self, flow_name=None, next_state=None, sync=True, request_data=None, client_id=None, base_session_id=None, **kwargs):

    client_id = (client_id or self.runner_args.client_id)
    state = rdf_flows.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)
    logs_urn = self._GetLogCollectionURN((kwargs.pop('logs_collection_urn', None) or self.runner_args.logs_collection_urn))
    write_intermediate = (kwargs.pop('write_intermediate_results', False) or self.runner_args.write_intermediate_results)
    try:
        event_id = self.runner_args.event_id
    except AttributeError:
        event_id = None
    child_urn = self.flow_obj.StartFlow(client_id=client_id, flow_name=flow_name, base_session_id=(base_session_id or self.session_id), event_id=event_id, request_state=state, token=self.token, notify_to_user=False, parent_flow=self.flow_obj, sync=sync, queue=self.runner_args.queue, write_intermediate_results=write_intermediate, logs_collection_urn=logs_urn, **kwargs)
    self.QueueRequest(state)
    return child_urn
