class Wrapper(ContextManager[None]):
    """Special wrapper for coroutines to wake them up in case of some error.

    Example:

    .. code-block:: python3

        w = Wrapper()

        async def blocking_call():
            with w:
                await asyncio.sleep(10)

        # and somewhere else:
        w.cancel(NoNeedToWaitError('With explanation'))

    """
    _error: Optional[Exception] = None

    cancelled: Optional[bool] = None
    cancel_failed: Optional[bool] = None

    def __init__(self) -> None:
        self._tasks: Set['asyncio.Task[Any]'] = set()

    def __enter__(self) -> None:
        if self._error is not None:
            raise self._error

        task = _current_task()
        if task is None:
            raise RuntimeError('Called not inside a task')

        self._tasks.add(task)

    def __exit__(
        self,
        exc_type: Optional[Type[BaseException]],
        exc_val: Optional[BaseException],
        exc_tb: Optional[TracebackType],
    ) -> None:
        task = _current_task()
        assert task
        self._tasks.discard(task)
        if self._error is not None:
            self.cancel_failed = exc_type is not asyncio.CancelledError
            raise self._error

    def cancel(self, error: Exception) -> None:
        self._error = error
        for task in self._tasks:
            task.cancel()
        self.cancelled = True class DeadlineWrapper(Wrapper):
    """Deadline wrapper to specify deadline once for any number of awaiting
    method calls.

    Example:

    .. code-block:: python3

        dw = DeadlineWrapper()

        with dw.start(deadline):
            await handle_request()

        # somewhere during request handling:

        async def blocking_call():
            with dw:
                await asyncio.sleep(10)

    """
    @contextmanager
    def start(self, deadline: 'Deadline') -> Iterator[None]:
        timeout = deadline.time_remaining()
        if not timeout:
            raise asyncio.TimeoutError('Deadline exceeded')

        def callback() -> None:
            self.cancel(asyncio.TimeoutError('Deadline exceeded'))

        loop = asyncio.get_event_loop()
        timer = loop.call_later(timeout, callback)
        try:
            yield
        finally:
            timer.cancel() class Status(enum.Enum):
    """Predefined gRPC status codes represented as enum

    See also: https://github.com/grpc/grpc/blob/master/doc/statuscodes.md
    """
    #: The operation completed successfully
    OK = 0
    #: The operation was cancelled (typically by the caller)
    CANCELLED = 1
    #: Generic status to describe error when it can't be described using
    #: other statuses
    UNKNOWN = 2
    #: Client specified an invalid argument
    INVALID_ARGUMENT = 3
    #: Deadline expired before operation could complete
    DEADLINE_EXCEEDED = 4
    #: Some requested entity was not found
    NOT_FOUND = 5
    #: Some entity that we attempted to create already exists
    ALREADY_EXISTS = 6
    #: The caller does not have permission to execute the specified operation
    PERMISSION_DENIED = 7
    #: Some resource has been exhausted, perhaps a per-user quota, or perhaps
    #: the entire file system is out of space
    RESOURCE_EXHAUSTED = 8
    #: Operation was rejected because the system is not in a state required
    #: for the operation's execution
    FAILED_PRECONDITION = 9
    #: The operation was aborted
    ABORTED = 10
    #: Operation was attempted past the valid range
    OUT_OF_RANGE = 11
    #: Operation is not implemented or not supported/enabled in this service
    UNIMPLEMENTED = 12
    #: Internal errors
    INTERNAL = 13
    #: The service is currently unavailable
    UNAVAILABLE = 14
    #: Unrecoverable data loss or corruption
    DATA_LOSS = 15
    #: The request does not have valid authentication credentials for the
    #: operation
    UNAUTHENTICATED = 16 class Cardinality(_Cardinality, enum.Enum):
    UNARY_UNARY = _Cardinality(False, False)
    UNARY_STREAM = _Cardinality(False, True)
    STREAM_UNARY = _Cardinality(True, False)
    STREAM_STREAM = _Cardinality(True, True) class Configuration:
    _keepalive_time: Optional[float] = field(
        default=cast(None, _DEFAULT),
        metadata={
            'validate': _optional(_chain(_of_type(int, float), _positive)),
            'server-default': 7200.0,
            'client-default': None,
            'test-default': None,
        },
    )
    _keepalive_timeout: float = field(
        default=20.0,
        metadata={
            'validate': _chain(_of_type(int, float), _positive),
        },
    )
    _keepalive_permit_without_calls: bool = field(
        default=False,
        metadata={
            'validate': _optional(_of_type(bool)),
        },
    )
    _http2_max_pings_without_data: int = field(
        default=2,
        metadata={
            'validate': _optional(_chain(_of_type(int), _non_negative)),
        },
    )
    _http2_min_sent_ping_interval_without_data: float = field(
        default=300,
        metadata={
            'validate': _optional(_chain(_of_type(int, float), _positive)),
        },
    )
    #: Sets inbound window size for a connection. HTTP/2 spec allows this value
    #: to be from 64 KiB to 2 GiB, 4 MiB is used by default
    http2_connection_window_size: int = field(
        default=_4MiB,
        metadata={
            'validate': _chain(_of_type(int), _range(_WMIN, _WMAX)),
        },
    )
    #: Sets inbound window size for a stream. HTTP/2 spec allows this value
    #: to be from 64 KiB to 2 GiB, 4 MiB is used by default
    http2_stream_window_size: int = field(
        default=_4MiB,
        metadata={
            'validate': _chain(_of_type(int), _range(_WMIN, _WMAX)),
        },
    )

    def __post_init__(self) -> None:
        _validate(self)

    def __for_server__(self) -> 'Configuration':
        return _with_defaults(self, 'server-default')

    def __for_client__(self) -> 'Configuration':
        return _with_defaults(self, 'client-default')

    def __for_test__(self) -> 'Configuration':
        return _with_defaults(self, 'test-default') async def send_message(
    stream: 'Stream',
    codec: 'CodecBase',
    message: _SendType,
    message_type: Type[_SendType],
    *,
    end: bool = False,
) -> None:
    reply_bin = codec.encode(message, message_type)
    reply_data = (struct.pack('?', False)
                  + struct.pack('>I', len(reply_bin))
                  + reply_bin)
    await stream.send_data(reply_data, end_stream=end) async def recv_message(
    stream: 'Stream',
    codec: 'CodecBase',
    message_type: Type[_RecvType],
) -> Optional[_RecvType]:
    meta = await stream.recv_data(5)
    if not meta:
        return None

    compressed_flag = struct.unpack('?', meta[:1])[0]
    if compressed_flag:
        raise NotImplementedError('Compression not implemented')

    message_len = struct.unpack('>I', meta[1:])[0]
    message_bin = await stream.recv_data(message_len)
    assert len(message_bin) == message_len, \
        '{} != {}'.format(len(message_bin), message_len)
    message = codec.decode(message_bin, message_type)
    return cast(_RecvType, message) class StreamIterator(AsyncIterator[_RecvType], metaclass=abc.ABCMeta):

    @abc.abstractmethod
    async def recv_message(self) -> Optional[_RecvType]:
        pass

    def __aiter__(self) -> AsyncIterator[_RecvType]:
        return self

    async def __anext__(self) -> _RecvType:
        message = await self.recv_message()
        if message is None:
            raise StopAsyncIteration()
        else:
            return message async def recv_message(
    stream: 'Stream',
    codec: 'CodecBase',
    message_type: Type[_RecvType],
) -> Optional[_RecvType]:
async def send_message(
    stream: 'Stream',
    codec: 'CodecBase',
    message: _SendType,
    message_type: Type[_SendType],
    *,
    end: bool = False,
) -> None:
    async def recv_message(self) -> Optional[_RecvType]:
    def __aiter__(self) -> AsyncIterator[_RecvType]:
    async def __anext__(self) -> _RecvType:
