# Path: grpclib/utils.py
# 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
# 
# Path: grpclib/config.py
# 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')
# 
# Path: grpclib/exceptions.py
# 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.
#     """


import asyncio
import struct
import time
import socket
import logging
from io import BytesIO
from abc import ABC, abstractmethod
from typing import Optional, List, Tuple, Dict, NamedTuple, Callable, Any
from typing import cast, TYPE_CHECKING
from asyncio import Transport, Protocol, Event, BaseTransport, TimerHandle
from asyncio import Queue
from functools import partial
from collections import deque
from h2.errors import ErrorCodes
from h2.config import H2Configuration
from h2.events import Event as H2Event
from h2.events import RequestReceived, DataReceived, StreamEnded, WindowUpdated
from h2.events import ConnectionTerminated, RemoteSettingsChanged, StreamReset
from h2.events import SettingsAcknowledged, ResponseReceived, TrailersReceived
from h2.events import PriorityUpdated, PingReceived, PingAckReceived
from h2.settings import SettingCodes
from h2.connection import H2Connection, ConnectionState
from h2.exceptions import ProtocolError, TooManyStreamsError, StreamClosedError
from .utils import Wrapper
from .config import Configuration
from .exceptions import StreamTerminatedError
    from typing import Deque
            self.handler.cancel(stream)

        self.connection.streams_failed += 1

    def process_priority_updated(self, event: PriorityUpdated) -> None:
        pass

    def process_connection_terminated(
        self,
        event: ConnectionTerminated,
    ) -> None:
        self.close(reason=(
            'Received GOAWAY frame, closing connection; error_code: {}'
            .format(event.error_code)
        ))

    def process_ping_received(self, event: PingReceived) -> None:
        pass

    def process_ping_ack_received(self, event: PingAckReceived) -> None:
        self.connection.ping_ack_process()


class H2Protocol(Protocol):
    connection: Connection
    processor: EventsProcessor

    def __init__(
        self,
        handler: AbstractHandler,

# Path: grpclib/utils.py
# 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
# 
# Path: grpclib/config.py
# 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')
# 
# Path: grpclib/exceptions.py
# 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.
#     """


# Path: grpclib/protocol.py
import asyncio
import struct
import time
import socket
import logging
from io import BytesIO
from abc import ABC, abstractmethod
from typing import Optional, List, Tuple, Dict, NamedTuple, Callable, Any
from typing import cast, TYPE_CHECKING
from asyncio import Transport, Protocol, Event, BaseTransport, TimerHandle
from asyncio import Queue
from functools import partial
from collections import deque
from h2.errors import ErrorCodes
from h2.config import H2Configuration
from h2.events import Event as H2Event
from h2.events import RequestReceived, DataReceived, StreamEnded, WindowUpdated
from h2.events import ConnectionTerminated, RemoteSettingsChanged, StreamReset
from h2.events import SettingsAcknowledged, ResponseReceived, TrailersReceived
from h2.events import PriorityUpdated, PingReceived, PingAckReceived
from h2.settings import SettingCodes
from h2.connection import H2Connection, ConnectionState
from h2.exceptions import ProtocolError, TooManyStreamsError, StreamClosedError
from .utils import Wrapper
from .config import Configuration
from .exceptions import StreamTerminatedError
    from typing import Deque

            self.handler.cancel(stream)

        self.connection.streams_failed += 1

    def process_priority_updated(self, event: PriorityUpdated) -> None:
        pass

    def process_connection_terminated(
        self,
        event: ConnectionTerminated,
    ) -> None:
        self.close(reason=(
            'Received GOAWAY frame, closing connection; error_code: {}'
            .format(event.error_code)
        ))

    def process_ping_received(self, event: PingReceived) -> None:
        pass

    def process_ping_ack_received(self, event: PingAckReceived) -> None:
        self.connection.ping_ack_process()


class H2Protocol(Protocol):
    connection: Connection
    processor: EventsProcessor

    def __init__(
        self,
        handler: AbstractHandler,

        config: Configuration,