# Path: grpclib/events.py
# class _Event:
#     __slots__ = ('__interrupted__',)
#     __payload__: Collection[str] = ()
#     __readonly__: FrozenSet[str] = frozenset()
#     __interrupted__: bool
# 
#     def __init__(self, **kwargs):  # type: ignore
#         assert len(kwargs) == len(self.__slots__), self.__slots__
#         super().__setattr__('__interrupted__', False)
#         for key, value in kwargs.items():
#             super().__setattr__(key, value)
# 
#     def __setattr__(self, key: str, value: Any) -> None:
#         if key in self.__readonly__:
#             raise AttributeError('Read-only property: {!r}'.format(key))
#         else:
#             super().__setattr__(key, value)
# 
#     def interrupt(self) -> None:
#         super().__setattr__('__interrupted__', True)
# 
# class _EventMeta(type):
# 
#     def __new__(mcs, name, bases, params):  # type: ignore
#         annotations = params.get('__annotations__') or {}
#         payload = params.get('__payload__') or ()
#         params['__slots__'] = tuple(name for name in annotations)
#         params['__readonly__'] = frozenset(name for name in annotations
#                                            if name not in payload)
#         return super().__new__(mcs, name, bases, params)
# 
# async def _ident(*args, **_):  # type: ignore
#     return args
# 
# def listen(
#     target: 'IEventsTarget',
#     event_type: Type[_EventType],
#     callback: Callable[[_EventType], Coroutine[Any, Any, None]],
# ) -> None:
#     """Registers a listener function for the given target and event type
# 
#     .. code-block:: python3
# 
#         async def callback(event: SomeEvent):
#             print(event.data)
# 
#         listen(target, SomeEvent, callback)
#     """
#     target.__dispatch__.add_listener(event_type, callback)
# 
# Path: grpclib/events.py
# class _Dispatch:
#     __dispatch_methods__: Dict[Type[_Event], str] = {}
# 
#     def __init__(self) -> None:
#         self._listeners: Dict[Type[_Event], List[_Callback]] = defaultdict(list)
#         for name in self.__dispatch_methods__.values():
#             self.__dict__[name] = _ident
# 
#     def add_listener(
#         self,
#         event_type: Type[_Event],
#         callback: _Callback,
#     ) -> None:
#         self.__dict__.pop(self.__dispatch_methods__[event_type], None)
#         self._listeners[event_type].append(callback)
# 
#     async def __dispatch__(self, event: _Event) -> Any:
#         for callback in self._listeners[event.__class__]:
#             await callback(event)
#             if event.__interrupted__:
#                 break
#         return tuple(getattr(event, name) for name in event.__payload__)
# 
# class _DispatchMeta(type):
# 
#     def __new__(mcs, name, bases, params):  # type: ignore
#         dispatch_methods = dict(chain.from_iterable(
#             getattr(base, '__dispatch_methods__', {}).items()
#             for base in bases
#         ))
#         for key, value in params.items():
#             dispatches = getattr(value, '__dispatches__', None)
#             if dispatches is not None:
#                 assert (isinstance(dispatches, type)
#                         and issubclass(dispatches, _Event)), dispatches
#                 assert dispatches not in dispatch_methods, dispatches
#                 dispatch_methods[dispatches] = key
#         params['__dispatch_methods__'] = dispatch_methods
#         return super().__new__(mcs, name, bases, params)
# 
# def _dispatches(event_type: Type[_Event]) -> Callable[[Any], Any]:
#     def decorator(func: Any) -> Any:
#         func.__dispatches__ = event_type
#         return func
#     return decorator


from unittest.mock import Mock, patch
from grpclib.events import _Event, _EventMeta, _ident, listen
from grpclib.events import _Dispatch, _DispatchMeta, _dispatches
import pytest

@pytest.mark.asyncio
async def test_dispatch():
    class MyEvent(_Event, metaclass=_EventMeta):
        __payload__ = ('a', 'b')
        a: int
        b: str
        c: int
        d: str

    class MyDispatch(_Dispatch, metaclass=_DispatchMeta):
        @_dispatches(MyEvent)
        async def my_event(self, a, b, *, c, d):
            return await self.__dispatch__(MyEvent(a=a, b=b, c=c, d=d))

    class Target:
        __dispatch__ = MyDispatch()

    assert Target.__dispatch__.my_event is _ident
    assert await Target.__dispatch__.my_event(1, 3, c=6, d=9) == (1, 3)

    @mock_callback
    async def callback(event: MyEvent):
        assert event.a == 2
        assert event.b == 4
        assert event.c == 8
        assert event.d == 16

    listen(Target, MyEvent, callback)


