ZMQServerPool

hololinked.server.zmq.brokers.ZMQServerPool

Bases: BaseZMQServer

Implements pool of async ZMQ servers (& their sockets).

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

class ZMQServerPool(BaseZMQServer):
    """Implements pool of async ZMQ servers (& their sockets)."""

    def __init__(self, *, ids: list[str] | None = None, **kwargs) -> None:
        """
        Initialize the server pool.

        Parameters
        ----------
        ids: List[str], optional
            list of server IDs to create the server pool. If None, an empty pool is created and servers can be
            registered later using `register_server()`.
        kwargs: dict
            Additional arguments for `BaseZMQ` and `BaseAsyncZMQ` classes.

            - `logger`: logger instance to use. If None, a default logger is created.
        """
        self.context = global_config.zmq_context()
        self.poller = zmq.asyncio.Poller()
        self.pool = dict()  # type: dict[str, AsyncZMQServer]
        if ids:
            for id in ids:
                self.pool[id] = AsyncZMQServer(id=id, context=self.context, **kwargs)
            for server in self.pool.values():
                self.poller.register(server.socket, zmq.POLLIN)
        super().__init__(id=f"pool-{uuid_hex()}", **kwargs)

    def create_socket(
        self,
        *,
        id: str,
        bind: bool,
        context: zmq.asyncio.Context | zmq.Context,
        access_point: str,
        socket_type: zmq.SocketType = zmq.SocketType.ROUTER,
        **kwargs,
    ) -> None:
        raise NotImplementedError("create socket not supported by ZMQServerPool")
        # we override this method to prevent socket creation. id set to pool is simply a filler
        # return super().create_socket()

    def register_server(self, server: AsyncZMQServer) -> None:
        if not isinstance(server, (AsyncZMQServer)):
            raise TypeError(f"registration possible only for subclass of AsyncZMQServer. Given type {type(server)}")
        self.pool[server.id] = server
        self.poller.register(server.socket, zmq.POLLIN)

    def deregister_server(self, server: AsyncZMQServer) -> None:
        self.poller.unregister(server.socket)
        self.pool.pop(server.id)

    @property
    def poll_timeout(self) -> int:
        """Socket polling timeout in milliseconds greater than 0."""
        return self._poll_timeout

    @poll_timeout.setter
    def poll_timeout(self, value) -> None:
        if not isinstance(value, int) or value < 0:
            raise ValueError(
                "polling period must be an integer greater than 0, not {}.".format(value)
                + " Value is considered in milliseconds."
            )
        self._poll_timeout = value

    async def async_recv_request(self, id: str) -> RequestMessage:
        """
        Receive message for server specified by id.

        Parameters
        ----------
        id: str
            id of the ZMQ server

        Returns
        -------
        RequestMessage
            received message
        """
        return await self.pool[id].async_recv_request()

    async def async_recv_requests(self, id: str) -> list[RequestMessage]:
        """
        Receive all available messages for server specified by id.

        Parameters
        ----------
        id: str
            id of the ZMQ server

        Returns
        -------
        list[RequestMessage]
            list of received messages
        """
        return await self.pool[id].async_recv_requests()

    async def async_send_response(
        self,
        *,
        id: str,
        request_message: RequestMessage,
        payload: SerializableData = SerializableNone,
        preserialized_payload: PreserializedData = PreserializedEmptyByte,
    ) -> None:
        """
        Send response for a request message for server specified by id.

        Parameters
        ----------
        id: str
            id of the ZMQ server
        request_message: RequestMessage
            request message for which response is being given
        payload: SerializableData
            serializable data to be sent as response
        preserialized_payload: PreserializedData
            pre-encoded data to be sent as payload, generally used for large or custom data that is already serialized
        """
        await self.pool[id].async_send_response(
            request_message=request_message,
            payload=payload,
            preserialized_payload=preserialized_payload,
        )

    async def poll(self) -> list[RequestMessage]:
        """
        Pool for messages in the entire server pool.

        Use the message to identify the server by using `receiver_id` of the message header.

        Returns
        -------
        list[RequestMessage]
            list of received messages across all servers in the pool
        """
        self.stop_poll = False
        messages = []
        while not self.stop_poll:
            sockets = await self.poller.poll(self._poll_timeout)
            for socket, _ in sockets:
                while True:
                    try:
                        raw_message = await socket.recv_multipart(zmq.NOBLOCK)
                    except zmq.Again:
                        break
                    else:
                        if raw_message:
                            request_message = RequestMessage(raw_message)
                            self.logger.debug(
                                "received message from client",
                                sender_id=request_message.sender_id,
                                receiver_id=request_message.receiver_id,
                                msg_id=request_message.id,
                            )
                            messages.append(request_message)
        return messages

    def stop_polling(self) -> None:
        """Stop polling method `poll()`."""
        self.stop_poll = True

    def __getitem__(self, key) -> AsyncZMQServer:
        return self.pool[key]

    def __iter__(self) -> Iterator[str]:
        return self.pool.__iter__()

    def __contains__(self, name: str) -> bool:
        return name in self.pool.keys()

    def exit(self) -> None:
        for server in self.pool.values():
            try:
                self.poller.unregister(server.socket)
            except Exception as ex:
                self.logger.warning(f"could not unregister poller - {str(ex)}")
            server.exit()

