hololinked.server.zmq.message.RequestMessage

A single unit of message from a ZMQ client to server. The message may be parsed and deserialized into header and body.

Message indices:

Index	0	1	2	3	4
Desc	address	empty byte	header	payload	preserialized payload

For header's JSON schema, visit here.

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

class RequestMessage:
    """
    A single unit of message from a ZMQ client to server. The message may be parsed and deserialized into header and body.

    Message indices:

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

    For header's JSON schema, visit [here](https://github.com/hololinked-dev/hololinked/blob/main/hololinked/server/zmq/request_message_header_schema.json).
    """

    length = Integer(default=5, readonly=True, class_member=True, doc="length of the message")  # type: int

    _header: RequestHeader
    _body: list[SerializableData | PreserializedData]

    def __init__(self, msg: list[bytes]) -> None:
        self._bytes = msg
        self._header = None  # ty: ignore[invalid-assignment]  # deserialized header
        self._body = None  # ty: ignore[invalid-assignment]  # type: Optional[tuple[SerializableData, PreserializedData]]
        self._sender_id = None

    @property
    def byte_array(self) -> list[bytes]:
        """
        Message byte array, either after being composed or as received from the socket.

        Message indices:

        | Index | 0       | 1          | 2      |   3     |          4            |
        |-------|---------|------------|--------|---------|-----------------------|
        | Desc  | address | empty byte | header | payload | preserialized payload |
        """
        return self._bytes

    @property
    def header(self) -> RequestHeader:
        """Header of the message, namely index 1 of the byte array, deserizalized to a dictionary."""
        if self._header is None:
            self.parse_header()
        return self._header

    @property
    def body(self) -> list[SerializableData | PreserializedData]:
        """Body of the message."""
        if self._body is None:
            self.parse_body()
        return self._body

    @property
    def id(self) -> str:
        """ID of the message."""
        return self.header["messageID"]

    @property
    def receiver_id(self) -> str:
        """ID of the sender."""
        return self.header["receiverID"]

    @property
    def sender_id(self) -> str:
        """ID of the receiver."""
        return self.header["senderID"]

    @property
    def thing_id(self) -> str:
        """ID of the thing on which the operation is to be performed."""
        return self.header["thingID"]

    @property
    def type(self) -> str:
        """Type of the message."""
        return self.header["messageType"]

    @property
    def server_execution_context(self) -> dict[str, Any]:
        """Server execution context."""
        return self.header["serverExecutionContext"]

    @property
    def thing_execution_context(self) -> dict[str, Any]:
        """Thing execution context."""
        return self.header["thingExecutionContext"]

    @property
    def qualified_name(self) -> str:
        """A key identifying this operation on this affordance of this `Thing`."""
        return qualified_operation_key(
            self.header["thingID"],
            self.header["objekt"],
            self.header["operation"],
        )

    def parse_header(self) -> None:
        """
        Extract the header and deserialize it.

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

    def parse_body(self) -> None:
        """Extract the body and deserialize payload."""
        self._body = [
            SerializableData(self._bytes[INDEX_BODY], content_type=self.header["payloadContentType"]),
            PreserializedData(
                self._bytes[INDEX_PRESERIALIZED_BODY],
                content_type=self.header["preencodedPayloadContentType"],
            ),
        ]

    @classmethod
    def craft_from_arguments(
        cls,
        receiver_id: str,
        sender_id: str,
        thing_id: str,
        objekt: str,
        operation: str,
        payload: SerializableData = SerializableNone,
        preserialized_payload: PreserializedData = PreserializedEmptyByte,
        server_execution_context: SchedulerExecutionContext | dict[str, Any] = default_scheduler_execution_context,
        thing_execution_context: ThingExecutionContext | dict[str, Any] = default_thing_execution_context,
    ) -> "RequestMessage":
        """
        Create a request message from the given arguments.

        Parameters
        ----------
        receiver_id: str
            id of the server (ZMQ socket identity)
        sender_id: str
            id of the client (ZMQ socket identity)
        thing_id: str
            id of the thing to which the operation is to be performed
        objekt: str
            objekt of the thing on which the operation is to be performed, i.e. a property, action or event name
        operation: str
            operation to be performed (`invokeaction`, `readproperty`, `writeproperty` etc.)
        payload: SerializableData
            payload for the operation
        preserialized_payload: PreserializedData
            pre-encoded payload for the operation
        server_execution_context: Dict[str, Any]
            server-level execution context while performing the operation
        thing_execution_context: Dict[str, Any]
            thing-level execution context while performing the operation

        Returns
        -------
        message: RequestMessage
            the crafted message
        """
        message = RequestMessage([])
        message._header = RequestHeader(
            messageID=str(uuid4()),
            messageType=OPERATION,
            senderID=sender_id,
            receiverID=receiver_id,
            # i.e. the message type is 'OPERATION', not 'HANDSHAKE', 'REPLY', 'TIMEOUT' etc.
            # clients may pass a plain dict; msgspec stores it as given and it serializes the same way
            serverExecutionContext=server_execution_context,  # ty: ignore[invalid-argument-type]
            thingID=thing_id,
            objekt=objekt,
            operation=operation,
            payloadContentType=payload.content_type,
            preencodedPayloadContentType=preserialized_payload.content_type,
            thingExecutionContext=thing_execution_context,  # ty: ignore[invalid-argument-type]
        )
        message._body = [payload, preserialized_payload]
        message._bytes = [
            bytes(receiver_id, encoding="utf-8"),
            b"",
            Serializers.json.dumps(message._header.json()),
            payload.serialize(),
            preserialized_payload.value,
        ]
        return message

    @classmethod
    def craft_with_message_type(
        cls,
        sender_id: str,
        receiver_id: str,
        message_type: str = HANDSHAKE,
    ) -> "RequestMessage":
        """
        Create a plain message with a certain type, for example a handshake message.

        Parameters
        ----------
        sender_id: str
            id of the client (ZMQ socket identity)
        receiver_id: str
            id of the server (ZMQ socket identity)
        message_type: str
            message type to be sent (i.e. 'HANDSHAKE', 'EXIT' etc.)

        Returns
        -------
        message: RequestMessage
            the crafted message
        """
        message = RequestMessage([])
        message._header = RequestHeader(
            messageID=str(uuid4()),
            messageType=message_type,
            senderID=sender_id,
            receiverID=receiver_id,
            serverExecutionContext=default_scheduler_execution_context,
        )
        payload = SerializableNone
        preserialized_payload = PreserializedEmptyByte
        message._body = [payload, preserialized_payload]
        message._bytes = [
            bytes(receiver_id, encoding="utf-8"),
            b"",
            Serializers.json.dumps(message._header.json()),
            payload.serialize(),
            preserialized_payload.value,
        ]
        return message

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

Attributes

header property

header: RequestHeader

Header of the message, namely index 1 of the byte array, deserizalized to a dictionary.

body property

body: list[SerializableData | PreserializedData]

Body of the message.

thing_id property

thing_id: str

ID of the thing on which the operation is to be performed.

byte_array property

byte_array: list[bytes]

Message byte array, either after being composed or as received from the socket.

Message indices:

Index	0	1	2	3	4
Desc	address	empty byte	header	payload	preserialized payload

Functions

__init__

__init__(msg: list[bytes]) -> None

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

def __init__(self, msg: list[bytes]) -> None:
    self._bytes = msg
    self._header = None  # ty: ignore[invalid-assignment]  # deserialized header
    self._body = None  # ty: ignore[invalid-assignment]  # type: Optional[tuple[SerializableData, PreserializedData]]
    self._sender_id = None

craft_from_arguments classmethod

craft_from_arguments(receiver_id: str, sender_id: str, thing_id: str, objekt: str, operation: str, payload: SerializableData = SerializableNone, preserialized_payload: PreserializedData = PreserializedEmptyByte, server_execution_context: SchedulerExecutionContext | dict[str, Any] = default_scheduler_execution_context, thing_execution_context: ThingExecutionContext | dict[str, Any] = default_thing_execution_context) -> RequestMessage

Create a request message from the given arguments.

Parameters:

Name	Type	Description	Default
receiver_id	str	id of the server (ZMQ socket identity)	required
sender_id	str	id of the client (ZMQ socket identity)	required
thing_id	str	id of the thing to which the operation is to be performed	required
objekt	str	objekt of the thing on which the operation is to be performed, i.e. a property, action or event name	required
operation	str	operation to be performed (invokeaction, readproperty, writeproperty etc.)	required
payload	SerializableData	payload for the operation	SerializableNone
preserialized_payload	PreserializedData	pre-encoded payload for the operation	PreserializedEmptyByte
server_execution_context	SchedulerExecutionContext | dict[str, Any]	server-level execution context while performing the operation	default_scheduler_execution_context
thing_execution_context	ThingExecutionContext | dict[str, Any]	thing-level execution context while performing the operation	default_thing_execution_context

Returns:

Name	Type	Description
message	RequestMessage	the crafted message

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

@classmethod
def craft_from_arguments(
    cls,
    receiver_id: str,
    sender_id: str,
    thing_id: str,
    objekt: str,
    operation: str,
    payload: SerializableData = SerializableNone,
    preserialized_payload: PreserializedData = PreserializedEmptyByte,
    server_execution_context: SchedulerExecutionContext | dict[str, Any] = default_scheduler_execution_context,
    thing_execution_context: ThingExecutionContext | dict[str, Any] = default_thing_execution_context,
) -> "RequestMessage":
    """
    Create a request message from the given arguments.

    Parameters
    ----------
    receiver_id: str
        id of the server (ZMQ socket identity)
    sender_id: str
        id of the client (ZMQ socket identity)
    thing_id: str
        id of the thing to which the operation is to be performed
    objekt: str
        objekt of the thing on which the operation is to be performed, i.e. a property, action or event name
    operation: str
        operation to be performed (`invokeaction`, `readproperty`, `writeproperty` etc.)
    payload: SerializableData
        payload for the operation
    preserialized_payload: PreserializedData
        pre-encoded payload for the operation
    server_execution_context: Dict[str, Any]
        server-level execution context while performing the operation
    thing_execution_context: Dict[str, Any]
        thing-level execution context while performing the operation

    Returns
    -------
    message: RequestMessage
        the crafted message
    """
    message = RequestMessage([])
    message._header = RequestHeader(
        messageID=str(uuid4()),
        messageType=OPERATION,
        senderID=sender_id,
        receiverID=receiver_id,
        # i.e. the message type is 'OPERATION', not 'HANDSHAKE', 'REPLY', 'TIMEOUT' etc.
        # clients may pass a plain dict; msgspec stores it as given and it serializes the same way
        serverExecutionContext=server_execution_context,  # ty: ignore[invalid-argument-type]
        thingID=thing_id,
        objekt=objekt,
        operation=operation,
        payloadContentType=payload.content_type,
        preencodedPayloadContentType=preserialized_payload.content_type,
        thingExecutionContext=thing_execution_context,  # ty: ignore[invalid-argument-type]
    )
    message._body = [payload, preserialized_payload]
    message._bytes = [
        bytes(receiver_id, encoding="utf-8"),
        b"",
        Serializers.json.dumps(message._header.json()),
        payload.serialize(),
        preserialized_payload.value,
    ]
    return message

craft_with_message_type classmethod

craft_with_message_type(sender_id: str, receiver_id: str, message_type: str = HANDSHAKE) -> RequestMessage

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

Parameters:

Name	Type	Description	Default
sender_id	str	id of the client (ZMQ socket identity)	required
receiver_id	str	id of the server (ZMQ socket identity)	required
message_type	str	message type to be sent (i.e. 'HANDSHAKE', 'EXIT' etc.)	HANDSHAKE

Returns:

Name	Type	Description
message	RequestMessage	the crafted message

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

@classmethod
def craft_with_message_type(
    cls,
    sender_id: str,
    receiver_id: str,
    message_type: str = HANDSHAKE,
) -> "RequestMessage":
    """
    Create a plain message with a certain type, for example a handshake message.

    Parameters
    ----------
    sender_id: str
        id of the client (ZMQ socket identity)
    receiver_id: str
        id of the server (ZMQ socket identity)
    message_type: str
        message type to be sent (i.e. 'HANDSHAKE', 'EXIT' etc.)

    Returns
    -------
    message: RequestMessage
        the crafted message
    """
    message = RequestMessage([])
    message._header = RequestHeader(
        messageID=str(uuid4()),
        messageType=message_type,
        senderID=sender_id,
        receiverID=receiver_id,
        serverExecutionContext=default_scheduler_execution_context,
    )
    payload = SerializableNone
    preserialized_payload = PreserializedEmptyByte
    message._body = [payload, preserialized_payload]
    message._bytes = [
        bytes(receiver_id, encoding="utf-8"),
        b"",
        Serializers.json.dumps(message._header.json()),
        payload.serialize(),
        preserialized_payload.value,
    ]
    return message

parse_header

parse_header() -> None

Extract the header and deserialize it.

Raises:

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

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

def parse_header(self) -> None:
    """
    Extract the header and deserialize it.

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

parse_body

parse_body() -> None

Extract the body and deserialize payload.

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

def parse_body(self) -> None:
    """Extract the body and deserialize payload."""
    self._body = [
        SerializableData(self._bytes[INDEX_BODY], content_type=self.header["payloadContentType"]),
        PreserializedData(
            self._bytes[INDEX_PRESERIALIZED_BODY],
            content_type=self.header["preencodedPayloadContentType"],
        ),
    ]