# Path: grpclib/events.py
# class _Event:
#     __slots__ = ('__interrupted__',)
#     __payload__: Collection[str] = ()
#     __readonly__: FrozenSet[str] = frozenset()
#     __interrupted__: bool
# 
#     def __init__(self, **kwargs):  # type: ignore
#         assert len(kwargs) == len(self.__slots__), self.__slots__
#         super().__setattr__('__interrupted__', False)
#         for key, value in kwargs.items():
#             super().__setattr__(key, value)
# 
#     def __setattr__(self, key: str, value: Any) -> None:
#         if key in self.__readonly__:
#             raise AttributeError('Read-only property: {!r}'.format(key))
#         else:
#             super().__setattr__(key, value)
# 
#     def interrupt(self) -> None:
#         super().__setattr__('__interrupted__', True)
# 
# class _EventMeta(type):
# 
#     def __new__(mcs, name, bases, params):  # type: ignore
#         annotations = params.get('__annotations__') or {}
#         payload = params.get('__payload__') or ()
#         params['__slots__'] = tuple(name for name in annotations)
#         params['__readonly__'] = frozenset(name for name in annotations
#                                            if name not in payload)
#         return super().__new__(mcs, name, bases, params)
# 
# async def _ident(*args, **_):  # type: ignore
#     return args
# 
# def listen(
#     target: 'IEventsTarget',
#     event_type: Type[_EventType],
#     callback: Callable[[_EventType], Coroutine[Any, Any, None]],
# ) -> None:
#     """Registers a listener function for the given target and event type
# 
#     .. code-block:: python3
# 
#         async def callback(event: SomeEvent):
#             print(event.data)
# 
#         listen(target, SomeEvent, callback)
#     """
#     target.__dispatch__.add_listener(event_type, callback)
# 
# Path: grpclib/events.py
# class _Dispatch:
#     __dispatch_methods__: Dict[Type[_Event], str] = {}
# 
#     def __init__(self) -> None:
#         self._listeners: Dict[Type[_Event], List[_Callback]] = defaultdict(list)
#         for name in self.__dispatch_methods__.values():
#             self.__dict__[name] = _ident
# 
#     def add_listener(
#         self,
#         event_type: Type[_Event],
#         callback: _Callback,
#     ) -> None:
#         self.__dict__.pop(self.__dispatch_methods__[event_type], None)
#         self._listeners[event_type].append(callback)
# 
#     async def __dispatch__(self, event: _Event) -> Any:
#         for callback in self._listeners[event.__class__]:
#             await callback(event)
#             if event.__interrupted__:
#                 break
#         return tuple(getattr(event, name) for name in event.__payload__)
# 
# class _DispatchMeta(type):
# 
#     def __new__(mcs, name, bases, params):  # type: ignore
#         dispatch_methods = dict(chain.from_iterable(
#             getattr(base, '__dispatch_methods__', {}).items()
#             for base in bases
#         ))
#         for key, value in params.items():
#             dispatches = getattr(value, '__dispatches__', None)
#             if dispatches is not None:
#                 assert (isinstance(dispatches, type)
#                         and issubclass(dispatches, _Event)), dispatches
#                 assert dispatches not in dispatch_methods, dispatches
#                 dispatch_methods[dispatches] = key
#         params['__dispatch_methods__'] = dispatch_methods
#         return super().__new__(mcs, name, bases, params)
# 
# def _dispatches(event_type: Type[_Event]) -> Callable[[Any], Any]:
#     def decorator(func: Any) -> Any:
#         func.__dispatches__ = event_type
#         return func
#     return decorator


# Path: tests/test_events.py
from unittest.mock import Mock, patch
from grpclib.events import _Event, _EventMeta, _ident, listen
from grpclib.events import _Dispatch, _DispatchMeta, _dispatches
import pytest


@pytest.mark.asyncio
async def test_dispatch():
    class MyEvent(_Event, metaclass=_EventMeta):
        __payload__ = ('a', 'b')
        a: int
        b: str
        c: int
        d: str

    class MyDispatch(_Dispatch, metaclass=_DispatchMeta):
        @_dispatches(MyEvent)
        async def my_event(self, a, b, *, c, d):
            return await self.__dispatch__(MyEvent(a=a, b=b, c=c, d=d))

    class Target:
        __dispatch__ = MyDispatch()

    assert Target.__dispatch__.my_event is _ident
    assert await Target.__dispatch__.my_event(1, 3, c=6, d=9) == (1, 3)

    @mock_callback
    async def callback(event: MyEvent):
        assert event.a == 2
        assert event.b == 4
        assert event.c == 8
        assert event.d == 16

    listen(Target, MyEvent, callback)


    assert Target.__dispatch__.my_event is not _ident