-
Notifications
You must be signed in to change notification settings - Fork 2
/
yroom.pyi
123 lines (117 loc) · 4.53 KB
/
yroom.pyi
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
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
from typing import Dict, List, Optional, TypedDict
class YRoomClientOptions:
"""
Client options for interacting with a YRoom.
`allow_write` flag indicates that the client can apply document updates.
`allow_write_awareness` flag indicates that the client can send awareness updates.
"""
allow_write: bool = True
allow_write_awareness: bool = True
class YRoomMessage:
"""
Container that holds three members: `payload`, `broadcast_payload` and `has_edits`.
`payload` is a message that should be sent to the connection that sent the message.
`broadcast_payload` is a message that should be sent to all connections in the room.
Either or both of the members can be of zero length and then must not be sent.
`has_edits` flag indicates if data changes occurred. Does not include awareness updates.
"""
payloads: List[bytes]
broadcast_payloads: List[bytes]
has_edits: bool
class YRoomSettings(TypedDict):
wire_version: int # The Yjs encoding/decoding version to use (1 or 2, default: 1)
name_prefixed: bool # Whether to expect a prefix for wire messages (default: False)
server_start_sync: bool # Whether to start sync on connect (default: True)
class YRoomManager:
def __init__(self, settings: Optional[Dict[str, YRoomSettings]]) -> None:
"""
Create a new YRoomManager with an optional settings dict.
"""
def connect(
self, room: str, conn_id: int, options: Optional[YRoomClientOptions]
) -> YRoomMessage:
"""
Connect to a room with a connection id, returning a YRoomMessage.
"""
def connect_with_data(
self,
room: str,
conn_id: int,
data: bytes,
options: Optional[YRoomClientOptions],
) -> YRoomMessage:
"""
Connect to a room with a connection id and bytes of serialized Yjs document in
`data`. The `data` will be used to initialize the Yjs document in the room if
the room needs to be created.
Returns a YRoomMessage.
"""
def handle_message(
self,
room: str,
conn_id: int,
payload: bytes,
options: Optional[YRoomClientOptions],
) -> YRoomMessage:
"""
Handle a Yjs protocol message from a connection in a room.
Returns a YRoomMessage.
"""
def disconnect(
self, room: str, conn_id: int, options: Optional[YRoomClientOptions]
) -> YRoomMessage:
"""
Disconnect a connection from a room and removes the associated client from the
awareness.
Returns a YRoomMessage.
"""
def has_room(self, room: str) -> bool:
"""
Returns True if the room exists, otherwise False.
"""
def is_room_alive(self, room: str) -> bool:
"""
Returns True if the room exists and has connections, otherwise False.
"""
def serialize_room(self, room: str) -> bytes:
"""
Encode the document of the room as an Yjs update in bytes.
"""
def remove_room(self, room: str) -> None:
"""
Remove the room, dropping the document, awareness and connection mapping.
"""
def list_rooms(self) -> List[str]:
"""
Return list all room names that are available.
"""
def get_map(self, room: str, name: str) -> Optional[str]:
"""
Return the named map from the doc of that room as a JSON string or
None if map or room does not exist.
"""
def get_array(self, room: str, name: str) -> Optional[str]:
"""
Return the named array from the doc of that room as a JSON string or
None if array or room does not exist.
"""
def get_text(self, room: str, name: str) -> Optional[str]:
"""
Return the named text from the doc of that room or
None if text or room does not exist.
"""
def get_xml_element(self, room: str, name: str) -> Optional[str]:
"""
Return the named xml element from the doc of that room as serialized string or
None if xml element or room does not exist.
"""
def get_xml_text(self, room: str, name: str) -> Optional[str]:
"""
Return the named xml text from the doc of that room as serialized string or
None if xml text or room does not exist.
"""
def get_xml_fragment(self, room: str, name: str) -> Optional[str]:
"""
Return the named xml fragment from the doc of that room as serialized string or
None if xml fragment or room does not exist.
"""