class StreamIterator(AsyncIterator[_RecvType], metaclass=abc.ABCMeta): class _DispatchChannelEvents(_DispatchCommonEvents):

    @_dispatches(SendRequest)
    async def send_request(
        self,
        metadata: _Metadata,
        *,
        method_name: str,
        deadline: Optional[Deadline],
        content_type: str,
    ) -> Tuple[_Metadata]:
        return await self.__dispatch__(SendRequest(  # type: ignore
            metadata=metadata,
            method_name=method_name,
            deadline=deadline,
            content_type=content_type,
        ))

    @_dispatches(RecvInitialMetadata)
    async def recv_initial_metadata(
        self,
        metadata: _Metadata,
    ) -> Tuple[_Metadata]:
        return await self.__dispatch__(RecvInitialMetadata(  # type: ignore
            metadata=metadata,
        ))

    @_dispatches(RecvTrailingMetadata)
    async def recv_trailing_metadata(
        self,
        metadata: _Metadata,
        *,
        status: Status,
        status_message: Optional[str],
        status_details: Any,
    ) -> Tuple[_Metadata]:
        return await self.__dispatch__(RecvTrailingMetadata(  # type: ignore
            metadata=metadata,
            status=status,
            status_message=status_message,
            status_details=status_details,
        )) class H2Protocol(Protocol):
    connection: Connection
    processor: EventsProcessor

    def __init__(
        self,
        handler: AbstractHandler,
        config: Configuration,
        h2_config: H2Configuration,
    ) -> None:
        self.handler = handler
        self.config = config
        self.h2_config = h2_config

    def connection_made(self, transport: BaseTransport) -> None:
        sock = transport.get_extra_info('socket')
        if sock is not None:
            _set_nodelay(sock)

        h2_conn = H2Connection(config=self.h2_config)
        h2_conn.initiate_connection()

        initial = h2_conn.local_settings.initial_window_size
        conn_delta = self.config.http2_connection_window_size - initial
        stream_delta = self.config.http2_stream_window_size - initial
        if conn_delta:
            h2_conn.increment_flow_control_window(conn_delta)
        if stream_delta:
            h2_conn.update_settings({
                SettingCodes.INITIAL_WINDOW_SIZE:
                    self.config.http2_stream_window_size,
            })

        self.connection = Connection(
            h2_conn,
            cast(Transport, transport),
            config=self.config,
        )
        self.connection.flush()
        self.connection.initialize()

        self.processor = EventsProcessor(self.handler, self.connection)

    def data_received(self, data: bytes) -> None:
        try:
            events = self.connection.feed(data)
        except ProtocolError:
            log.debug('Protocol error', exc_info=True)
            self.processor.close('Protocol error')
        else:
            self.connection.flush()
            for event in events:
                self.processor.process(event)
            self.connection.flush()

    def pause_writing(self) -> None:
        self.connection.pause_writing()

    def resume_writing(self) -> None:
        self.connection.resume_writing()

    def connection_lost(self, exc: Optional[BaseException]) -> None:
        self.processor.close(reason='Connection lost') class AbstractHandler(ABC):

    @abstractmethod
    def accept(
        self,
        stream: Stream,
        headers: _Headers,
        release_stream: Callable[[], None],
    ) -> None:
        pass

    @abstractmethod
    def cancel(self, stream: Stream) -> None:
        pass

    @abstractmethod
    def close(self) -> None:
        pass class Stream:
    """
    API for working with streams, used by clients and request handlers
    """
    id: Optional[int] = None

    # stats
    created: Optional[float] = None
    data_sent = 0
    data_received = 0

    def __init__(
        self,
        connection: Connection,
        h2_connection: H2Connection,
        transport: Transport,
        *,
        stream_id: Optional[int] = None,
        wrapper: Optional[Wrapper] = None
    ) -> None:
        self.connection = connection
        self._h2_connection = h2_connection
        self._transport = transport
        self.wrapper = wrapper

        if stream_id is not None:
            self.init_stream(stream_id, self.connection)

        self.window_updated = Event()
        self.headers: Optional['_Headers'] = None
        self.headers_received = Event()
        self.trailers: Optional['_Headers'] = None
        self.trailers_received = Event()

    def init_stream(self, stream_id: int, connection: Connection) -> None:
        self.id = stream_id
        self.buffer = Buffer(partial(connection.ack, self.id))

        self.connection.streams_started += 1
        self.created = self.connection.last_stream_created = time.monotonic()

    async def recv_headers(self) -> _Headers:
        if self.headers is None:
            await self.headers_received.wait()
        assert self.headers is not None
        return self.headers

    async def recv_data(self, size: int) -> bytes:
        return await self.buffer.read(size)

    async def recv_trailers(self) -> _Headers:
        if self.trailers is None:
            await self.trailers_received.wait()
        assert self.trailers is not None
        return self.trailers

    async def send_request(
        self,
        headers: _Headers,
        end_stream: bool = False,
        *,
        _processor: 'EventsProcessor',
    ) -> Callable[[], None]:
        assert self.id is None, self.id
        while True:
            # this is the first thing we should check before even trying to
            # create new stream, because this wait() can be cancelled by timeout
            # and we wouldn't need to create new stream at all
            await self.connection.write_ready.wait()

            # `get_next_available_stream_id()` should be as close to
            # `connection.send_headers()` as possible, without any async
            # interruptions in between, see the docs on the
            # `get_next_available_stream_id()` method
            stream_id = self._h2_connection.get_next_available_stream_id()
            try:
                self._h2_connection.send_headers(stream_id, headers,
                                                 end_stream=end_stream)
            except TooManyStreamsError:
                # we're going to wait until any of currently opened streams will
                # be closed, and we will be able to open a new one
                # TODO: maybe implement FIFO for waiters, but this limit
                #       shouldn't be reached in a normal case, so why bother
                # TODO: maybe we should raise an exception here instead of
                #       waiting, if timeout wasn't set for the current request
                self.connection.stream_close_waiter.clear()
                await self.connection.stream_close_waiter.wait()
                # while we were trying to create a new stream, write buffer
                # can became full, so we need to repeat checks from checking
                # if we can write() data
                continue
            else:
                self.init_stream(stream_id, self.connection)
                release_stream = _processor.register(self)
                self._transport.write(self._h2_connection.data_to_send())
                self.connection.headers_send_process()
                return release_stream

    async def send_headers(
        self,
        headers: _Headers,
        end_stream: bool = False,
    ) -> None:
        assert self.id is not None
        await self.connection.write_ready.wait()

        # Workaround for the H2Connection.send_headers method, which will try
        # to create a new stream if it was removed earlier from the
        # H2Connection.streams, and therefore will raise StreamIDTooLowError
        if self.id not in self._h2_connection.streams:
            raise StreamClosedError(self.id)

        self._h2_connection.send_headers(self.id, headers,
                                         end_stream=end_stream)
        self._transport.write(self._h2_connection.data_to_send())
        self.connection.headers_send_process()

    async def send_data(self, data: bytes, end_stream: bool = False) -> None:
        f = BytesIO(data)
        f_pos, f_last = 0, len(data)

        while True:
            await self.connection.write_ready.wait()

            window = self._h2_connection.local_flow_control_window(self.id)
            # window can become negative
            if not window > 0:
                self.window_updated.clear()
                await self.window_updated.wait()
                # during "await" above other streams were able to send data and
                # decrease current window size, so try from the beginning
                continue

            max_frame_size = self._h2_connection.max_outbound_frame_size
            f_chunk = f.read(min(window, max_frame_size, f_last - f_pos))
            f_chunk_len = len(f_chunk)
            f_pos = f.tell()

            if f_pos == f_last:
                self._h2_connection.send_data(self.id, f_chunk,
                                              end_stream=end_stream)
                self._transport.write(self._h2_connection.data_to_send())
                self.data_sent += f_chunk_len
                self.connection.data_sent += f_chunk_len
                self.connection.data_send_process()
                break
            else:
                self._h2_connection.send_data(self.id, f_chunk)
                self._transport.write(self._h2_connection.data_to_send())
                self.data_sent += f_chunk_len
                self.connection.data_sent += f_chunk_len
                self.connection.data_send_process()

    async def end(self) -> None:
        await self.connection.write_ready.wait()
        self._h2_connection.end_stream(self.id)
        self._transport.write(self._h2_connection.data_to_send())

    async def reset(self, error_code: ErrorCodes = ErrorCodes.NO_ERROR) -> None:
        await self.connection.write_ready.wait()
        self._h2_connection.reset_stream(self.id, error_code=error_code)
        self._transport.write(self._h2_connection.data_to_send())

    def reset_nowait(
        self,
        error_code: ErrorCodes = ErrorCodes.NO_ERROR,
    ) -> None:
        self._h2_connection.reset_stream(self.id, error_code=error_code)
        if self.connection.write_ready.is_set():
            self._transport.write(self._h2_connection.data_to_send())

    def __ended__(self) -> None:
        self.buffer.eof()

    def __terminated__(self, reason: str) -> None:
        if self.wrapper is not None:
            self.wrapper.cancel(StreamTerminatedError(reason))

    @property
    def closable(self) -> bool:
        if self._h2_connection.state_machine.state is ConnectionState.CLOSED:
            return False
        stream = self._h2_connection.streams.get(self.id)
        if stream is None:
            return False
        return not stream.closed class Peer:
    """
    Represents an information about a connection's peer
    """
    def __init__(self, transport: Transport) -> None:
        self._transport = transport

    def addr(self) -> Optional[Tuple[str, int]]:
        """Returns the remote address to which we are connected"""
        return self._transport.get_extra_info('peername')  # type: ignore

    def cert(self) -> Optional[Dict[str, Any]]:
        """Returns the peer certificate

        Result of the :py:meth:`python:ssl.SSLSocket.getpeercert`
        """
        ssl_object = self._transport.get_extra_info('ssl_object')
        if ssl_object is not None:
            return ssl_object.getpeercert()  # type: ignore
        else:
            return None class Deadline:
    """Represents request's deadline - fixed point in time
    """
    def __init__(self, *, _timestamp: float) -> None:
        self._timestamp = _timestamp

    def __lt__(self, other: object) -> bool:
        if not isinstance(other, Deadline):
            raise TypeError('comparison is not supported between '
                            'instances of \'{}\' and \'{}\''
                            .format(type(self).__name__, type(other).__name__))
        return self._timestamp < other._timestamp

    def __eq__(self, other: object) -> bool:
        if not isinstance(other, Deadline):
            return False
        return self._timestamp == other._timestamp

    @classmethod
    def from_headers(cls, headers: _Headers) -> Optional['Deadline']:
        timeout = min(map(decode_timeout,
                          (v for k, v in headers if k == 'grpc-timeout')),
                      default=None)
        if timeout is not None:
            return cls.from_timeout(timeout)
        else:
            return None

    @classmethod
    def from_timeout(cls, timeout: float) -> 'Deadline':
        return cls(_timestamp=time.monotonic() + timeout)

    def time_remaining(self) -> float:
        """Calculates remaining time for the current request completion

        This function returns time in seconds as a floating point number,
        greater or equal to zero.
        """
        return max(0, self._timestamp - time.monotonic()) USER_AGENT = (
    'grpc-python-grpclib/{lib_ver} ({sys}; {py}/{py_ver})'
    .format(
        lib_ver=__version__,
        sys=platform.system(),
        py=platform.python_implementation(),
        py_ver=platform.python_version(),
    )
    .lower()
) def decode_grpc_message(value: str) -> str:
    return unquote(value, encoding='utf-8', errors='replace') def encode_timeout(timeout: float) -> str:
    if timeout > 10:
        return '{}S'.format(int(timeout))
    elif timeout > 0.01:
        return '{}m'.format(int(timeout * 10 ** 3))
    elif timeout > 0.00001:
        return '{}u'.format(int(timeout * 10 ** 6))
    else:
        return '{}n'.format(int(timeout * 10 ** 9)) USER_AGENT = (
    'grpc-python-grpclib/{lib_ver} ({sys}; {py}/{py_ver})'
    .format(
        lib_ver=__version__,
        sys=platform.system(),
        py=platform.python_implementation(),
        py_ver=platform.python_version(),
    )
    .lower()
)
_UNITS = {
    'H': 60 * 60,
    'M': 60,
    'S': 1,
    'm': 10 ** -3,
    'u': 10 ** -6,
    'n': 10 ** -9,
}
_TIMEOUT_RE = re.compile(r'^(\d+)([{}])$'.format(''.join(_UNITS)))
_STATUS_DETAILS_KEY = 'grpc-status-details-bin'
_UNQUOTED = ''.join([chr(i) for i in range(0x20, 0x24 + 1)]
                    + [chr(i) for i in range(0x26, 0x7E + 1)])
