1# -*- coding: utf-8 -*- 

2 

3# -- stdlib -- 

4from typing import Any, Callable, Optional, Sequence, Tuple, TypeVar, Union 

5import functools 

6import logging 

7 

8# -- third party -- 

9# -- own -- 

10from server.endpoint import Client 

11from utils.events import EventHub 

12from wire.msg import Message 

13 

14 

15# -- code -- 

16log = logging.getLogger("server.utils") 

17STOP = EventHub.STOP_PROPAGATION 

18 

19T = TypeVar('T', bound=Message) 

20Self = TypeVar('Self') 

21 

22CommandHandler = Callable[[Self, Client, T], Optional[EventHub.StopPropagation]] 

23WrappedCommandHandler = Callable[[Self, Tuple[Client, T]], Union[Tuple[Client, T], EventHub.StopPropagation]] 

24 

25 

26def command(*states: Sequence[str]) -> Callable[[CommandHandler], WrappedCommandHandler]: 

27 def decorate(f: CommandHandler) -> WrappedCommandHandler: 

28 @functools.wraps(f) 

29 def wrapper(self: Self, ev: Tuple[Client, T]) -> Union[Tuple[Client, T], EventHub.StopPropagation]: 

30 core = self.core # type: ignore 

31 u, msg = ev 

32 if '*' not in states and core.lobby.state_of(u) not in states: 

33 return ev 

34 else: 

35 ret = f(self, u, msg) 

36 if ret is STOP: 

37 return STOP 

38 return ev 

39 

40 return wrapper 

41 

42 return decorate