hololinked.server.zmq.message.ResponseMessage

A single unit of message from a ZMQ server to client.

The message may be parsed and deserialized into header and body.

Message indices:

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

For header's JSON schema, visit here.

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

class ResponseMessage:
    """
    A single unit of message from a ZMQ server to client.

    The message may be parsed and deserialized into header and body.

    Message indices:

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

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

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

    # `EventMessage` subclasses this and stores an `EventHeader` instead
    _header: ResponseHeader | EventHeader
    _body: list[SerializableData | PreserializedData]

    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

    @property
    def byte_array(self) -> 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 |
        """
        return self._bytes

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

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

    @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 header(self) -> ResponseHeader | EventHeader:
        """Header of the message."""
        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 payload(self) -> SerializableData:
        """Payload of the message."""
        return self.body[0]  # ty: ignore[invalid-return-type]

    @property
    def preserialized_payload(self) -> PreserializedData:
        """pre-encoded payload of the message."""
        return self.body[1]  # ty: ignore[invalid-return-type]

    @property
    def oneof_valid_payload(self) -> SerializableData | PreserializedData:
        """
        Checks if only one of payload or preserialized payload is valid (non-empty), and returns that.

        To be used with non-multipart messages (multipart as in
        containing multiple content types). This property can lead to loss of information
        if any response contains both payload and preserialized payload.
        """
        if self._body[1].value != b"":
            return self._body[1]
        return self._body[0]

    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, ResponseHeader):
            self._header = header
        elif isinstance(header, byte_types):
            self._header = ResponseHeader(**Serializers.json.loads(header))
        else:
            raise ValueError(f"header must be of type ResponseHeader or bytes, not {type(self._bytes[INDEX_HEADER])}")

    def parse_body(self) -> None:
        """Parse the body."""
        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,
        message_type: str,
        message_id: str = "",
        payload: SerializableData = SerializableNone,
        preserialized_payload: PreserializedData = PreserializedEmptyByte,
    ) -> "ResponseMessage":
        """
        Crafts an arbitrary response to the client using the method's arguments.

        Parameters
        ----------
        receiver_id: str
            id of the client (ZMQ socket identity)
        sender_id: str
            id of the server (ZMQ socket identity)
        message_type: str
            type of the message, possible values are 'REPLY', 'HANDSHAKE' and 'TIMEOUT'
        message_id: str
            message id of the original client message for which the response is being crafted
        payload: SerializableData
            response payload to send to the client
        preserialized_payload: bytes
            pre-encoded data, generally used for large or custom data that is already serialized

        Returns
        -------
        ResponseMessage
            the crafted response
        """
        message = ResponseMessage([])
        message._header = ResponseHeader(
            messageType=message_type,
            messageID=message_id,
            receiverID=receiver_id,
            senderID=sender_id,
            payloadContentType=payload.content_type,
            preencodedPayloadContentType=preserialized_payload.content_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_reply_from_request(
        cls,
        request_message: RequestMessage,
        payload: SerializableData = SerializableNone,
        preserialized_payload: PreserializedData = PreserializedEmptyByte,
    ) -> "ResponseMessage":
        """
        Craft a response with certain data extracted from an originating client message, like the client's address, message id etc.

        Parameters
        ----------
        request_message: RequestMessage
            The message originated by the client for which the response is being crafted
        payload: SerializableData
            response payload to send to the client
        preserialized_payload: PreserializedData
            pre-encoded data, generally used for large or custom data that is already serialized

        Returns
        -------
        ResponseMessage
            the crafted response
        """
        message = ResponseMessage([])
        message._header = ResponseHeader(
            messageType=REPLY,
            messageID=request_message.id,
            receiverID=request_message.sender_id,
            senderID=request_message.receiver_id,
            payloadContentType=payload.content_type,
            preencodedPayloadContentType=preserialized_payload.content_type,
        )
        message._body = [payload, preserialized_payload]
        message._bytes = [
            bytes(request_message.sender_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,
        request_message: RequestMessage,
        message_type: str,
        payload: SerializableData = SerializableNone,
        preserialized_payload: PreserializedData = PreserializedEmptyByte,
    ) -> "ResponseMessage":
        """
        Create a plain message with a certain type, for example a handshake message.

        Parameters
        ----------
        request_message: RequestMessage
            The message originated by the client for which the response is being crafted
        message_type: str
            message type to be sent
        payload: SerializableData
            response payload to send to the client
        preserialized_payload: PreserializedData
            pre-encoded data, generally used for large or custom data that is already serialized

        Returns
        -------
        ResponseMessage
            the crafted response
        """
        message = ResponseMessage([])
        message._header = ResponseHeader(
            messageType=message_type,
            messageID=request_message.id,
            receiverID=request_message.sender_id,
            senderID=request_message.receiver_id,
            payloadContentType=payload.content_type,
            preencodedPayloadContentType=preserialized_payload.content_type,
        )
        message._body = [payload, preserialized_payload]
        message._bytes = [
            bytes(request_message.sender_id, encoding="utf-8"),
            b"",
            Serializers.json.dumps(message._header.json()),
            payload.serialize(),
            preserialized_payload.value,
        ]
        return message

    def __str__(self) -> str:
        return f"ResponseMessage(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.

payload property

payload: SerializableData

Payload of the message.

preserialized_payload property

preserialized_payload: PreserializedData

pre-encoded payload of the message.

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(receiver_id: str, sender_id: str, message_type: str, message_id: str = '', payload: SerializableData = SerializableNone, preserialized_payload: PreserializedData = PreserializedEmptyByte) -> ResponseMessage

Crafts an arbitrary response to the client using the method's arguments.

Parameters:

Name	Type	Description	Default
receiver_id	str	id of the client (ZMQ socket identity)	required
sender_id	str	id of the server (ZMQ socket identity)	required
message_type	str	type of the message, possible values are 'REPLY', 'HANDSHAKE' and 'TIMEOUT'	required
message_id	str	message id of the original client message for which the response is being crafted	''
payload	SerializableData	response 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
ResponseMessage	the crafted response

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

@classmethod
def craft_from_arguments(
    cls,
    receiver_id: str,
    sender_id: str,
    message_type: str,
    message_id: str = "",
    payload: SerializableData = SerializableNone,
    preserialized_payload: PreserializedData = PreserializedEmptyByte,
) -> "ResponseMessage":
    """
    Crafts an arbitrary response to the client using the method's arguments.

    Parameters
    ----------
    receiver_id: str
        id of the client (ZMQ socket identity)
    sender_id: str
        id of the server (ZMQ socket identity)
    message_type: str
        type of the message, possible values are 'REPLY', 'HANDSHAKE' and 'TIMEOUT'
    message_id: str
        message id of the original client message for which the response is being crafted
    payload: SerializableData
        response payload to send to the client
    preserialized_payload: bytes
        pre-encoded data, generally used for large or custom data that is already serialized

    Returns
    -------
    ResponseMessage
        the crafted response
    """
    message = ResponseMessage([])
    message._header = ResponseHeader(
        messageType=message_type,
        messageID=message_id,
        receiverID=receiver_id,
        senderID=sender_id,
        payloadContentType=payload.content_type,
        preencodedPayloadContentType=preserialized_payload.content_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_reply_from_request classmethod

craft_reply_from_request(request_message: RequestMessage, payload: SerializableData = SerializableNone, preserialized_payload: PreserializedData = PreserializedEmptyByte) -> ResponseMessage

Craft a response with certain data extracted from an originating client message, like the client's address, message id etc.

Parameters:

Name	Type	Description	Default
request_message	RequestMessage	The message originated by the client for which the response is being crafted	required
payload	SerializableData	response 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
ResponseMessage	the crafted response

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

@classmethod
def craft_reply_from_request(
    cls,
    request_message: RequestMessage,
    payload: SerializableData = SerializableNone,
    preserialized_payload: PreserializedData = PreserializedEmptyByte,
) -> "ResponseMessage":
    """
    Craft a response with certain data extracted from an originating client message, like the client's address, message id etc.

    Parameters
    ----------
    request_message: RequestMessage
        The message originated by the client for which the response is being crafted
    payload: SerializableData
        response payload to send to the client
    preserialized_payload: PreserializedData
        pre-encoded data, generally used for large or custom data that is already serialized

    Returns
    -------
    ResponseMessage
        the crafted response
    """
    message = ResponseMessage([])
    message._header = ResponseHeader(
        messageType=REPLY,
        messageID=request_message.id,
        receiverID=request_message.sender_id,
        senderID=request_message.receiver_id,
        payloadContentType=payload.content_type,
        preencodedPayloadContentType=preserialized_payload.content_type,
    )
    message._body = [payload, preserialized_payload]
    message._bytes = [
        bytes(request_message.sender_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(request_message: RequestMessage, message_type: str, payload: SerializableData = SerializableNone, preserialized_payload: PreserializedData = PreserializedEmptyByte) -> ResponseMessage

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

Parameters:

Name	Type	Description	Default
request_message	RequestMessage	The message originated by the client for which the response is being crafted	required
message_type	str	message type to be sent	required
payload	SerializableData	response 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
ResponseMessage	the crafted response

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

@classmethod
def craft_with_message_type(
    cls,
    request_message: RequestMessage,
    message_type: str,
    payload: SerializableData = SerializableNone,
    preserialized_payload: PreserializedData = PreserializedEmptyByte,
) -> "ResponseMessage":
    """
    Create a plain message with a certain type, for example a handshake message.

    Parameters
    ----------
    request_message: RequestMessage
        The message originated by the client for which the response is being crafted
    message_type: str
        message type to be sent
    payload: SerializableData
        response payload to send to the client
    preserialized_payload: PreserializedData
        pre-encoded data, generally used for large or custom data that is already serialized

    Returns
    -------
    ResponseMessage
        the crafted response
    """
    message = ResponseMessage([])
    message._header = ResponseHeader(
        messageType=message_type,
        messageID=request_message.id,
        receiverID=request_message.sender_id,
        senderID=request_message.receiver_id,
        payloadContentType=payload.content_type,
        preencodedPayloadContentType=preserialized_payload.content_type,
    )
    message._body = [payload, preserialized_payload]
    message._bytes = [
        bytes(request_message.sender_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, ResponseHeader):
        self._header = header
    elif isinstance(header, byte_types):
        self._header = ResponseHeader(**Serializers.json.loads(header))
    else:
        raise ValueError(f"header must be of type ResponseHeader or bytes, not {type(self._bytes[INDEX_HEADER])}")

parse_body

parse_body() -> None

Parse the body.

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

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