'Calls the client asynchronously.
This sends a message to the client to invoke an Action. The run
action may send back many responses. These will be queued by the
framework until a status message is sent by the client. The status
message will cause the entire transaction to be committed to the
specified state.
Args:
action_cls: The function to call on the client.
request: The request to send to the client. If not specified (Or None) we
create a new RDFValue using the kwargs.
next_state: The state in this flow, that responses to this
message should go to.
client_id: rdf_client.ClientURN to send the request to.
request_data: A dict which 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). Valid values are
strings, unicode and protobufs.
start_time: Call the client at this time. This Delays the client request
for into the future.
**kwargs: These args will be used to construct the client action semantic
protobuf.
Raises:
FlowRunnerError: If next_state is not one of the allowed next states.
RuntimeError: The request passed to the client does not have the correct
type.'

def CallClient(self, action_cls, request=None, next_state=None, client_id=None, request_data=None, start_time=None, **kwargs):

    if (client_id is None):
        client_id = self.runner_args.client_id
    if (client_id is None):
        raise FlowRunnerError('CallClient()    is    used    on    a    flow    which    was    not    started    with    a    client.')
    if (not isinstance(client_id, rdf_client.ClientURN)):
        client_id = rdf_client.ClientURN(client_id)
    if (action_cls.in_rdfvalue is None):
        if request:
            raise RuntimeError(('Client    action    %s    does    not    expect    args.' % action_cls.__name__))
    elif (request is None):
        request = action_cls.in_rdfvalue(**kwargs)
    elif (not isinstance(request, action_cls.in_rdfvalue)):
        raise RuntimeError(('Client    action    expected    %s    but    got    %s' % (action_cls.in_rdfvalue, type(request))))
    outbound_id = self.GetNextOutboundId()
    state = rdf_flows.RequestState(id=outbound_id, session_id=self.session_id, next_state=next_state, client_id=client_id)
    if (request_data is not None):
        state.data = rdf_protodict.Dict(request_data)
    msg = rdf_flows.GrrMessage(session_id=utils.SmartUnicode(self.session_id), name=action_cls.__name__, request_id=outbound_id, priority=self.runner_args.priority, require_fastpoll=self.runner_args.require_fastpoll, queue=client_id.Queue(), payload=request, generate_task_id=True)
    cpu_usage = self.context.client_resources.cpu_usage
    if self.runner_args.cpu_limit:
        msg.cpu_limit = max(((self.runner_args.cpu_limit - cpu_usage.user_cpu_time) - cpu_usage.system_cpu_time), 0)
        if (msg.cpu_limit == 0):
            raise FlowRunnerError('CPU    limit    exceeded.')
    if self.runner_args.network_bytes_limit:
        msg.network_bytes_limit = max((self.runner_args.network_bytes_limit - self.context.network_bytes_sent), 0)
        if (msg.network_bytes_limit == 0):
            raise FlowRunnerError('Network    limit    exceeded.')
    state.request = msg
    self.QueueRequest(state, timestamp=start_time)
