hololinked.core.zmq.brokers.MessageMappedZMQClientPool

Bases: BaseZMQClient

Pool of async ZMQ clients, to be primarily used within protocol bindings where multiple things may be served. Use message ID to track responses.

Source code in hololinked/hololinked/core/zmq/brokers.py

class MessageMappedZMQClientPool(BaseZMQClient):
    """
    Pool of async ZMQ clients, to be primarily used within protocol bindings where multiple things may be served.
    Use message ID to track responses.
    """

    def __init__(
        self,
        id: str,
        client_ids: list[str],
        server_ids: list[str],
        handshake: bool = True,
        context: zmq.asyncio.Context = None,
        access_point: str = ZMQ_TRANSPORTS.IPC,
        poll_timeout: int = 25,
        **kwargs,
    ) -> None:
        """
        Parameters
        ----------
        id: str
            ID of the pool, must be unique.
        client_ids: List[str]
            list of id's of clients in the pool, must be unique. Clients are created with these ID's.
            For pre-existing clients, use `register` method instead and leave this list empty.
        server_ids: List[str]
            list of id's of servers to connect to, must be same length as client_ids. For pre-existing clients that
            are connected to servers, use `register` method instead and leave this list empty.
        handshake: bool
            when true, handshake with the server first before allowing first message and block until that handshake was
            accomplished.
        context: zmq.asyncio.Context
            ZMQ context to use, if None, a global context is used.
        access_point: Enum | str, default ZMQ_TRANSPORTS.IPC
            Default access point for all clients in the pool, usually not helpful except `INPROC` transport.
            Use `TCP` or `tcp://<host>:<port>` for network access, `IPC` for multi-process applications,
            and `INPROC` for multi-threaded applications.
        poll_timeout: int
            socket polling timeout in milliseconds greater than 0.
        **kwargs:
            Additional arguments:

            - `logger`: `logging.Logger`, logger instance to use. If None, a default logger is created.
        """
        super().__init__(id=id, server_id=None, **kwargs)
        if len(client_ids) != len(server_ids):
            raise ValueError("client_ids and server_ids must have same length")
        # this class does not call create_socket method
        self.context = context or global_config.zmq_context()
        self.pool = dict()  # type: dict[str, AsyncZMQClient]
        self.poller = zmq.asyncio.Poller()
        for client_id, server_id in zip(client_ids, server_ids):
            client = AsyncZMQClient(
                id=client_id,
                server_id=server_id,
                handshake=handshake,
                context=self.context,
                access_point=access_point,
                logger=self.logger,
            )
            self.register(client)
        # Both the client pool as well as the individual client get their serializers and client_types
        # This is required to implement pool level sending and receiving messages like polling of pool of sockets
        self.event_pool = AsyncioEventPool(len(server_ids))
        self.events_map = dict()  # type: dict[bytes, asyncio.Event]
        self.message_map = dict()
        self.cancelled_messages = []
        self.poll_timeout = poll_timeout
        self.stop_poll = False
        self._thing_to_client_map = dict()  # type: dict[str, AsyncZMQClient]
        self._client_to_thing_map = dict()  # type: dict[str, str]

    def create_new(self, id: str, server_id: str, access_point: str = ZMQ_TRANSPORTS.IPC) -> None:
        """
        Create new server with specified transport & add it to the pool.
        Other arguments are taken from pool default specifications.

        Parameters
        ----------
        id: str
            id of the new client to be created
        server_id: str
            id of the server to connect to
        access_point: str
            transport method used by the server - `IPC`, `INPROC` or `tcp://<host>:<port>`
        """
        if server_id not in self.pool.keys():
            client = AsyncZMQClient(
                id=id,
                server_id=server_id,
                handshake=True,
                context=self.context,
                access_point=access_point,
                logger=self.logger,
            )
            client._monitor_socket = client.socket.get_monitor_socket()
            self.poller.register(client._monitor_socket, zmq.POLLIN)
            self.pool[id] = client
        else:
            raise ValueError(f"client for instance name '{server_id}' already present in pool")

    def register(self, client: AsyncZMQClient, thing_id: str | None = None) -> None:
        """
        Register a pre-existing client with the pool.

        Parameters
        ----------
        client: AsyncZMQClient
            client to be registered
        thing_id: Optional, str
            thing_id to which this client is mapped, especially when the client is connected to a server that serves
            only one `Thing`.
        """
        if not isinstance(client, AsyncZMQClient):
            raise TypeError(
                "registration possible for clients only subclass of AsyncZMQClient." + f" Given type {type(client)}"
            )
        if client.id not in self.pool:
            self.pool[client.id] = client
            self.poller.register(client.socket, zmq.POLLIN)
            self.poller.register(client._monitor_socket, zmq.POLLIN)
        elif self.pool[client.id] != client:
            warnings.warn(
                f"client with id '{client.id}' already present in pool. Replacing with {client}",
                category=UserWarning,
            )
        if thing_id:
            self._thing_to_client_map[thing_id] = client.id
            self._client_to_thing_map[client.id] = thing_id

    def get_client_id_from_thing_id(self, thing_id: str) -> str:
        """
        Retrieve client mapped to a `thing_id`. The value must have been previously set using `register` method.

        Parameters
        ----------
        thing_id: str
            the `thing_id` for which the client is to be retrieved
        """
        if thing_id not in self._thing_to_client_map:
            raise ValueError(f"client for thing_id '{thing_id}' not present in pool")
        return self._thing_to_client_map.get(thing_id, None)

    def get_thing_id_from_client_id(self, client_id: str) -> str:
        """
        Retrieve `thing_id` mapped to a client. The value must have been previously set using `register` method.

        Parameters
        ----------
        client_id: str
            the client id for which the `thing_id` is to be retrieved
        """
        if client_id not in self._client_to_thing_map:
            raise ValueError(f"thing_id for client_id '{client_id}' not present in pool")
        return self._client_to_thing_map.get(client_id, None)

    def get_client_protocol(self, thing_id: str) -> str:
        """
        Retrieve protocol used by a client in the pool.

        Parameters
        ----------
        client_id: str
            the client id for which the protocol is to be retrieved
        """
        if thing_id not in self._thing_to_client_map:
            raise ValueError(f"client_id '{thing_id}' not present in pool")
        client_id = self._thing_to_client_map[thing_id]
        return self.pool[client_id].socket_address.split("://")[0].upper()

    @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 {}. Value is considered in milliseconds".format(
                    value
                )
            )
        self._poll_timeout = value

    async def handshake_complete(self) -> None:
        """wait for handshake to complete for all clients in the pool"""
        for client in self.pool.values():
            await client.handshake_complete()  # sufficient to wait serially

    def handshake(self, timeout: int | None = 60000) -> None:
        """
        schedules handshake coroutines for each client in the running event loop
        or completes handshake synchronously if no event loop is running.
        Use `handshake_complete()` async method to check if handshake is complete.

        Parameters
        ----------
        timeout: float | int
            timeout in milliseconds to wait for handshake to complete. If handshake does not complete within this time,
            a `ConnectionError` is raised. If None, wait indefinitely until handshake completes.
        """
        for client in self.pool.values():
            client.handshake(timeout)

    async def poll_responses(self) -> None:
        """
        Poll for replies from server.This method should be independently started in the event loop by calling `start_polling()`.
        Sending message requests and retrieving a response is still carried out by other methods.
        Do not duplicate this method call as there are no checks how many pollers exist and messages will become malformed
        if multiple pollers are active.
        """
        self.logger.info("client polling started for sockets for {}".format(list(self.pool.keys())))
        self.stop_poll = False
        event_loop = asyncio.get_event_loop()
        while not self.stop_poll:
            sockets = await self.poller.poll(self.poll_timeout)  # type hints dont work in this line
            for socket, _ in sockets:
                while True:
                    try:
                        raw_response = await socket.recv_multipart(zmq.NOBLOCK)
                        response_message = ResponseMessage(raw_response)
                    except zmq.Again:
                        # errors in handle_message should reach the client.
                        break
                    except ConnectionAbortedError:
                        for client in self.pool.values():
                            if client.socket.get_monitor_socket() == socket:
                                self.poller.unregister(client.socket)  # leave the monitor in the pool
                                client.handshake(timeout=None)
                                self.logger.error(
                                    f"client {client.id} disconnected from server."
                                    + " Unregistering from poller temporarily until server comes back."
                                )
                                break
                    else:
                        if self.handled_default_message_types(response_message):
                            continue
                        message_id = response_message.id
                        self.logger.debug(
                            "received response from server",
                            sender_id=response_message.sender_id,
                            receiver_id=response_message.receiver_id,
                            msg_id=response_message.id,
                            message_type=response_message.type,
                        )
                        if message_id in self.cancelled_messages:
                            self.cancelled_messages.remove(message_id)
                            self.logger.debug("dropping a cancelled message", msg_id=message_id)
                            continue
                        self.message_map[message_id] = response_message
                        event = self.events_map.get(message_id, None)
                        if event:
                            event.set()
                        else:
                            event_loop.create_task(self._resolve_response(message_id, response_message))

    async def _resolve_response(self, message_id: str, data: Any) -> None:
        """
        This method is called when there is no asyncio Event available for a message ID. This can happen only
        when the server replied before the client created a asyncio.Event object. check `async_execute()` for details.

        Parameters
        ----------
        message_id: bytes
            the message for which the event was not created
        data: ResponseMessage
            the response message received from server
        """
        max_number_of_retries = 100
        for i in range(max_number_of_retries):
            await asyncio.sleep(0.025)
            try:
                event = self.events_map[message_id]
            except KeyError:
                if message_id in self.cancelled_messages:
                    # Only for safety, likely should never reach here
                    self.cancelled_messages.remove(message_id)
                    self.logger.debug("message cancelled, not retrieving response", msg_id=message_id)
                    return
                if i >= max_number_of_retries - 1:
                    self.logger.error("unknown message id without corresponding event object", msg_id=message_id)
                    return
            else:
                self.message_map[message_id] = data
                event.set()
                break

    def assert_client_ready(self, client: AsyncZMQClient):
        if not client._handshake_event.is_set():
            raise ConnectionAbortedError(f"{client.id} is currently not alive")
        if client.socket not in self.poller._map:
            raise ConnectionError(
                "handshake complete, server is alive but client socket not yet ready to be polled."
                + "Application using MessageMappedClientPool should register the socket manually for polling."
                + "If using hololinked.server.HTTPServer, socket is waiting until HTTP Server updates its "
                "routing logic as the server has just now come alive, please try again soon."
            )

    async def async_send_request(
        self,
        thing_id: str,
        objekt: str,
        operation: str,
        payload: SerializableData = SerializableNone,
        preserialized_payload: PreserializedData = PreserializedEmptyByte,
        server_execution_context: ServerExecutionContext = default_server_execution_context,
        thing_execution_context: ThingExecutionContext = default_thing_execution_context,
    ) -> str:
        """
        send request message to server.

        Parameters
        ----------
        client_id: str
            id of the client in the pool to be used to send the message
        thing_id: str
            `id` of the `Thing` on which an operation is to be performed
        objekt: str
            name of property, action or event (usually only property or action)
        operation: str
            operation to be performed, like `readproperty`, `writeproperty`, `invokeaction` etc.
        payload: SerializableData
            serializable data to be sent as payload
        preserialized_payload: PreserializedData
            pre-encoded data to be sent as payload, generally used for large or custom data that is already serialized
        server_execution_context: dict[str, Any]
            Specify server level execution context like `invokationTimeout`, `executionTimeout`, `oneway` operation etc.
        thing_execution_context: dict[str, Any]
            Specify thing level execution context like `fetchExecutionLogs` etc.

        Returns
        -------
        bytes
            a message id in bytes
        """
        client_id = self.get_client_id_from_thing_id(thing_id)
        self.assert_client_ready(self.pool[client_id])
        message_id = await self.pool[client_id].async_send_request(
            thing_id=thing_id,
            objekt=objekt,
            operation=operation,
            payload=payload,
            preserialized_payload=preserialized_payload,
            server_execution_context=server_execution_context,
            thing_execution_context=thing_execution_context,
        )
        event = self.event_pool.pop()
        self.events_map[message_id] = event
        return message_id

    async def async_recv_response(
        self, thing_id: str, message_id: bytes, timeout: float | int | None = None
    ) -> ResponseMessage:
        """
        Receive response for specified message ID.

        Parameters
        ----------
        client_id: str
            id of the client in the pool to be used to receive the message
        message_id: bytes
            the message id for which response needs to be fetched
        timeout: float | int | None
            Client side timeout, not the same as timeout passed to server. Recommended to be None in general cases.
            The reply is dropped if this timeout occurs. Usually the server always replies when either of invokation or
            execution timeout occurs. This timeout ensures that the protocol binding does not wait indefinitely for a message
            that may never arrive.

        Returns
        -------
        response: ResponseMessage
            response message from server corresponding to the message id

        Raises
        ------
        ValueError
            if supplied message id is not valid
        TimeoutError
            if timeout is not None and response did not arrive
        """
        try:
            client_id = self.get_client_id_from_thing_id(thing_id)
            event = self.events_map[message_id]
        except KeyError:
            raise KeyError(f"message id {message_id} unknown.") from None
        while True:
            try:
                await asyncio.wait_for(event.wait(), timeout)
                # default 5 seconds because we want to check if server is also dead
                if event.is_set():  # i.e. if timeout is not None, check if event is set
                    break
                self.assert_client_ready(self.pool[client_id])
            except TimeoutError:
                self.cancelled_messages.append(message_id)
                self.logger.error("message added to list of cancelled messages", msg_id=message_id)
                raise TimeoutError(f"Execution not completed within {timeout} seconds") from None
        self.events_map.pop(message_id)
        self.event_pool.completed(event)
        response = self.message_map.pop(message_id)
        return response

    async def async_execute(
        self,
        thing_id: str,
        objekt: str,
        operation: str,
        payload: SerializableData = SerializableNone,
        preserialized_payload: PreserializedData = PreserializedEmptyByte,
        server_execution_context: ServerExecutionContext = default_server_execution_context,
        thing_execution_context: ThingExecutionContext = default_thing_execution_context,
    ) -> ResponseMessage:
        """
        send an operation and receive the response for it.

        Parameters
        ----------
        client_id: str
            id of the client in the pool to be used to send the message
        thing_id: str
            `id` of the `Thing` on which an operation is to be performed
        objekt: str
            name of property, action or event (usually only property or action)
        operation: str
            operation to be performed, like `readproperty`, `writeproperty`, `invokeaction` etc.
        payload: SerializableData
            serializable data to be sent as payload
        preserialized_payload: PreserializedData
            pre-encoded data to be sent as payload, generally used for large or custom data that is already serialized
        server_execution_context: dict[str, Any]
            Specify server level execution context like `invokationTimeout`, `executionTimeout`, `oneway` operation etc.
        thing_execution_context: dict[str, Any]
            Specify thing level execution context like `fetchExecutionLogs` etc.

        Returns
        -------
        ResponseMessage
            response message from server after completing the operation
        """
        message_id = await self.async_send_request(
            thing_id=thing_id,
            objekt=objekt,
            operation=operation,
            payload=payload,
            preserialized_payload=preserialized_payload,
            server_execution_context=server_execution_context,
            thing_execution_context=thing_execution_context,
        )
        return await self.async_recv_response(
            thing_id=thing_id,
            message_id=message_id,
        )

    def start_polling(self) -> None:
        """register the server message polling loop in the asyncio event loop"""
        get_current_async_loop().create_task(self.poll_responses())

    def stop_polling(self):
        """
        stop polling for replies from server
        """
        self.stop_poll = True
        for client in self.pool.values():
            client.stop()

    async def async_execute_in_all(
        self,
        objekt: str,
        operation: str,
        payload: SerializableData = SerializableNone,
        preserialized_payload: PreserializedData = PreserializedEmptyByte,
        thing_ids: list[str] | None = None,
        server_execution_context: ServerExecutionContext = default_server_execution_context,
        thing_execution_context: ThingExecutionContext = default_thing_execution_context,
    ) -> dict[str, ResponseMessage]:
        if not thing_ids:
            thing_ids = self._client_to_thing_map.values()

        gathered_replies = await asyncio.gather(
            *[
                self.async_execute(
                    thing_id=id,
                    objekt=objekt,
                    operation=operation,
                    payload=payload,
                    preserialized_payload=preserialized_payload,
                    server_execution_context=server_execution_context,
                    thing_execution_context=thing_execution_context,
                )
                for id in thing_ids
            ]
        )
        replies = dict()
        for id, response in zip(thing_ids, gathered_replies):
            replies[id] = response
        return replies

    async def async_execute_in_all_things(
        self,
        objekt: str,
        operation: str,
        payload: SerializableData = SerializableNone,
        preserialized_payload: PreserializedData = PreserializedEmptyByte,
        server_execution_context: ServerExecutionContext = default_server_execution_context,
        thing_execution_context: ThingExecutionContext = default_thing_execution_context,
    ) -> dict[str, ResponseMessage]:
        """execute the same operation in all `Thing`s"""
        return await self.async_execute_in_all(
            objekt=objekt,
            operation=operation,
            payload=payload,
            preserialized_payload=preserialized_payload,
            server_execution_context=server_execution_context,
            thing_execution_context=thing_execution_context,
        )

    async def ping_all_servers(self):
        return await self.async_execute_in_all()  # operation='invokeAction', objekt=CommonRPC.PING)

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

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

    def __iter__(self) -> Iterator[AsyncZMQClient]:
        return iter(self.pool.values())

    def exit(self) -> None:
        try:
            BaseZMQ.exit(self)
            for client in self.pool.values():
                self.poller.unregister(client.socket)
                self.poller.unregister(client.socket.get_monitor_socket())
                client.exit()
            self.logger.info("all client socket unregistered")
        except Exception as ex:
            self.logger.warning(
                "could not properly terminate context or attempted to terminate an already terminated context."
                + f" Exception message: {str(ex)}"
            )

    """
    Inheritance tree:

    BaseZMQ
    BaseAsyncZMQ
    BaseSyncZMQ
    BaseZMQClient
    SyncZMQClient
    AsyncZMQClient
    MessageMappedClientPool
    """

