Python grpc 模块，_channel() 实例源码

def future(self, request, timeout=None, metadata=None, credentials=None):
        state, operations, deadline, deadline_timespec, rendezvous = self._prepare(
            request, timeout, metadata)
        if rendezvous:
            return rendezvous
        else:
            channel = self._channel.get_channel()
            call, drive_call = channel.managed_call(
                None, 0, self._method, None, deadline_timespec)

            if credentials is not None:
                call.set_credentials(credentials._credentials)

            event_handler = _event_handler(
                state, call, self._response_deserializer)
            with state.condition:
                call_error = call.start_client_batch(
                    cygrpc.Operations(operations), event_handler)
                if call_error != cygrpc.CallError.ok:
                    _call_error_set_RPCstate(state, call_error, metadata)
                    return _Rendezvous(state, None, None, deadline)
                drive_call()
            return _Rendezvous(state, call, self._response_deserializer,
                               deadline)

def __init__(
            self, channel, method, request_serializer,
            response_deserializer):
        self._channel = channel
        self._method = method
        self._request_serializer = request_serializer
        self._response_deserializer = response_deserializer

def __call__(self, request, timeout=None, metadata=None, credentials=None):
        channel = self._channel.get_channel()
        state, call, deadline = self._blocking(
            channel.channel, request, timeout, metadata, credentials)
        return _end_unary_response_blocking(state, call, False, deadline)

def with_call(self, request, timeout=None, metadata=None, credentials=None):
        channel = self._channel.get_channel()
        state, call, deadline = self._blocking(
            channel.channel, request, timeout, metadata, credentials)
        return _end_unary_response_blocking(state, call, True, deadline)

def __init__(self, channel, method, request_serializer,
                 response_deserializer):
        self._channel = channel
        self._method = method
        self._request_serializer = request_serializer
        self._response_deserializer = response_deserializer

def __init__(
            self, channel, method, request_serializer, response_deserializer):
        self._channel = channel
        self._method = method
        self._request_serializer = request_serializer
        self._response_deserializer = response_deserializer

def __call__(
            self, request_iterator, timeout=None, metadata=None,
            credentials=None):
        channel = self._channel.get_channel()
        state, call, deadline = self._blocking(
            channel.channel, request_iterator, timeout, metadata,
            credentials)
        return _end_unary_response_blocking(state, call, False, deadline)

def with_call(
            self, request_iterator, timeout=None, metadata=None,
            credentials=None):
        channel = self._channel.get_channel()
        state, call, deadline = self._blocking(
            channel.channel, request_iterator, timeout, metadata, credentials)
        return _end_unary_response_blocking(state, call, True, deadline)

def future(self,
               request_iterator,
               timeout=None,
               metadata=None,
               credentials=None):
        deadline, deadline_timespec = _deadline(timeout)
        state = _RPCState(_STREAM_UNARY_INITIAL_DUE, None, None, None, None)
        channel = self._channel.get_channel()
        call, drive_call = channel.managed_call(
            None, 0, self._method, None, deadline_timespec)
        if credentials is not None:
            call.set_credentials(credentials._credentials)
        event_handler = _event_handler(state, call, self._response_deserializer)
        with state.condition:
            call.start_client_batch(
                cygrpc.Operations(
                    (cygrpc.operation_receive_initial_metadata(_EMPTY_FLAGS),)),
                event_handler)
            operations = (
                cygrpc.operation_send_initial_metadata(
                    _common.to_cygrpc_metadata(metadata), _EMPTY_FLAGS),
                cygrpc.operation_receive_message(_EMPTY_FLAGS),
                cygrpc.operation_receive_status_on_client(_EMPTY_FLAGS),)
            call_error = call.start_client_batch(
                cygrpc.Operations(operations), event_handler)
            if call_error != cygrpc.CallError.ok:
                _call_error_set_RPCstate(state, call_error, metadata)
                return _Rendezvous(state, None, None, deadline)
            drive_call()
            _consume_request_iterator(request_iterator, state, call,
                                      self._request_serializer)
        return _Rendezvous(state, call, self._response_deserializer, deadline)

def __init__(
            self, channel, method, request_serializer, response_deserializer):
        self._channel = channel
        self._method = method
        self._request_serializer = request_serializer
        self._response_deserializer = response_deserializer

def invoke(self, stub, method_name, request, metadata, retry=1):
        """
        Invoke a gRPC call to the remote server and return the response.
        :param stub: Reference to the *_pb2 service stub
        :param method_name: The method name inside the service stub
        :param request: The request protobuf message
        :param metadata: [(str, str), (str, str), ...]
        :return: The response protobuf message and returned trailing metadata
        """

        if not self.connected:
            raise ServiceUnavailable()

        try:
            method = getattr(stub(self.channel), method_name)
            response, rendezvous = method.with_call(request, metadata=metadata)
            returnValue((response, rendezvous.trailing_metadata()))

        except grpc._channel._Rendezvous, e:
            code = e.code()
            if code == grpc.StatusCode.UNAVAILABLE:
                e = ServiceUnavailable()

                if self.connected:
                    self.connected = False
                    yield self.connect()
                    if retry > 0:
                        response = yield self.invoke(stub, method_name,
                                                     request, metadata,
                                                     retry=retry - 1)
                        returnValue(response)

            elif code in (
                    grpc.StatusCode.NOT_FOUND,
                    grpc.StatusCode.INVALID_ARGUMENT,
                    grpc.StatusCode.ALREADY_EXISTS):

                pass  # don't log error, these occur naturally

            else:
                log.exception(e)

            raise e

    # Below is an adaptation of Google's MessageToDict() which includes
    # protobuf options extensions