Skip to content

Protocol

elva.protocol

Module holding the Y-Protocol specification.

Classes:

Functions:

  • write_var_uint –

    Calculate the variable unsigned integer length of data.

  • read_var_uint –

    Read and strip off the variable unsigned interger value from data.

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 payload.

encode(payload, errors='strict')

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

Parameters:

  • payload (bytes) –

    data to encode

  • 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 
57
58
59
60
61
62
63
64
65
66
67
68
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: data to encode
        errors: no-op.

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

decode(message, errors='strict')

Read and strip off the encoded size from payload.

Parameters:

  • message (bytes) –

    data to decode

  • 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 
70
71
72
73
74
75
76
77
78
79
80
81
def decode(self, message: bytes, errors: str = "strict") -> tuple[bytes, int]:
    """
    Read and strip off the encoded size from `payload`.

    Arguments:
        message: data to decode
        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 read_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:

  • encode –

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

  • decode –

    Remove magic bytes and decode message.

  • infer_and_decode –

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

Source code in src/elva/protocol.py 
125
126
127
128
129
130
def __init__(self, *magic_bytes: bytes):
    """
    Arguments:
        magic_bytes: arbitrary number of bytes prepended in the encoded payload.
    """
    self.magic_bytes = bytes(magic_bytes)

encode(payload, errors='strict')

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

Parameters:

  • payload (bytes) –

    data to be encoded.

  • 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 
135
136
137
138
139
140
141
142
143
144
145
146
147
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: data to be encoded.
        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 decode message.

Parameters:

  • message (bytes) –

    data to be decoded.

  • 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 
149
150
151
152
153
154
155
156
157
158
159
160
161
162
def decode(self, message: bytes, errors: str = "strict") -> tuple[bytes, int]:
    """
    Remove magic bytes and decode `message`.

    Arguments:
        message: data to be decoded.
        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 = super().decode(message, errors=errors)
    return payload, length + len(self.magic_bytes)

infer_and_decode(message, errors='strict') classmethod

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

Parameters:

  • message (bytes) –

    data to decode

  • 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 
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
@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: data to decode
        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`
    """
    first = message[0]
    match first:
        case 0:
            ymsg = cls((first, message[1]))
            return ymsg, *ymsg.decode(message, errors=errors)
        case 1:
            ymsg = cls((first,))
            return ymsg, *ymsg.decode(message, errors=errors)
        case _:
            raise ValueError(f"given message '{message}' is not a valid YMessage")

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 
125
126
127
128
129
130
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 
125
126
127
128
129
130
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 = (2, 0) class-attribute instance-attribute

Identitity message.

READ = (2, 1) class-attribute instance-attribute

Read-only message.

READ_WRITE = (2, 2) class-attribute instance-attribute

Read-write message.

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

Message requesting a specific blob of data.

DATA_OFFER = (3, 1) class-attribute instance-attribute

Message offering a requested blob of data.

DATA_ORDER = (3, 2) class-attribute instance-attribute

Message ordering a offered blob of data.

DATA_TRANSFER = (3, 3) class-attribute instance-attribute

Message transferring a ordered blob of data.

write_var_uint(data)

Calculate the variable unsigned integer length of data.

Parameters:

  • data (bytes) –

    bytes object.

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 
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
def write_var_uint(data: bytes) -> tuple[bytes, int]:
    """
    Calculate the variable unsigned integer length of `data`.

    Arguments:
        data: bytes object.

    Returns:
        A tuple of two values: `data` with the variable unsigned integer prepended and the length of `data`.
    """
    num = len(data)
    res = []
    while num > 127:
        res.append(128 | (127 & num))
        num >>= 7
    res.append(num)
    return bytes(res) + data, len(data)

read_var_uint(data)

Read and strip off the variable unsigned interger value from data.

Parameters:

  • data (bytes) –

    bytes object.

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 
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
def read_var_uint(data: bytes) -> tuple[bytes, int]:
    """
    Read and strip off the variable unsigned interger value from `data`.

    Arguments:
        data: bytes object.

    Returns:
        A tuple of two values: `data` with the variable unsigned integer stripped and the length of bytes of `data` being processed.
    """
    uint = 0
    bit = 0
    byte_idx = 0
    while True:
        byte = data[byte_idx]
        uint += (byte & 127) << bit
        bit += 7
        byte_idx += 1
        if byte < 128:
            break
    return data[byte_idx : byte_idx + uint], min(byte_idx + uint, len(data))