Functions

__init__

__init__(id: str, client_ids: list[str], server_ids: list[str], handshake: bool = True, context: Context = None, access_point: str = ZMQ_TRANSPORTS.IPC, poll_timeout: int = 25, **kwargs) -> None

Parameters:

Name	Type	Description	Default
id	str	ID of the pool, must be unique.	required
client_ids	list[str]	list of id's of clients in the pool, must be unique. Clients are created with these ID's. For pre-existing clients, use register method instead and leave this list empty.	required
server_ids	list[str]	list of id's of servers to connect to, must be same length as client_ids. For pre-existing clients that are connected to servers, use register method instead and leave this list empty.	required
handshake	bool	when true, handshake with the server first before allowing first message and block until that handshake was accomplished.	True
context	Context	ZMQ context to use, if None, a global context is used.	None
access_point	str	Default access point for all clients in the pool, usually not helpful except INPROC transport. Use TCP or tcp://<host>:<port> for network access, IPC for multi-process applications, and INPROC for multi-threaded applications.	IPC
poll_timeout	int	socket polling timeout in milliseconds greater than 0.	25
**kwargs		Additional arguments: logger: logging.Logger, logger instance to use. If None, a default logger is created.	{}

Source code in hololinked/hololinked/core/zmq/brokers.py