Attributes

id instance-attribute

id: str = id

context instance-attribute

context = zmq_context()

poller instance-attribute

poller = Poller()

pool instance-attribute

pool = dict()

logger instance-attribute

logger = bind(component='broker', impl=__name__, id=id)

poll_timeout property writable

poll_timeout: int

Socket polling timeout in milliseconds greater than 0.

Functions

__init__

__init__(*, ids: list[str] | None = None, **kwargs) -> None

Initialize the server pool.

Parameters:

Name	Type	Description	Default
ids	list[str] | None	list of server IDs to create the server pool. If None, an empty pool is created and servers can be registered later using register_server().	None
kwargs		Additional arguments for BaseZMQ and BaseAsyncZMQ classes. logger: logger instance to use. If None, a default logger is created.	{}

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

def __init__(self, *, ids: list[str] | None = None, **kwargs) -> None:
    """
    Initialize the server pool.

    Parameters
    ----------
    ids: List[str], optional
        list of server IDs to create the server pool. If None, an empty pool is created and servers can be
        registered later using `register_server()`.
    kwargs: dict
        Additional arguments for `BaseZMQ` and `BaseAsyncZMQ` classes.

        - `logger`: logger instance to use. If None, a default logger is created.
    """
    self.context = global_config.zmq_context()
    self.poller = zmq.asyncio.Poller()
    self.pool = dict()  # type: dict[str, AsyncZMQServer]
    if ids:
        for id in ids:
            self.pool[id] = AsyncZMQServer(id=id, context=self.context, **kwargs)
        for server in self.pool.values():
            self.poller.register(server.socket, zmq.POLLIN)
    super().__init__(id=f"pool-{uuid_hex()}", **kwargs)

create_socket

create_socket(*, id: str, bind: bool, context: Context | Context, access_point: str, socket_type: SocketType = zmq.SocketType.ROUTER, **kwargs) -> None

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

def create_socket(
    self,
    *,
    id: str,
    bind: bool,
    context: zmq.asyncio.Context | zmq.Context,
    access_point: str,
    socket_type: zmq.SocketType = zmq.SocketType.ROUTER,
    **kwargs,
) -> None:
    raise NotImplementedError("create socket not supported by ZMQServerPool")

register_server

register_server(server: AsyncZMQServer) -> None

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

def register_server(self, server: AsyncZMQServer) -> None:
    if not isinstance(server, (AsyncZMQServer)):
        raise TypeError(f"registration possible only for subclass of AsyncZMQServer. Given type {type(server)}")
    self.pool[server.id] = server
    self.poller.register(server.socket, zmq.POLLIN)

deregister_server

deregister_server(server: AsyncZMQServer) -> None

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

def deregister_server(self, server: AsyncZMQServer) -> None:
    self.poller.unregister(server.socket)
    self.pool.pop(server.id)

poll async

poll() -> list[RequestMessage]

Pool for messages in the entire server pool.

Use the message to identify the server by using receiver_id of the message header.