_KEY_RE = re.compile(r'^[0-9a-z_.\-]+$')
_VALUE_RE = re.compile(r'^[ !-~]+$')  # 0x20-0x7E - space and printable ASCII
_SPECIAL = {
    'te',
    'content-type',
    'user-agent',
}
def decode_timeout(value: str) -> float:
def encode_timeout(timeout: float) -> str:
    def __init__(self, *, _timestamp: float) -> None:
    def __lt__(self, other: object) -> bool:
    def __eq__(self, other: object) -> bool:
    def from_headers(cls, headers: _Headers) -> Optional['Deadline']:
    def from_timeout(cls, timeout: float) -> 'Deadline':
    def time_remaining(self) -> float:
def encode_grpc_message(message: str) -> str:
def decode_grpc_message(value: str) -> str:
def decode_bin_value(value: bytes) -> bytes:
def decode_metadata(headers: _Headers) -> _Metadata:
def encode_bin_value(value: bytes) -> bytes:
def encode_metadata(metadata: _MetadataLike) -> _Headers:
class Deadline: _STATUS_DETAILS_KEY = 'grpc-status-details-bin' def decode_bin_value(value: bytes) -> bytes:
    return b64decode(value + (b'=' * (len(value) % 4))) class GRPCError(Exception):
    """Expected error, may be raised during RPC call

    There can be multiple origins of this error. It can be generated
    on the server-side and on the client-side. If this error originates from
    the server, on the wire this error is represented as ``grpc-status`` and
    ``grpc-message`` trailers. Possible values of the ``grpc-status`` trailer
    are described in the gRPC protocol definition. In ``grpclib`` these values
    are represented as :py:class:`~grpclib.const.Status` enum.

    Here are possible origins of this error:

      - you may raise this error to cancel current call on the server-side or
        return non-OK :py:class:`~grpclib.const.Status` using
        :py:meth:`~grpclib.server.Stream.send_trailing_metadata` method
        `(e.g. resource not found)`
      - server may return non-OK ``grpc-status`` in different failure
        conditions `(e.g. invalid request)`
      - client raises this error for non-OK ``grpc-status`` from the server
      - client may raise this error in different failure conditions
        `(e.g. server returned unsupported` ``:content-type`` `header)`

    """
    def __init__(
        self,
        status: Status,
        message: Optional[str] = None,
        details: Any = None,
    ) -> None:
        super().__init__(status, message, details)
        #: :py:class:`~grpclib.const.Status` of the error
        self.status = status
        #: Error message
        self.message = message
        #: Error details
        self.details = details class ProtocolError(Exception):
    """Unexpected error, raised by ``grpclib`` when your code violates
    gRPC protocol

    This error means that you probably should fix your code.
    """ class StreamTerminatedError(Exception):
    """Unexpected error, raised when we receive ``RST_STREAM`` frame from
    the other side

    This error means that the other side decided to forcefully cancel current
    call, probably because of a protocol error.
    """ GRPC_CONTENT_TYPE = 'application/grpc' class CodecBase(abc.ABC):

    @property
    @abc.abstractmethod
    def __content_subtype__(self) -> str:
        pass

    @abc.abstractmethod
    def encode(self, message: Any, message_type: Any) -> bytes:
        pass

    @abc.abstractmethod
    def decode(self, data: bytes, message_type: Any) -> Any:
        pass class StatusDetailsCodecBase(abc.ABC):

    @abc.abstractmethod
    def encode(
        self, status: Status, message: Optional[str], details: Any,
    ) -> bytes:
        pass

    @abc.abstractmethod
    def decode(
        self, status: Status, message: Optional[str], data: bytes,
    ) -> Any:
        pass class ProtoCodec(CodecBase):
    __content_subtype__ = 'proto'

    def encode(
        self,
        message: 'IProtoMessage',
        message_type: Type['IProtoMessage'],
    ) -> bytes:
        if not isinstance(message, message_type):
            raise TypeError('Message must be of type {!r}, not {!r}'
                            .format(message_type, type(message)))
        return message.SerializeToString()

    def decode(
        self,
        data: bytes,
        message_type: Type['IProtoMessage'],
    ) -> 'IProtoMessage':
        return message_type.FromString(data) class ProtoStatusDetailsCodec(StatusDetailsCodecBase):

    def encode(
        self,
        status: Status,
        message: Optional[str],
        details: Sequence['Message'],
    ) -> bytes:
        status_pb2 = _status_pb2()

        status_proto = status_pb2.Status(code=status.value, message=message)
        if details is not None:
            for detail in details:
                detail_container = status_proto.details.add()
                detail_container.Pack(detail)
        return status_proto.SerializeToString()  # type: ignore

    def decode(
        self, status: Status, message: Optional[str], data: bytes,
    ) -> Sequence[Any]:
        status_pb2 = _status_pb2()
        sym_db = _sym_db()

        status_proto = status_pb2.Status.FromString(data)
        details = []
        for detail_container in status_proto.details:
            try:
                msg_type = sym_db.GetSymbol(detail_container.TypeName())
            except KeyError:
                details.append(_Unknown(detail_container.TypeName()))
                continue
            detail = msg_type()
            detail_container.Unpack(detail)
            details.append(detail)
        return details @_cached