def __init__(
    self,
    id: str,
    client_ids: list[str],
    server_ids: list[str],
    handshake: bool = True,
    context: zmq.asyncio.Context = None,
    access_point: str = ZMQ_TRANSPORTS.IPC,
    poll_timeout: int = 25,
    **kwargs,
) -> None:
    """
    Parameters
    ----------
    id: str
        ID of the pool, must be unique.
    client_ids: List[str]
        list of id's of clients in the pool, must be unique. Clients are created with these ID's.
        For pre-existing clients, use `register` method instead and leave this list empty.
    server_ids: List[str]
        list of id's of servers to connect to, must be same length as client_ids. For pre-existing clients that
        are connected to servers, use `register` method instead and leave this list empty.
    handshake: bool
        when true, handshake with the server first before allowing first message and block until that handshake was
        accomplished.
    context: zmq.asyncio.Context
        ZMQ context to use, if None, a global context is used.
    access_point: Enum | str, default ZMQ_TRANSPORTS.IPC
        Default access point for all clients in the pool, usually not helpful except `INPROC` transport.
        Use `TCP` or `tcp://<host>:<port>` for network access, `IPC` for multi-process applications,
        and `INPROC` for multi-threaded applications.
    poll_timeout: int
        socket polling timeout in milliseconds greater than 0.
    **kwargs:
        Additional arguments:

        - `logger`: `logging.Logger`, logger instance to use. If None, a default logger is created.
    """
    super().__init__(id=id, server_id=None, **kwargs)
    if len(client_ids) != len(server_ids):
        raise ValueError("client_ids and server_ids must have same length")
    # this class does not call create_socket method
    self.context = context or global_config.zmq_context()
    self.pool = dict()  # type: dict[str, AsyncZMQClient]
    self.poller = zmq.asyncio.Poller()
    for client_id, server_id in zip(client_ids, server_ids):
        client = AsyncZMQClient(
            id=client_id,
            server_id=server_id,
            handshake=handshake,
            context=self.context,
            access_point=access_point,
            logger=self.logger,
        )
        self.register(client)
    # Both the client pool as well as the individual client get their serializers and client_types
    # This is required to implement pool level sending and receiving messages like polling of pool of sockets
    self.event_pool = AsyncioEventPool(len(server_ids))
    self.events_map = dict()  # type: dict[bytes, asyncio.Event]
    self.message_map = dict()
    self.cancelled_messages = []
    self.poll_timeout = poll_timeout
    self.stop_poll = False
    self._thing_to_client_map = dict()  # type: dict[str, AsyncZMQClient]
    self._client_to_thing_map = dict()  # type: dict[str, str]

