Coverage for server/parts/connect.py : 67%
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 -*-
2from __future__ import annotations
4# -- stdlib --
5from typing import TYPE_CHECKING, Any, List
6import json
7import logging
9# -- third party --
10import websocket
12# -- own --
13from utils.misc import throttle
15# -- typing --
16if TYPE_CHECKING:
17 from server.core import Core # noqa: F401
20# -- code --
21log = logging.getLogger('Connect')
24class Connect(object):
25 def __init__(self, core: Core):
26 self.core = core
28 self._wsconn = None
30 core.events.game_created += self.refresh_status
31 core.events.game_started += self.refresh_status
32 core.events.game_ended += self.refresh_status
33 core.events.game_aborted += self.refresh_status
34 core.events.user_state_transition += self.refresh_status
36 def __repr__(self) -> str:
37 return self.__class__.__name__
39 def refresh_status(self, ev: Any) -> Any:
40 self._refresh_status()
41 return ev
43 # ----- Public Methods -----
44 def speaker(self, name: str, text: str) -> None:
45 core = self.core
46 self._wssend({
47 'op': 'Message',
48 'arg': {
49 'entity': 'Speaker',
50 'channel': core.options.node,
51 'text': text,
52 }
53 })
55 # ----- Methods -----
56 @throttle(1.5)
57 def _refresh_status(self) -> None:
58 core = self.core
59 self._wssend({
60 'op': 'Message',
61 'arg': {
62 'entity': 'Interconnect',
63 'channel': 'users',
64 'text': json.dumps([
65 core.view.User(u)
66 for u in core.lobby.all_users()
67 ]),
68 }
69 })
71 def _wssend(self, v: Any) -> None:
72 core = self.core
73 for i in range(3):
74 try:
75 conn = self._wsconn
76 if not conn or not conn.connected:
77 conn = websocket.create_connection(core.options.interconnect)
78 self._wsconn = conn
79 conn.send(json.dumps(v))
80 return
81 except Exception:
82 log.exception('Error sending interconnect message')
83 core.runner.sleep(3)
85 log.error('WebSocket send with multiple failed attempts, giving up, message: %s', v)
88class MockConnect(Connect):
89 def __init__(self, core: Core):
90 super().__init__(core)
92 self.mock_sent_messages: List[Any] = []
94 def _wssend(self, v: Any) -> None:
95 self.mock_sent_messages.append(v)
96 log.info("MockClient: Send %s", v)