1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
|
import asyncio
from .base_manager import BaseManager
class AsyncManager(BaseManager):
"""Manage a client list for an asyncio server."""
async def can_disconnect(self, sid, namespace):
return self.is_connected(sid, namespace)
async def emit(self, event, data, namespace, room=None, skip_sid=None,
callback=None, **kwargs):
"""Emit a message to a single client, a room, or all the clients
connected to the namespace.
Note: this method is a coroutine.
"""
if namespace not in self.rooms or room not in self.rooms[namespace]:
return
tasks = []
if not isinstance(skip_sid, list):
skip_sid = [skip_sid]
for sid, eio_sid in self.get_participants(namespace, room):
if sid not in skip_sid:
if callback is not None:
id = self._generate_ack_id(sid, callback)
else:
id = None
tasks.append(self.server._emit_internal(eio_sid, event, data,
namespace, id))
if tasks == []: # pragma: no cover
return
await asyncio.wait(tasks)
async def close_room(self, room, namespace):
"""Remove all participants from a room.
Note: this method is a coroutine.
"""
return super().close_room(room, namespace)
async def trigger_callback(self, sid, id, data):
"""Invoke an application callback.
Note: this method is a coroutine.
"""
callback = None
try:
callback = self.callbacks[sid][id]
except KeyError:
# if we get an unknown callback we just ignore it
self._get_logger().warning('Unknown callback received, ignoring.')
else:
del self.callbacks[sid][id]
if callback is not None:
ret = callback(*data)
if asyncio.iscoroutine(ret):
try:
await ret
except asyncio.CancelledError: # pragma: no cover
pass
|
