Skip to content

Protocol

elva.protocol

Module holding the Y-Protocol specification.

Classes:

Functions:

  • write_var_uint –

    Encode an integer into a variable unsigned integer.

  • read_var_uint –

    Read the first variable unsigned integer in data.

  • prepend_var_uint –

    Prepend the length of data as variable unsigned integer to data.

  • strip_var_uint –

    Read and strip off the variable unsigned integer length from data.

Attributes:

STATE_ZERO = b'\x00' module-attribute

The state of an empty YDoc.

EMPTY_UPDATE = b'\x00\x00' module-attribute

The update between two equivalent YDoc states.

YCodec

Bases: Codec

Codec for Y messages according to the Yjs base encoding.

Methods:

  • encode –

    Prepend the size of payload to itself as a variable unsigned integer.

  • decode –

    Read and strip off the encoded size from message.

encode(payload, errors='strict')

Prepend the size of payload to itself as a variable unsigned integer.

Parameters:

  • payload (bytes) –

    the payload of a Y protocol message.

  • errors (str, default: 'strict' ) –

    no-op.

Returns:

  • tuple[bytes, int] –

    A tuple of two values: data with the variable unsigned integer prepended and the length of data.

Source code in src/elva/protocol.py 
130
131
132
133
134
135
136
137
138
139
140
141
def encode(self, payload: bytes, errors: str = "strict") -> tuple[bytes, int]:
    """
    Prepend the size of `payload` to itself as a variable unsigned integer.

    Arguments:
        payload: the payload of a Y protocol message.
        errors: no-op.

    Returns:
        A tuple of two values: `data` with the variable unsigned integer prepended and the length of `data`.
    """
    return prepend_var_uint(payload)

decode(message, errors='strict')

Read and strip off the encoded size from message.

Parameters:

  • message (bytes) –

    the payload of a Y protocol message plus its length as variable unsigned integer prepended.

  • errors (str, default: 'strict' ) –

    no-op.

Returns:

  • tuple[bytes, int] –

    A tuple of two values: data with the variable unsigned integer stripped and the length of bytes of data being processed.

Source code in src/elva/protocol.py 
143
144
145
146
147
148
149
150
151
152
153
154
def decode(self, message: bytes, errors: str = "strict") -> tuple[bytes, int]:
    """
    Read and strip off the encoded size from `message`.

    Arguments:
        message: the payload of a Y protocol message plus its length as variable unsigned integer prepended.
        errors: no-op.

    Returns:
        A tuple of two values: `data` with the variable unsigned integer stripped and the length of bytes of `data` being processed.
    """
    return strip_var_uint(message)

Message(*magic_bytes)

Bases: YCodec, Enum

Base class for Y messages according to the Yjs sync and awareness protocol.

Parameters:

  • magic_bytes (bytes, default: () ) –

    arbitrary number of bytes prepended in the encoded payload.

Methods:

  • get_types –

    The message types associated with this codec.

  • encode –

    Calculate the encoded payload with the message type's magic bytes prepended.

  • decode –

    Remove magic bytes and extract the payload of message.

  • infer_and_decode –

    Infer the type of the given message and return its decoded form.

Source code in src/elva/protocol.py 
162
163
164
165
166
167
def __init__(self, *magic_bytes: bytes):
    """
    Arguments:
        magic_bytes: arbitrary number of bytes prepended in the encoded payload.
    """
    self.magic_bytes = bytes(magic_bytes)

get_types() classmethod

The message types associated with this codec.

Returns:

  • tuple[str] –

    a tuple containing all message type names as strings.

Source code in src/elva/protocol.py 
172
173
174
175
176
177
178
179
180
@classmethod
def get_types(cls) -> tuple[str]:
    """
    The message types associated with this codec.

    Returns:
        a tuple containing all message type names as strings.
    """
    return tuple(attr for attr in dir(cls) if not attr.startswith("__"))

encode(payload, errors='strict')

Calculate the encoded payload with the message type's magic bytes prepended.

Parameters:

  • payload (bytes) –

    the payload of a Y protocol message.

  • errors (str, default: 'strict' ) –

    no-op.

Returns:

  • tuple[bytes, int] –

    A tuple of two objects: the encoded payload with the message type's magic bytes prepended and the length of bytes being processed.

Source code in src/elva/protocol.py 
182
183
184
185
186
187
188
189
190
191
192
193
194
def encode(self, payload: bytes, errors: str = "strict") -> tuple[bytes, int]:
    """
    Calculate the encoded `payload` with the message type's magic bytes prepended.

    Arguments:
        payload: the payload of a Y protocol message.
        errors: no-op.

    Returns:
        A tuple of two objects: the encoded payload with the message type's magic bytes prepended and the length of bytes being processed.
    """
    message, length = super().encode(payload, errors=errors)
    return self.magic_bytes + message, length

decode(message, errors='strict')

Remove magic bytes and extract the payload of message.