register

register(client: AsyncZMQClient, thing_id: str | None = None) -> None

Register a pre-existing client with the pool.

Parameters:

Name	Type	Description	Default
client	AsyncZMQClient	client to be registered	required
thing_id	str | None	thing_id to which this client is mapped, especially when the client is connected to a server that serves only one Thing.	None

Source code in hololinked/hololinked/core/zmq/brokers.py

def register(self, client: AsyncZMQClient, thing_id: str | None = None) -> None:
    """
    Register a pre-existing client with the pool.

    Parameters
    ----------
    client: AsyncZMQClient
        client to be registered
    thing_id: Optional, str
        thing_id to which this client is mapped, especially when the client is connected to a server that serves
        only one `Thing`.
    """
    if not isinstance(client, AsyncZMQClient):
        raise TypeError(
            "registration possible for clients only subclass of AsyncZMQClient." + f" Given type {type(client)}"
        )
    if client.id not in self.pool:
        self.pool[client.id] = client
        self.poller.register(client.socket, zmq.POLLIN)
        self.poller.register(client._monitor_socket, zmq.POLLIN)
    elif self.pool[client.id] != client:
        warnings.warn(
            f"client with id '{client.id}' already present in pool. Replacing with {client}",
            category=UserWarning,
        )
    if thing_id:
        self._thing_to_client_map[thing_id] = client.id
        self._client_to_thing_map[client.id] = thing_id