def _googleapis_available() -> bool:
    try:
        import google.rpc.status_pb2  # noqa
    except ImportError:
        return False
    else:
        return True
import sys
import enum
import http
import time
import asyncio
import warnings
    import ssl as _ssl
            import certifi
from types import TracebackType
from typing import Generic, Optional, Union, Type, List, Sequence, Any, cast
from typing import Dict, Tuple, TYPE_CHECKING
from h2.config import H2Configuration
from multidict import MultiDict
from .utils import Wrapper, DeadlineWrapper
from .const import Status, Cardinality
from .config import Configuration
from .stream import send_message, recv_message, StreamIterator
from .stream import _RecvType, _SendType
from .events import _DispatchChannelEvents
from .protocol import H2Protocol, AbstractHandler, Stream as _Stream, Peer
from .metadata import Deadline, USER_AGENT, decode_grpc_message, encode_timeout
from .metadata import encode_metadata, decode_metadata, _MetadataLike, _Metadata
from .metadata import _STATUS_DETAILS_KEY, decode_bin_value
from .exceptions import GRPCError, ProtocolError, StreamTerminatedError
from .encoding.base import GRPC_CONTENT_TYPE, CodecBase, StatusDetailsCodecBase
from .encoding.proto import ProtoCodec, ProtoStatusDetailsCodec
from .encoding.proto import _googleapis_available
from ._registry import channels as _channels
    from ._typing import IReleaseStream  # noqa
                self._channel._calls_succeeded += 1
            else:
                self._channel._calls_failed += 1