Returns:

Type	Description
list[RequestMessage]	list of received messages across all servers in the pool

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

async def poll(self) -> list[RequestMessage]:
    """
    Pool for messages in the entire server pool.

    Use the message to identify the server by using `receiver_id` of the message header.

    Returns
    -------
    list[RequestMessage]
        list of received messages across all servers in the pool
    """
    self.stop_poll = False
    messages = []
    while not self.stop_poll:
        sockets = await self.poller.poll(self._poll_timeout)
        for socket, _ in sockets:
            while True:
                try:
                    raw_message = await socket.recv_multipart(zmq.NOBLOCK)
                except zmq.Again:
                    break
                else:
                    if raw_message:
                        request_message = RequestMessage(raw_message)
                        self.logger.debug(
                            "received message from client",
                            sender_id=request_message.sender_id,
                            receiver_id=request_message.receiver_id,
                            msg_id=request_message.id,
                        )
                        messages.append(request_message)
    return messages

async_recv_request async

async_recv_request(id: str) -> RequestMessage

Receive message for server specified by id.

Parameters:

Name	Type	Description	Default
id	str	id of the ZMQ server	required

Returns:

Type	Description
RequestMessage	received message

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

async def async_recv_request(self, id: str) -> RequestMessage:
    """
    Receive message for server specified by id.

    Parameters
    ----------
    id: str
        id of the ZMQ server

    Returns
    -------
    RequestMessage
        received message
    """
    return await self.pool[id].async_recv_request()

async_recv_requests async

async_recv_requests(id: str) -> list[RequestMessage]

Receive all available messages for server specified by id.

Parameters:

Name	Type	Description	Default
id	str	id of the ZMQ server	required

Returns:

Type	Description
list[RequestMessage]	list of received messages

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

async def async_recv_requests(self, id: str) -> list[RequestMessage]:
    """
    Receive all available messages for server specified by id.

    Parameters
    ----------
    id: str
        id of the ZMQ server

    Returns
    -------
    list[RequestMessage]
        list of received messages
    """
    return await self.pool[id].async_recv_requests()

async_send_response async

async_send_response(*, id: str, request_message: RequestMessage, payload: SerializableData = SerializableNone, preserialized_payload: PreserializedData = PreserializedEmptyByte) -> None

Send response for a request message for server specified by id.

Parameters:

Name	Type	Description	Default
id	str	id of the ZMQ server	required
request_message	RequestMessage	request message for which response is being given	required
payload	SerializableData	serializable data to be sent as response	SerializableNone
preserialized_payload	PreserializedData	pre-encoded data to be sent as payload, generally used for large or custom data that is already serialized	PreserializedEmptyByte

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

async def async_send_response(
    self,
    *,
    id: str,
    request_message: RequestMessage,
    payload: SerializableData = SerializableNone,
    preserialized_payload: PreserializedData = PreserializedEmptyByte,
) -> None:
    """
    Send response for a request message for server specified by id.

    Parameters
    ----------
    id: str
        id of the ZMQ server
    request_message: RequestMessage
        request message for which response is being given
    payload: SerializableData
        serializable data to be sent as response
    preserialized_payload: PreserializedData
        pre-encoded data to be sent as payload, generally used for large or custom data that is already serialized
    """
    await self.pool[id].async_send_response(
        request_message=request_message,
        payload=payload,
        preserialized_payload=preserialized_payload,
    )

stop_polling

stop_polling() -> None

Stop polling method poll().

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

def stop_polling(self) -> None:
    """Stop polling method `poll()`."""
    self.stop_poll = True

exit

exit() -> None

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

def exit(self) -> None:
    for server in self.pool.values():
        try:
            self.poller.unregister(server.socket)
        except Exception as ex:
            self.logger.warning(f"could not unregister poller - {str(ex)}")
        server.exit()

handshake

handshake(request_message: RequestMessage) -> None

Pass a handshake message to client.

Absolutely mandatory to handshake with all clients to ensure initial messages do not get lost because of ZMQ's tiny but significant initial delay after creating socket.

Parameters:

Name	Type	Description	Default
request_message	RequestMessage	the client message for which the handshake is being sent	required

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