Parameters:

  • message (bytes) –

    the Y protocol message.

  • errors (str, default: 'strict' ) –

    no-op.

Returns:

  • tuple[bytes, int] –

    A tuple of two objects: the decoded message with the message type's magic bytes removed and the length of bytes being processed.

Source code in src/elva/protocol.py 
196
197
198
199
200
201
202
203
204
205
206
207
208
209
def decode(self, message: bytes, errors: str = "strict") -> tuple[bytes, int]:
    """
    Remove magic bytes and extract the payload of `message`.

    Arguments:
        message: the Y protocol message.
        errors: no-op.

    Returns:
        A tuple of two objects: the decoded message with the message type's magic bytes removed and the length of bytes being processed.
    """
    message = message.removeprefix(self.magic_bytes)
    payload, length = self._decode(message, errors=errors)
    return payload, length + len(self.magic_bytes)

_decode(data, errors='strict')

Hook extracting the payload of message.

Parameters:

  • data (bytes) –

    the payload of a Y protocol message plus its length as variable unsigned integer prepended.

  • errors (str, default: 'strict' ) –

    no-op.

Returns:

  • tuple[bytes, int] –

    A tuple of two objects: data with the variable unsigned integer stripped and the length of bytes of data being processed.

Source code in src/elva/protocol.py 
211
212
213
214
215
216
217
218
219
220
221
222
def _decode(self, data: bytes, errors: str = "strict") -> tuple[bytes, int]:
    """
    Hook extracting the payload of `message`.

    Arguments:
        data: the payload of a Y protocol message plus its length as variable unsigned integer prepended.
        errors: no-op.

    Returns:
        A tuple of two objects: `data` with the variable unsigned integer stripped and the length of bytes of `data` being processed.
    """
    return super().decode(data, errors=errors)

infer_and_decode(message, errors='strict') classmethod

Infer the type of the given message and return its decoded form.

Parameters:

  • message (bytes) –

    the Y protocol message.

  • errors (str, default: 'strict' ) –

    no-op.

Returns:

  • tuple[Self, bytes, int] –

    A tuple of three objects: the inferred message type of message, the decoded form of message and the length of processed bytes from message.

Source code in src/elva/protocol.py 
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
@classmethod
def infer_and_decode(
    cls, message: bytes, errors: str = "strict"
) -> tuple[Self, bytes, int]:
    """
    Infer the type of the given message and return its decoded form.

    Arguments:
        message: the Y protocol message.
        errors: no-op.

    Returns:
        A tuple of three objects: the inferred message type of `message`, the decoded form of `message` and the length of processed bytes from `message`.
    """
    # there might be no second magic byte
    mb2_off = 0

    mb1, mb1_off = read_var_uint(message)

    try:
        if mb1 == 1:
            # awareness message is the only type with a single magic byte
            ymsg = cls((mb1,))
        else:
            mb2, mb2_off = read_var_uint(message[mb1_off:])
            ymsg = cls((mb1, mb2))
    except ValueError:
        raise ValueError(
            f"Message with magic bytes {mb1}, {mb2} is not a valid {cls.__name__}"
        ) from None

    payload, length = ymsg._decode(message[mb1_off + mb2_off :], errors=errors)
    return ymsg, payload, mb1_off + mb2_off + length

YMessage(*magic_bytes)

Bases: Message

Implementation of Y messages according to the Yjs sync and awareness protocol.

Attributes:

Source code in src/elva/protocol.py 
162
163
164
165
166
167
def __init__(self, *magic_bytes: bytes):
    """
    Arguments:
        magic_bytes: arbitrary number of bytes prepended in the encoded payload.
    """
    self.magic_bytes = bytes(magic_bytes)

SYNC_STEP1 = (0, 0) class-attribute instance-attribute

Synchronization request message.

SYNC_STEP2 = (0, 1) class-attribute instance-attribute

Synchronization reply message.

SYNC_UPDATE = (0, 2) class-attribute instance-attribute

Update message.

AWARENESS = (1,) class-attribute instance-attribute

Awareness message.

ElvaMessage(*magic_bytes)

Bases: Message

Extension of Y messages with additional message types.

Attributes:

Source code in src/elva/protocol.py 
162
163
164
165
166
167
def __init__(self, *magic_bytes: bytes):
    """
    Arguments:
        magic_bytes: arbitrary number of bytes prepended in the encoded payload.
    """
    self.magic_bytes = bytes(magic_bytes)

SYNC_STEP1 = (0, 0) class-attribute instance-attribute

Synchronization request message.

SYNC_STEP2 = (0, 1) class-attribute instance-attribute

Synchronization reply message.

SYNC_UPDATE = (0, 2) class-attribute instance-attribute

Update message.

SYNC_CROSS = (0, 3) class-attribute instance-attribute

Cross-synchronization message holding SYNC_STEP1 and SYNC_STEP2.

AWARENESS = (1,) class-attribute instance-attribute

Awareness message.

