# Coverage for server/parts/lobby.py : 81%

## 80 statements 67 run 13 missing 4 excluded 4 partial

Hot-keys on this page

r m x p toggle line displays

j k next/prev highlighted chunk

0 (zero) top of page

1 (one) first highlighted chunk

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

2 from __future__ import annotations

3

4# -- stdlib --

5 from typing import Dict, Optional, Sequence, TYPE_CHECKING, Tuple

6 import logging

7

8# -- third party --

9# -- own --

10 from endpoint import Endpoint

11 from server.base import Game

12 from server.endpoint import Client

13 from utils.events import FSM

14 from utils.misc import BatchList, throttle

15 import wire

16

17# -- typing --

18 if TYPE_CHECKING:

19 from server.core import Core # noqa: F401

20

21

22# -- code --

23 log = logging.getLogger('Lobby')

24

25

26 class Lobby(object):

27 def __init__(self, core: Core):

28 self.core = core

29

30 core.events.client_connected += self.handle_client_connected

31 core.events.user_state_transition += self.handle_user_state_transition

32 core.events.game_ended += self.handle_game_ended

33

34 self.users: Dict[int, Client] = {} # all users

35 self.dropped_users: Dict[int, Client] = {} # passively dropped users

36

37 def __repr__(self) -> str:

38 return self.__class__.__name__

39

40 def handle_user_state_transition(self, ev: Tuple[Client, str, str]) -> Tuple[Client, str, str]:

41 c, f, t = ev

42

43 if (f, t) == ('uninitialized', 'freeslot'):

44 # Just don't bother, core is not running at this time

45 return ev

46

47 if (f, t) == ('connected', 'authed'):

48 self._user_join(c)

49

50 if t == 'dropped': 50 ↛ 51 line 50 didn't jump to line 51, because the condition on line 50 was never true

51 if f != 'connected':

52 self._user_leave(c)

53

54 ul = [u for u in self.users.values() if u._[self]['state'] == 'lobby']

55 self._notify_online_users(ul)

56

57 return ev

58

59 def handle_client_connected(self, c: Client) -> Client:

60 core = self.core

61 c._[self] = {

62 'state': FSM(

63 c,

64 [

65 'initial',

66 'connected',

67 'authed',

68 'lobby',

69 'room',

70 'ready',

71 'game',

72 'finishing',

73 'ob',

74 'dropped',

75 ],

76 'initial',

77 FSM.to_evhub(core.events.user_state_transition),

78 )

79 }

80 c._[self]['state'].transit('connected')

81 return c

82

83 def handle_game_ended(self, g: Game) -> Game:

84 core = self.core

85 users = core.room.users_of(g)

86

87 for u in users:

88 self.dropped_users.pop(core.auth.uid_of(u), 0)

89

90 return g

91

92 # ----- Client Commands -----

93 # ----- Public Methods -----

94 def state_of(self, u: Client) -> FSM:

95 return u._[self]['state']

96

97 def all_users(self) -> BatchList[Client]:

98 return BatchList(self.users.values())

99

100 def get(self, uid: int) -> Optional[Client]:

101 return self.users.get(uid)

102

103 def init_freeslot(self, c: Client) -> None:

104 core = self.core

105 c._[self] = {

106 'state': FSM(

107 c, ['uninitialized', 'freeslot'], 'uninitialized',

108 FSM.to_evhub(core.events.user_state_transition),

109 )

110 }

111

112 # ----- Methods -----

113 def _user_join(self, u: Client) -> None:

114 core = self.core

115 uid = core.auth.uid_of(u)

116 name = core.auth.name_of(u)

117

118 old = None

119

120 if uid in self.users: 120 ↛ 122 line 120 didn't jump to line 122, because the condition on line 120 was never true

121 # squeeze the original one out

122 log.info('%s[%s] has been squeezed out' % (name, uid))

123 old = self.users[uid]

124

125 if uid in self.dropped_users: 125 ↛ 126 line 125 didn't jump to line 126, because the condition on line 125 was never true

126 log.info('%s[%s] rejoining dropped game' % (name, uid))

127 old = self.dropped_users.pop(uid)

128

129 # XXX

130 '''

131 @core.runner.spawn

132 def reconnect():

133 self.send_account_info(user)

134 '''

135

136 if old: 136 ↛ 137 line 136 didn't jump to line 137, because the condition on line 136 was never true

137 u.pivot_to(old)

138 core.events.client_pivot.emit(old)

139 else:

140 self.users[uid] = u

141 self.state_of(u).transit('lobby')

142

143 log.info('User %s joined, online user %d' % (name, len(self.users)))

144

145 def _user_leave(self, u: Client) -> None:

146 core = self.core

147 uid = core.auth.uid_of(u)

148 name = core.auth.name_of(u)

149 self.users.pop(uid, 0)

150 log.info('User %s left, online user %d' % (name, len(self.users)))

151

152 @throttle(3)

153 def _notify_online_users(self, ul: Sequence[Client]) -> None:

154 core = self.core

155 lst = [core.view.User(u) for u in self.users.values()]

156 d = Endpoint.encode_bulk([wire.CurrentUsers(users=lst)])

157

158 @core.runner.spawn

159 def do_send() -> None:

160 for u in ul:

161 u.raw_write(d)