start_polling

start_polling() -> None

register the server message polling loop in the asyncio event loop

Source code in hololinked/hololinked/core/zmq/brokers.py

def start_polling(self) -> None:
    """register the server message polling loop in the asyncio event loop"""
    get_current_async_loop().create_task(self.poll_responses())

poll_responses async

poll_responses() -> None

Poll for replies from server.This method should be independently started in the event loop by calling start_polling(). Sending message requests and retrieving a response is still carried out by other methods. Do not duplicate this method call as there are no checks how many pollers exist and messages will become malformed if multiple pollers are active.

Source code in hololinked/hololinked/core/zmq/brokers.py

async def poll_responses(self) -> None:
    """
    Poll for replies from server.This method should be independently started in the event loop by calling `start_polling()`.
    Sending message requests and retrieving a response is still carried out by other methods.
    Do not duplicate this method call as there are no checks how many pollers exist and messages will become malformed
    if multiple pollers are active.
    """
    self.logger.info("client polling started for sockets for {}".format(list(self.pool.keys())))
    self.stop_poll = False
    event_loop = asyncio.get_event_loop()
    while not self.stop_poll:
        sockets = await self.poller.poll(self.poll_timeout)  # type hints dont work in this line
        for socket, _ in sockets:
            while True:
                try:
                    raw_response = await socket.recv_multipart(zmq.NOBLOCK)
                    response_message = ResponseMessage(raw_response)
                except zmq.Again:
                    # errors in handle_message should reach the client.
                    break
                except ConnectionAbortedError:
                    for client in self.pool.values():
                        if client.socket.get_monitor_socket() == socket:
                            self.poller.unregister(client.socket)  # leave the monitor in the pool
                            client.handshake(timeout=None)
                            self.logger.error(
                                f"client {client.id} disconnected from server."
                                + " Unregistering from poller temporarily until server comes back."
                            )
                            break
                else:
                    if self.handled_default_message_types(response_message):
                        continue
                    message_id = response_message.id
                    self.logger.debug(
                        "received response from server",
                        sender_id=response_message.sender_id,
                        receiver_id=response_message.receiver_id,
                        msg_id=response_message.id,
                        message_type=response_message.type,
                    )
                    if message_id in self.cancelled_messages:
                        self.cancelled_messages.remove(message_id)
                        self.logger.debug("dropping a cancelled message", msg_id=message_id)
                        continue
                    self.message_map[message_id] = response_message
                    event = self.events_map.get(message_id, None)
                    if event:
                        event.set()
                    else:
                        event_loop.create_task(self._resolve_response(message_id, response_message))

