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) 

20 

21 

22def command(*states: Sequence[str]) -> Callable[[Callable[[Any, Client, T], Optional[EventHub.StopPropagation]]], Callable[[Any, Tuple[Client, T]], Union[Tuple[Client, T], EventHub.StopPropagation]]]: 

23 def decorate(f: Any) -> Any: 

24 @functools.wraps(f) 

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

26 core = self.core 

27 u, msg = ev 

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

29 return ev 

30 else: 

31 ret = f(self, u, msg) 

32 if ret is STOP: 

33 return STOP 

34 return ev 

35 

36 return wrapper 

37 

38 return decorate