Module graia.broadcast.interrupt.waiter

Classes

class Waiter

Expand source code

class Waiter(metaclass=ABCMeta):
    listening_events: List[Type[BaseEvent]]
    using_dispatchers: List[T_Dispatcher]
    using_decorators: List[Decorator]
    priority: int
    enable_internal_access: bool
    block_propagation: bool

    @classmethod
    def create(
        cls,
        listening_events: List[Type[BaseEvent]],
        using_dispatchers: List[T_Dispatcher] = None,
        using_decorators: List[Decorator] = None,
        priority: int = 15,  # 默认情况下都是需要高于默认 16 的监听吧...
        enable_internal_access: bool = False,
        block_propagation: bool = False,
    ) -> Type["Waiter"]:
        async def detected_event(self) -> Any:
            pass

        return type(
            "AbstractWaiter",
            (cls,),
            {
                "listening_events": listening_events,
                "using_dispatchers": using_dispatchers,
                "using_decorators": using_decorators,
                "priority": priority,
                "enable_internal_access": enable_internal_access,
                "block_propagation": block_propagation,
                "detected_event": abstractmethod(detected_event),
            },
        )

    async def detected_event(self) -> Any:
        pass

    @classmethod
    def create_using_function(
        cls,
        listening_events: List[Type[BaseEvent]],
        using_dispatchers: List[T_Dispatcher] = None,
        using_decorators: List[Decorator] = None,
        priority: int = 15,  # 默认情况下都是需要高于默认 16 的监听吧...
        enable_internal_access: bool = False,
        block_propagation: bool = False,
    ):
        def wrapper(func):
            return type(
                "SingleWaiter",
                (
                    cls.create(
                        listening_events,
                        using_dispatchers,
                        using_decorators,
                        priority,
                        enable_internal_access,
                        block_propagation,
                    ),
                ),
                {"detected_event": staticmethod(func)},
            )()

        return wrapper

Class variables

var block_propagation : bool

var enable_internal_access : bool

var listening_events : List[Type[BaseEvent]]

var priority : int

var using_decorators : List[Decorator]

var using_dispatchers : List[Union[Type[BaseDispatcher], BaseDispatcher, Callable[[DispatcherInterface], Any]]]

Static methods

def create(listening_events: List[Type[BaseEvent]], using_dispatchers: List[Union[Type[ForwardRef('BaseDispatcher')], ForwardRef('BaseDispatcher'), Callable[[ForwardRef('DispatcherInterface')], Any]]] = None, using_decorators: List[Decorator] = None, priority: int = 15, enable_internal_access: bool = False, block_propagation: bool = False) ‑> Type[Waiter]

Expand source code

@classmethod
def create(
    cls,
    listening_events: List[Type[BaseEvent]],
    using_dispatchers: List[T_Dispatcher] = None,
    using_decorators: List[Decorator] = None,
    priority: int = 15,  # 默认情况下都是需要高于默认 16 的监听吧...
    enable_internal_access: bool = False,
    block_propagation: bool = False,
) -> Type["Waiter"]:
    async def detected_event(self) -> Any:
        pass

    return type(
        "AbstractWaiter",
        (cls,),
        {
            "listening_events": listening_events,
            "using_dispatchers": using_dispatchers,
            "using_decorators": using_decorators,
            "priority": priority,
            "enable_internal_access": enable_internal_access,
            "block_propagation": block_propagation,
            "detected_event": abstractmethod(detected_event),
        },
    )

def create_using_function(listening_events: List[Type[BaseEvent]], using_dispatchers: List[Union[Type[ForwardRef('BaseDispatcher')], ForwardRef('BaseDispatcher'), Callable[[ForwardRef('DispatcherInterface')], Any]]] = None, using_decorators: List[Decorator] = None, priority: int = 15, enable_internal_access: bool = False, block_propagation: bool = False)

Expand source code

@classmethod
def create_using_function(
    cls,
    listening_events: List[Type[BaseEvent]],
    using_dispatchers: List[T_Dispatcher] = None,
    using_decorators: List[Decorator] = None,
    priority: int = 15,  # 默认情况下都是需要高于默认 16 的监听吧...
    enable_internal_access: bool = False,
    block_propagation: bool = False,
):
    def wrapper(func):
        return type(
            "SingleWaiter",
            (
                cls.create(
                    listening_events,
                    using_dispatchers,
                    using_decorators,
                    priority,
                    enable_internal_access,
                    block_propagation,
                ),
            ),
            {"detected_event": staticmethod(func)},
        )()

    return wrapper

Methods

async def detected_event(self) ‑> Any

Expand source code

async def detected_event(self) -> Any:
    pass