def handshake(self, request_message: RequestMessage) -> None:
    """
    Pass a handshake message to client.

    Absolutely mandatory to handshake with all clients to ensure initial messages do not get lost because of
    ZMQ's tiny but significant initial delay after creating socket.

    Parameters
    ----------
    request_message: RequestMessage
        the client message for which the handshake is being sent
    """
    run_callable_somehow(self._handshake(request_message))

handle_invalid_message

handle_invalid_message(request_message: RequestMessage, exception: Exception) -> None

Pass an invalid message to the client when an exception occurred while parsing the message from the client (in handled_default_message_types()).

Parameters:

Name	Type	Description	Default
request_message	RequestMessage	the client message for which the parsing exception occurred	required
exception	Exception	exception object raised	required

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

def handle_invalid_message(self, request_message: RequestMessage, exception: Exception) -> None:
    """
    Pass an invalid message to the client when an exception occurred while parsing the message from the client (in `handled_default_message_types()`).

    Parameters
    ----------
    request_message: RequestMessage
        the client message for which the parsing exception occurred
    exception: Exception
        exception object raised
    """
    run_callable_somehow(self._handle_invalid_message(request_message, exception))

handle_timeout

handle_timeout(request_message: RequestMessage, timeout_type: str) -> None

Pass timeout message to the client when the operation could not be executed within specified timeouts.

Parameters:

Name	Type	Description	Default
request_message	RequestMessage	the client message which could not executed within the specified timeout. timeout value is generally specified within the execution context values.	required

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

def handle_timeout(self, request_message: RequestMessage, timeout_type: str) -> None:
    """
    Pass timeout message to the client when the operation could not be executed within specified timeouts.

    Parameters
    ----------
    request_message: RequestMessage
        the client message which could not executed within the specified timeout. timeout value is
        generally specified within the execution context values.
    """
    run_callable_somehow(self._handle_timeout(request_message, timeout_type=timeout_type))

handle_error_message

handle_error_message(request_message: RequestMessage, exception: Exception) -> None

Pass an exception message to the client when an exception occurred while executing the operation.

Parameters:

Name	Type	Description	Default
request_message	RequestMessage	the client message for which the exception occurred	required
exception	Exception	exception object raised	required

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

def handle_error_message(self, request_message: RequestMessage, exception: Exception) -> None:
    """
    Pass an exception message to the client when an exception occurred while executing the operation.

    Parameters
    ----------
    request_message: RequestMessage
        the client message for which the exception occurred
    exception: Exception
        exception object raised
    """
    run_callable_somehow(self._handle_error_message(request_message, exception))

handled_default_message_types

handled_default_message_types(request_message: RequestMessage) -> bool

Handle default cases for the server without further processing of the request (for example, HANDSHAKE).

This method is called once/supposed to be called when the message is received or popped out of the socket.

Parameters:

Name	Type	Description	Default
request_message	RequestMessage	the client message to handle	required

Returns:

Name	Type	Description
handled	bool	whether the message was handled here and needs no further processing

Raises:

Type	Description
BreakLoop	if an EXIT message is received from the client

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

def handled_default_message_types(self, request_message: RequestMessage) -> bool:
    """
    Handle default cases for the server without further processing of the request (for example, `HANDSHAKE`).

    This method is called once/supposed to be called when the message is received or popped out of the socket.

    Parameters
    ----------
    request_message: RequestMessage
        the client message to handle

    Returns
    -------
    handled: bool
        whether the message was handled here and needs no further processing

    Raises
    ------
    BreakLoop
        if an `EXIT` message is received from the client
    """
    if request_message.type == HANDSHAKE:
        self.handshake(request_message)
        return True
    elif request_message.type == EXIT:
        # self.send response with message type EXIT
        raise BreakLoop(f"exit message received from {request_message.sender_id} with msg-ID {request_message.id}")
    elif request_message.length != len(request_message.byte_array):
        self.handle_invalid_message(
            request_message,
            exception=ValueError(
                f"message length {request_message.length} does not match "
                + f"the number of message parts {len(request_message.byte_array)}"
            ),
        )
        return True
    return False