stop_polling

stop_polling()

stop polling for replies from server

Source code in hololinked/hololinked/core/zmq/brokers.py

def stop_polling(self):
    """
    stop polling for replies from server
    """
    self.stop_poll = True
    for client in self.pool.values():
        client.stop()

async_execute async

async_execute(thing_id: str, objekt: str, operation: str, payload: SerializableData = SerializableNone, preserialized_payload: PreserializedData = PreserializedEmptyByte, server_execution_context: ServerExecutionContext = default_server_execution_context, thing_execution_context: ThingExecutionContext = default_thing_execution_context) -> ResponseMessage

send an operation and receive the response for it.

Parameters:

Name	Type	Description	Default
client_id		id of the client in the pool to be used to send the message	required
thing_id	str	id of the Thing on which an operation is to be performed	required
objekt	str	name of property, action or event (usually only property or action)	required
operation	str	operation to be performed, like readproperty, writeproperty, invokeaction etc.	required
payload	SerializableData	serializable data to be sent as payload	SerializableNone
preserialized_payload	PreserializedData	pre-encoded data to be sent as payload, generally used for large or custom data that is already serialized	PreserializedEmptyByte
server_execution_context	ServerExecutionContext	Specify server level execution context like invokationTimeout, executionTimeout, oneway operation etc.	default_server_execution_context
thing_execution_context	ThingExecutionContext	Specify thing level execution context like fetchExecutionLogs etc.	default_thing_execution_context