ID = (ord('I'), ord('D')) class-attribute instance-attribute

Identitity message.

READ = (ord('R'), ord('O')) class-attribute instance-attribute

Read-only message.

READ_WRITE = (ord('R'), ord('W')) class-attribute instance-attribute

Read-write message.

DATA_REQUEST = (ord('D'), 1) class-attribute instance-attribute

Message requesting a specific blob of data.

DATA_OFFER = (ord('D'), 2) class-attribute instance-attribute

Message offering a requested blob of data.

DATA_ORDER = (ord('D'), 3) class-attribute instance-attribute

Message ordering an offered blob of data.

DATA_TRANSFER = (ord('D'), 4) class-attribute instance-attribute

Message transferring an ordered blob of data.

write_var_uint(num)

Encode an integer into a variable unsigned integer.

Parameters:

  • num (int) –

    the integer to encode.

Returns:

  • list[int] –

    a list of integers representing the bits of the variable unsigned integer.

Source code in src/elva/protocol.py 
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
def write_var_uint(num: int) -> list[int]:
    """
    Encode an integer into a variable unsigned integer.

    Arguments:
        num: the integer to encode.

    Returns:
        a list of integers representing the bits of the variable unsigned integer.
    """
    res = []

    # while `num` is bigger than 1 byte minus 1 continuation bit, i.e. 7 bits
    while num > 127:
        #        127 & num   <=>  extract the last 7 bits of `num`
        # 128 | (127 & num)  <=>  put a "1" as continuation bit before the last 7 bits of `num`
        res.append(128 | (127 & num))

        # discard the last 7 bits of `num`
        num >>= 7

    # append the remaining bits in `num`
    res.append(num)

    return res

read_var_uint(data)

Read the first variable unsigned integer in data.

Parameters:

  • data (bytes) –

    data holding the variable unsigned integer.

Returns:

  • tuple[int, int] –

    a tuple of the decoded variable unsigned integer and its number of bytes

Source code in src/elva/protocol.py 
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
def read_var_uint(data: bytes) -> tuple[int, int]:
    """
    Read the first variable unsigned integer in `data`.

    Arguments:
        data: data holding the variable unsigned integer.

    Returns:
        a tuple of the decoded variable unsigned integer and its number of bytes
    """
    # we start at the very first byte/bit of `data`
    uint = 0
    bit = 0
    byte_idx = 0

    while True:
        byte = data[byte_idx]

        #  byte & 127          <=>  extract the last 7 bits in the current `byte`
        # (byte & 127) << bit  <=>  shift bits in `res` `bit` times to the left
        uint += (byte & 127) << bit

        # move the bit offset by 7
        bit += 7

        # move the byte offset by 1
        byte_idx += 1

        if byte < 128:
            # the first bit of this byte, i.e. the continuation bit, is not a "1",
            # so this is the last byte being part of the variable unsigned integer
            break

    return uint, byte_idx

prepend_var_uint(data)

Prepend the length of data as variable unsigned integer to data.

See Yjs base encoding for reference.

Parameters:

  • data (bytes) –

    the payload of a Y protocol message.

Returns:

  • tuple[bytes, int] –

    A tuple of two values: data with the variable unsigned integer prepended and the length of data.

Source code in src/elva/protocol.py 
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
def prepend_var_uint(data: bytes) -> tuple[bytes, int]:
    """
    Prepend the length of `data` as variable unsigned integer to `data`.

    See [Yjs base encoding](https://github.com/yjs/y-protocols/blob/master/PROTOCOL.md#base-encoding-approaches) for reference.

    Arguments:
        data: the payload of a Y protocol message.

    Returns:
        A tuple of two values: `data` with the variable unsigned integer prepended and the length of `data`.
    """
    len_data = len(data)
    res = write_var_uint(len_data)

    return bytes(res) + data, len_data

strip_var_uint(data)

Read and strip off the variable unsigned integer length from data.

See Yjs base encoding for reference.

Parameters:

  • data (bytes) –

    the payload of a Y protocol message plus its length as variable unsigned integer prepended.

Returns:

  • tuple[bytes, int] –

    A tuple of two values: data with the variable unsigned integer length stripped and the length of bytes of data being processed.

Source code in src/elva/protocol.py 
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
def strip_var_uint(data: bytes) -> tuple[bytes, int]:
    """
    Read and strip off the variable unsigned integer length from `data`.

    See [Yjs base encoding](https://github.com/yjs/y-protocols/blob/master/PROTOCOL.md#base-encoding-approaches) for reference.

    Arguments:
        data: the payload of a Y protocol message plus its length as variable unsigned integer prepended.

    Returns:
        A tuple of two values: `data` with the variable unsigned integer length stripped and the length of bytes of `data` being processed.
    """
    uint, byte_idx = read_var_uint(data)

    # extract the payload from `data`, return the actual length of the payload
    return data[byte_idx : byte_idx + uint], min(byte_idx + uint, len(data))