hololinked.server.zmq.message.EventMessage

Bases: ResponseMessage

Event message class for handling events in the system.

Message indices:

Index	0	1	2	3
Desc	address	header	payload	preserialized payload

Source code in repo/hololinked/hololinked/server/zmq/message.py

class EventMessage(ResponseMessage):
    """
    Event message class for handling events in the system.

    Message indices:

    | Index | 0       | 1      |   2     |          3            |
    |-------|---------|--------|---------|-----------------------|
    | Desc  | address | header | payload | preserialized payload |

    """

    # For header's JSON schema, visit [here](https://hololinked.readthedocs.io/en/latest/protocols/zmq/event-message-header.json).

    @classmethod
    def craft_from_arguments(  # ty: ignore[invalid-method-override]
        cls,
        event_id: str,
        sender_id: str,
        message_type: str = EVENT,
        payload: SerializableData = SerializableNone,
        preserialized_payload: PreserializedData = PreserializedEmptyByte,
    ) -> "EventMessage":
        """
        Create a plain message with a certain type, for example a handshake message.

        Parameters
        ----------
        event_id: str
            id of the event (used by ZMQ pub-sub)
        sender_id: str
            id of the sender (ZMQ socket identity)
        message_type: str
            message type to be sent (`EVENT` usually)
        payload: SerializableData
            event payload to send to the client
        preserialized_payload: PreserializedData
            pre-encoded data, generally used for large or custom data that is already serialized

        Returns
        -------
        EventMessage
            the crafted message
        """
        message = EventMessage([])
        message._header = EventHeader(
            messageType=message_type,
            messageID=str(uuid4()),
            eventID=event_id,
            senderID=sender_id,
            payloadContentType=payload.content_type,
            preencodedPayloadContentType=preserialized_payload.content_type,
        )
        message._body = [payload, preserialized_payload]
        message._bytes = [
            bytes(event_id, encoding="utf-8"),
            b"",
            Serializers.json.dumps(message._header.json()),
            payload.serialize(),
            preserialized_payload.value,
        ]
        return message

    @property
    def event_id(self) -> str:
        """Unique ID of the event by which ZMQ pub-sub works."""
        return self.header["eventID"]

    def parse_header(self) -> None:
        """
        Parse the header.

        Raises
        ------
        ValueError
            if the header is neither a header struct nor bytes
        """
        header = self._bytes[INDEX_HEADER]
        if isinstance(header, EventHeader):
            self._header = header
        elif isinstance(header, byte_types):
            self._header = EventHeader(**Serializers.json.loads(header))
        else:
            raise ValueError(f"header must be of type ResponseHeader or bytes, not {type(self._bytes[INDEX_HEADER])}")

    def __str__(self) -> str:
        return f"EventMessage(id={self.id}, type={self.type}, header={self.header})"

Attributes

header property

header: ResponseHeader | EventHeader

Header of the message.

body property

body: list[SerializableData | PreserializedData]

Body of the message.

event_id property

event_id: str

Unique ID of the event by which ZMQ pub-sub works.

id property

id: str

ID of the message.

type property

type: str

Type of the message.

sender_id property

sender_id: str

ID of the receiver.

receiver_id property

receiver_id: str

ID of the sender.

byte_array property

byte_array: list[bytes]

The message in bytes, either after being composed or as received from the socket.

Message indices:

Index	0	2	3	4
Desc	address	header	data	pre encoded data

Functions

__init__

__init__(msg: list[bytes])

Source code in repo/hololinked/hololinked/server/zmq/message.py

def __init__(self, msg: list[bytes]):
    self._bytes = msg
    # header and body are only populated by parse_header()/parse_body(); the `None` here is a placeholder,
    # so the class-level annotations above state the contract that every use site relies on
    self._header = None  # ty: ignore[invalid-assignment]
    self._body = None  # ty: ignore[invalid-assignment]
    self._sender_id = None

craft_from_arguments classmethod

craft_from_arguments(event_id: str, sender_id: str, message_type: str = EVENT, payload: SerializableData = SerializableNone, preserialized_payload: PreserializedData = PreserializedEmptyByte) -> EventMessage

Create a plain message with a certain type, for example a handshake message.

Parameters:

Name	Type	Description	Default
event_id	str	id of the event (used by ZMQ pub-sub)	required
sender_id	str	id of the sender (ZMQ socket identity)	required
message_type	str	message type to be sent (EVENT usually)	EVENT
payload	SerializableData	event payload to send to the client	SerializableNone
preserialized_payload	PreserializedData	pre-encoded data, generally used for large or custom data that is already serialized	PreserializedEmptyByte

Returns:

Type	Description
EventMessage	the crafted message

Source code in repo/hololinked/hololinked/server/zmq/message.py

@classmethod
def craft_from_arguments(  # ty: ignore[invalid-method-override]
    cls,
    event_id: str,
    sender_id: str,
    message_type: str = EVENT,
    payload: SerializableData = SerializableNone,
    preserialized_payload: PreserializedData = PreserializedEmptyByte,
) -> "EventMessage":
    """
    Create a plain message with a certain type, for example a handshake message.

    Parameters
    ----------
    event_id: str
        id of the event (used by ZMQ pub-sub)
    sender_id: str
        id of the sender (ZMQ socket identity)
    message_type: str
        message type to be sent (`EVENT` usually)
    payload: SerializableData
        event payload to send to the client
    preserialized_payload: PreserializedData
        pre-encoded data, generally used for large or custom data that is already serialized

    Returns
    -------
    EventMessage
        the crafted message
    """
    message = EventMessage([])
    message._header = EventHeader(
        messageType=message_type,
        messageID=str(uuid4()),
        eventID=event_id,
        senderID=sender_id,
        payloadContentType=payload.content_type,
        preencodedPayloadContentType=preserialized_payload.content_type,
    )
    message._body = [payload, preserialized_payload]
    message._bytes = [
        bytes(event_id, encoding="utf-8"),
        b"",
        Serializers.json.dumps(message._header.json()),
        payload.serialize(),
        preserialized_payload.value,
    ]
    return message

parse_header

parse_header() -> None

Parse the header.

Raises:

Type	Description
ValueError	if the header is neither a header struct nor bytes

Source code in repo/hololinked/hololinked/server/zmq/message.py

def parse_header(self) -> None:
    """
    Parse the header.

    Raises
    ------
    ValueError
        if the header is neither a header struct nor bytes
    """
    header = self._bytes[INDEX_HEADER]
    if isinstance(header, EventHeader):
        self._header = header
    elif isinstance(header, byte_types):
        self._header = EventHeader(**Serializers.json.loads(header))
    else:
        raise ValueError(f"header must be of type ResponseHeader or bytes, not {type(self._bytes[INDEX_HEADER])}")