Returns:

Type	Description
ResponseMessage	response message from server after completing the operation

Source code in hololinked/hololinked/core/zmq/brokers.py

async def async_execute(
    self,
    thing_id: str,
    objekt: str,
    operation: str,
    payload: SerializableData = SerializableNone,
    preserialized_payload: PreserializedData = PreserializedEmptyByte,
    server_execution_context: ServerExecutionContext = default_server_execution_context,
    thing_execution_context: ThingExecutionContext = default_thing_execution_context,
) -> ResponseMessage:
    """
    send an operation and receive the response for it.

    Parameters
    ----------
    client_id: str
        id of the client in the pool to be used to send the message
    thing_id: str
        `id` of the `Thing` on which an operation is to be performed
    objekt: str
        name of property, action or event (usually only property or action)
    operation: str
        operation to be performed, like `readproperty`, `writeproperty`, `invokeaction` etc.
    payload: SerializableData
        serializable data to be sent as payload
    preserialized_payload: PreserializedData
        pre-encoded data to be sent as payload, generally used for large or custom data that is already serialized
    server_execution_context: dict[str, Any]
        Specify server level execution context like `invokationTimeout`, `executionTimeout`, `oneway` operation etc.
    thing_execution_context: dict[str, Any]
        Specify thing level execution context like `fetchExecutionLogs` etc.

    Returns
    -------
    ResponseMessage
        response message from server after completing the operation
    """
    message_id = await self.async_send_request(
        thing_id=thing_id,
        objekt=objekt,
        operation=operation,
        payload=payload,
        preserialized_payload=preserialized_payload,
        server_execution_context=server_execution_context,
        thing_execution_context=thing_execution_context,
    )
    return await self.async_recv_response(
        thing_id=thing_id,
        message_id=message_id,
    )

async_send_request async

async_send_request(thing_id: str, objekt: str, operation: str, payload: SerializableData = SerializableNone, preserialized_payload: PreserializedData = PreserializedEmptyByte, server_execution_context: ServerExecutionContext = default_server_execution_context, thing_execution_context: ThingExecutionContext = default_thing_execution_context) -> str

send request message to server.

Parameters:

Name	Type	Description	Default
client_id		id of the client in the pool to be used to send the message	required
thing_id	str	id of the Thing on which an operation is to be performed	required
objekt	str	name of property, action or event (usually only property or action)	required
operation	str	operation to be performed, like readproperty, writeproperty, invokeaction etc.	required
payload	SerializableData	serializable data to be sent as payload	SerializableNone
preserialized_payload	PreserializedData	pre-encoded data to be sent as payload, generally used for large or custom data that is already serialized	PreserializedEmptyByte
server_execution_context	ServerExecutionContext	Specify server level execution context like invokationTimeout, executionTimeout, oneway operation etc.	default_server_execution_context
thing_execution_context	ThingExecutionContext	Specify thing level execution context like fetchExecutionLogs etc.	default_thing_execution_context

Returns:

Type	Description
bytes	a message id in bytes

Source code in hololinked/hololinked/core/zmq/brokers.py

async def async_send_request(
    self,
    thing_id: str,
    objekt: str,
    operation: str,
    payload: SerializableData = SerializableNone,
    preserialized_payload: PreserializedData = PreserializedEmptyByte,
    server_execution_context: ServerExecutionContext = default_server_execution_context,
    thing_execution_context: ThingExecutionContext = default_thing_execution_context,
) -> str:
    """
    send request message to server.

    Parameters
    ----------
    client_id: str
        id of the client in the pool to be used to send the message
    thing_id: str
        `id` of the `Thing` on which an operation is to be performed
    objekt: str
        name of property, action or event (usually only property or action)
    operation: str
        operation to be performed, like `readproperty`, `writeproperty`, `invokeaction` etc.
    payload: SerializableData
        serializable data to be sent as payload
    preserialized_payload: PreserializedData
        pre-encoded data to be sent as payload, generally used for large or custom data that is already serialized
    server_execution_context: dict[str, Any]
        Specify server level execution context like `invokationTimeout`, `executionTimeout`, `oneway` operation etc.
    thing_execution_context: dict[str, Any]
        Specify thing level execution context like `fetchExecutionLogs` etc.

    Returns
    -------
    bytes
        a message id in bytes
    """
    client_id = self.get_client_id_from_thing_id(thing_id)
    self.assert_client_ready(self.pool[client_id])
    message_id = await self.pool[client_id].async_send_request(
        thing_id=thing_id,
        objekt=objekt,
        operation=operation,
        payload=payload,
        preserialized_payload=preserialized_payload,
        server_execution_context=server_execution_context,
        thing_execution_context=thing_execution_context,
    )
    event = self.event_pool.pop()
    self.events_map[message_id] = event
    return message_id