class _ChannelState(enum.IntEnum):
    IDLE = 1
    CONNECTING = 2
    READY = 3
    TRANSIENT_FAILURE = 4


class Channel:
    """
    Represents a connection to the server, which can be used with generated
    stub classes to perform gRPC calls.

    .. code-block:: python3

        channel = Channel()
        client = cafe_grpc.CoffeeMachineStub(channel)

        ...

        request = cafe_pb2.LatteOrder(
            size=cafe_pb2.SMALL,
            temperature=70,
            sugar=3,
        )
        reply: empty_pb2.Empty = await client.MakeLatte(request)

        ...

        channel.close()
    """
    _protocol = None

    # stats
    _calls_started = 0
    _calls_succeeded = 0
    _calls_failed = 0
    _last_call_started: Optional[float] = None

    def __init__(
        self,
        host: Optional[str] = None,
        port: Optional[int] = None,
        *,
        loop: Optional[asyncio.AbstractEventLoop] = None,
        path: Optional[str] = None,
        codec: Optional[CodecBase] = None,
        status_details_codec: Optional[StatusDetailsCodecBase] = None,
        ssl: Union[None, bool, '_ssl.SSLContext'] = None,
        config: Optional[Configuration] = None,
    ):
        """Initialize connection to the server

        :param host: server host name.

        :param port: server port number.

        :param loop: (deprecated) asyncio-compatible event loop

        :param path: server socket path. If specified, host and port should be
            omitted (must be None).

        :param codec: instance of a codec to encode and decode messages,
            if omitted ``ProtoCodec`` is used by default

        :param status_details_codec: instance of a status details codec to
            decode error details in a trailing metadata, if omitted
            ``ProtoStatusDetailsCodec`` is used by default

        :param ssl: ``True`` or :py:class:`~python:ssl.SSLContext` object; if
            ``True``, default SSL context is used.
        """
        if path is not None and (host is not None or port is not None):
            raise ValueError("The 'path' parameter can not be used with the "
                             "'host' or 'port' parameters.")
        else:
            if host is None:
                host = '127.0.0.1'

            if port is None:
                port = 50051

        if ssl is True:
            ssl = self._get_default_ssl_context()

        if codec is None:
            codec = ProtoCodec()
            if status_details_codec is None and _googleapis_available():
                status_details_codec = ProtoStatusDetailsCodec()

        if loop:
            warnings.warn("The loop argument is deprecated and scheduled "
                          "for removal in grpclib 0.5",
                          DeprecationWarning, stacklevel=2)

        self._host = host
        self._port = port
        self._loop = loop or asyncio.get_event_loop()
        self._path = path
        self._codec = codec
        self._status_details_codec = status_details_codec
        self._ssl = ssl or None
        self._authority = '{}:{}'.format(self._host, self._port)
        self._scheme = 'https' if self._ssl else 'http'
        self._authority = '{}:{}'.format(self._host, self._port)
        self._h2_config = H2Configuration(
            client_side=True,
            header_encoding='ascii',
            validate_inbound_headers=False,
            validate_outbound_headers=False,
            normalize_inbound_headers=False,
            normalize_outbound_headers=False,
        )
        self._connect_lock = asyncio.Lock()
        self._state = _ChannelState.IDLE


        config = Configuration() if config is None else config