async_recv_response async

async_recv_response(thing_id: str, message_id: bytes, timeout: float | int | None = None) -> ResponseMessage

Receive response for specified message ID.

Parameters:

Name	Type	Description	Default
client_id		id of the client in the pool to be used to receive the message	required
message_id	bytes	the message id for which response needs to be fetched	required
timeout	float | int | None	Client side timeout, not the same as timeout passed to server. Recommended to be None in general cases. The reply is dropped if this timeout occurs. Usually the server always replies when either of invokation or execution timeout occurs. This timeout ensures that the protocol binding does not wait indefinitely for a message that may never arrive.	None

Returns:

Name	Type	Description
response	ResponseMessage	response message from server corresponding to the message id

Raises:

Type	Description
ValueError	if supplied message id is not valid
TimeoutError	if timeout is not None and response did not arrive

Source code in hololinked/hololinked/core/zmq/brokers.py

async def async_recv_response(
    self, thing_id: str, message_id: bytes, timeout: float | int | None = None
) -> ResponseMessage:
    """
    Receive response for specified message ID.

    Parameters
    ----------
    client_id: str
        id of the client in the pool to be used to receive the message
    message_id: bytes
        the message id for which response needs to be fetched
    timeout: float | int | None
        Client side timeout, not the same as timeout passed to server. Recommended to be None in general cases.
        The reply is dropped if this timeout occurs. Usually the server always replies when either of invokation or
        execution timeout occurs. This timeout ensures that the protocol binding does not wait indefinitely for a message
        that may never arrive.

    Returns
    -------
    response: ResponseMessage
        response message from server corresponding to the message id

    Raises
    ------
    ValueError
        if supplied message id is not valid
    TimeoutError
        if timeout is not None and response did not arrive
    """
    try:
        client_id = self.get_client_id_from_thing_id(thing_id)
        event = self.events_map[message_id]
    except KeyError:
        raise KeyError(f"message id {message_id} unknown.") from None
    while True:
        try:
            await asyncio.wait_for(event.wait(), timeout)
            # default 5 seconds because we want to check if server is also dead
            if event.is_set():  # i.e. if timeout is not None, check if event is set
                break
            self.assert_client_ready(self.pool[client_id])
        except TimeoutError:
            self.cancelled_messages.append(message_id)
            self.logger.error("message added to list of cancelled messages", msg_id=message_id)
            raise TimeoutError(f"Execution not completed within {timeout} seconds") from None
    self.events_map.pop(message_id)
    self.event_pool.completed(event)
    response = self.message_map.pop(message_id)
    return response

handshake

handshake(timeout: int | None = 60000) -> None

schedules handshake coroutines for each client in the running event loop or completes handshake synchronously if no event loop is running. Use handshake_complete() async method to check if handshake is complete.

Parameters:

Name	Type	Description	Default
timeout	int | None	timeout in milliseconds to wait for handshake to complete. If handshake does not complete within this time, a ConnectionError is raised. If None, wait indefinitely until handshake completes.	60000

Source code in hololinked/hololinked/core/zmq/brokers.py

def handshake(self, timeout: int | None = 60000) -> None:
    """
    schedules handshake coroutines for each client in the running event loop
    or completes handshake synchronously if no event loop is running.
    Use `handshake_complete()` async method to check if handshake is complete.

    Parameters
    ----------
    timeout: float | int
        timeout in milliseconds to wait for handshake to complete. If handshake does not complete within this time,
        a `ConnectionError` is raised. If None, wait indefinitely until handshake completes.
    """
    for client in self.pool.values():
        client.handshake(timeout)

handshake_complete async

handshake_complete() -> None

wait for handshake to complete for all clients in the pool

Source code in hololinked/hololinked/core/zmq/brokers.py

async def handshake_complete(self) -> None:
    """wait for handshake to complete for all clients in the pool"""
    for client in self.pool.values():
        await client.handshake_complete()  # sufficient to wait serially