hololinked.core.zmq.brokers.MessageMappedZMQClientPool

Bases: BaseZMQClient

Pool of clients where message ID can track the replies irrespective of order of arrival.

Parameters:

Name	Type	Description	Default
`server_ids`	`List[str]`	list of instance names of servers to connect to	required
`id`	`str`	Unique identity of the client to receive messages from the server. Each client connecting to same server must still have unique ID.	required
`client_type`		ZMQ or HTTP Server	required
`handshake`	`bool`	when true, handshake with the server first before allowing first message and block until that handshake was accomplished.	`True`
`poll_timeout`	`int`	socket polling timeout in milliseconds greater than 0.	`25`
`transport`	`str`	transport implemented by ZMQ server	`'IPC'`
`context`	`Context`	ZMQ context	`None`
`deserialize_server_messages`		deserializes the data field of the message	required
`**kwargs`		zmq_serializer: BaseSerializer custom implementation of ZMQ serializer if necessary http_serializer: JSONSerializer custom implementation of JSON serializer if necessary	`{}`

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

class MessageMappedZMQClientPool(BaseZMQClient):
    """
    Pool of clients where message ID can track the replies irrespective of order of arrival. 

    Parameters
    ----------
    server_ids: List[str]
        list of instance names of servers to connect to
    id: str
        Unique identity of the client to receive messages from the server. Each client connecting to same server must 
        still have unique ID.
    client_type: str
        ZMQ or HTTP Server
    handshake: bool
        when true, handshake with the server first before allowing first message and block until that handshake was
        accomplished.
    poll_timeout: int
        socket polling timeout in milliseconds greater than 0.
    transport: str
        transport implemented by ZMQ server
    context: zmq.asyncio.Context
        ZMQ context
    deserialize_server_messages: bool
        deserializes the data field of the message
    **kwargs:
        zmq_serializer: BaseSerializer
            custom implementation of ZMQ serializer if necessary
        http_serializer: JSONSerializer
            custom implementation of JSON serializer if necessary
    """

    def __init__(self, 
                id: str,
                client_ids: typing.List[str], 
                server_ids: typing.List[str], 
                handshake: bool = True, 
                poll_timeout: int = 25, 
                transport: str = 'IPC',  
                context: zmq.asyncio.Context = None, 
                **kwargs
            ) -> None:
        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 zmq.asyncio.Context()
        self.pool = dict() # type: typing.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, 
                            transport=transport, 
                            context=self.context, 
                            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: typing.Dict[bytes, asyncio.Event]
        self.message_map = dict()
        self.cancelled_messages = []
        self.poll_timeout = poll_timeout
        self.stop_poll = False 


    def create_new(self, id: str, server_id: str, transport: str = 'IPC') -> None:
        """
        Create new server with specified transport. other arguments are taken from pool specifications. 

        Parameters
        ----------
        id: str
            instance name of server 
        transport: str
            transport implemented by ZMQ server
        """
        if server_id not in self.pool.keys():
            client = AsyncZMQClient(
                        id=id,
                        server_id=server_id,
                        handshake=True, 
                        transport=transport, 
                        context=self.context, 
                        logger=self.logger
                    )
            client._monitor_socket = client.socket.get_monitor_socket()
            self.poller.register(client._monitor_socket, zmq.POLLIN)
            self.pool[server_id] = client
        else: 
            raise ValueError(f"client for instance name '{server_id}' already present in pool")


    def register(self, client: AsyncZMQClient) -> None:
        """
        Register a client with the pool. 

        Parameters
        ----------
        client: AsyncZMQClient
            client to be registered
        """
        if not isinstance(client, AsyncZMQClient):
            raise TypeError("registration possible for clients only subclass of AsyncZMQClient." +
                           f" Given type {type(client)}")
        self.pool[client.id] = client 
        self.poller.register(client.socket, zmq.POLLIN)
        self.poller.register(client._monitor_socket, zmq.POLLIN)


    def get_client_id_from_thing_id(self, thing_id: str) -> typing.Dict[str, AsyncZMQClient]:
        """
        map of thing_id to client
        """
        raise NotImplementedError("get_client_id_from_thing_id not implemented for MessageMappedZMQClientPool")

    @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:
        """
        automatically called when handshake argument at init is True. When not automatically called, it is necessary
        to call this method before awaiting ``handshake_complete()``.
        """
        for client in self.pool.values():
            run_callable_somehow(client._handshake(timeout))


    async def poll_responses(self) -> None:
        """
        Poll for replies from server. Since the client is message mapped, this method should be independently started
        in the event loop. Sending message and retrieving a message mapped is still carried out by other methods.
        """
        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.id} disconnected." +
                                    " Unregistering from poller temporarily until server comes back.")
                                break
                    else:
                        message_id = response_message.id
                        self.logger.debug(f"received response from server '{response_message.sender_id}' with msg-ID '{message_id}'")
                        if message_id in self.cancelled_messages:
                            self.cancelled_messages.remove(message_id)
                            self.logger.debug(f"msg-ID '{message_id}' cancelled")
                            continue
                        event = self.events_map.get(message_id, None) 
                        final_data = response_message.body
                        # if len(response_message.pre_encoded_payload ) > 0 and response_message.payload:
                        #     final_data = tuple(response_message.body)
                        # elif len(response_message.pre_encoded_payload) > 0:
                        #     final_data = response_message.pre_encoded_payload
                        # else:
                        #     final_data = response_message.payload
                        if event:
                            event.set()
                        else:    
                            invalid_event_task = asyncio.create_task(self._resolve_response(message_id, final_data))
                            event_loop.call_soon(lambda: invalid_event_task)


    async def _resolve_response(self, message_id: str, data: typing.Any) -> None:
        """
        This method is called when there is an asyncio Event not 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: bytes
            the data given by the server which needs to mapped to the message
        """
        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(f'message_id {message_id} cancelled')
                    return 
                if i >= max_number_of_retries - 1:
                    self.logger.error("unknown message id {} without corresponding event object".format(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 not client.socket 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, 
                            client_id: str, 
                            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
                        ) -> bytes:
        """
        Send operation to server with instance name. Replies are automatically polled & to be retrieved using 
        ``async_recv_response()``

        Parameters
        ----------
        id: str
            instance name of the server
        operation: str
            unique str identifying a server side or ``Thing`` resource. These values corresponding 
            to automatically extracted name from the object name or the URL_path prepended with the instance name. 
        arguments: Dict[str, Any]
            if the operation invokes a method, arguments of that method. 
        server_execution_context: Dict[str, Any]
            see execution context definitions
        thing_execution_context: Dict[str, Any]
            see execution context definitions

        Returns
        -------
        message_id: bytes
            created message 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, 
                                client_id: str, 
                                message_id: bytes, 
                                timeout: float | int | None = None
                            ) -> typing.Dict[str, typing.Any]:
        """
        Receive response for specified message ID. 

        Parameters
        ----------
        message_id: bytes
            the message id for which response needs to eb fetched
        raise_client_side_exceptions: bool, default False
            raise exceptions from server on client side
        timeout: float, 
            client side timeout, not the same as timeout passed to server, recommended to be None in general cases. 
            Server side timeouts ensure start of execution of operations within specified timeouts and 
            drops execution altogether if timeout occured. Client side timeouts only wait for message to come within 
            the timeout, but do not gaurantee non-execution.  

        Returns
        -------
        response: dict, Any
            dictionary when plain response is False, any value returned from execution on the server side if plain response is
            True.

        Raises
        ------
        ValueError: 
            if supplied message id is not valid
        TimeoutError:
            if timeout is not None and response did not arrive
        """
        try:
            event = self.events_map[message_id]
        except KeyError:
            raise ValueError(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.debug(f'message_id {message_id} added to list of cancelled messages')
                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, 
                        client_id: str, 
                        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, 
                    ) -> typing.Dict[str, typing.Any]:
        """
        sends message and receives response.

        Parameters
        ----------
        id: str
            instance name of the server
        operation: str
            unique str identifying a server side or ``Thing`` resource. These values corresponding 
            to automatically extracted name from the object name or the URL_path prepended with the instance name. 
        arguments: Dict[str, Any]
            if the operation invokes a method, arguments of that method. 
        context: Dict[str, Any]
            see execution context definitions
        raise_client_side_exceptions: bool, default False
            raise exceptions from server on client side
        invokation_timeout: float, default 5
            server side timeout
        execution_timeout: float, default None
            client side timeout, not the same as timeout passed to server, recommended to be None in general cases. 
            Server side timeouts ensure start of execution of operations within specified timeouts and 
            drops execution altogether if timeout occured. Client side timeouts only wait for message to come within 
            the timeout, but do not gaurantee non-execution.  
        """
        message_id = await self.async_send_request(
                                                client_id=client_id, 
                                                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(
                                            client_id=client_id, 
                                            message_id=message_id, 
                                        )

    def start_polling(self) -> None:
        """
        register the server message polling loop in the asyncio event loop. 
        """
        event_loop = asyncio.get_event_loop()
        event_loop.call_soon(lambda: asyncio.create_task(self.poll_responses()))

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

    async def async_execute_in_all(self, 
                                objekt: str, 
                                operation: str, 
                                payload: SerializableData = SerializableNone, 
                                preserialized_payload: PreserializedData = PreserializedEmptyByte,
                                ids: typing.Optional[typing.List[str]] = None,
                                server_execution_context: ServerExecutionContext = default_server_execution_context,
                                thing_execution_context: ThingExecutionContext = default_thing_execution_context,
                            ) -> typing.Dict[str, typing.Any]:
        """
        execute a specified operation in all Thing including eventloops
        """
        if not ids:
            ids = self.pool.keys()
        gathered_replies = await asyncio.gather(*[
            self.async_execute(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 ids])
        replies = dict()
        for id, response in zip(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,        
                                    ) -> typing.Dict[str, typing.Any]:
        """
        execute the same operation in all Things, eventloops are excluded. 
        """
        return await self.async_execute_in_all(
                        objekt=objekt, operation=operation, payload=payload, 
                        preserialized_payload=preserialized_payload,
                        ids=[id for id, client in self.pool.items()],
                        server_execution_context=server_execution_context,
                        thing_execution_context=thing_execution_context
                    )

    async def ping_all_servers(self):
        """
        ping all servers connected to the client pool, calls ping() on Thing
        """
        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) -> typing.Iterator[AsyncZMQClient]:
        return iter(self.pool.values())

    def exit(self) -> None:
        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 from pool for '{}'".format(self.__class__))
        try:
            self.context.term()
            self.logger.info("context terminated for '{}'".format(self.__class__))        
        except Exception as ex:
            self.logger.warning("could not properly terminate context or attempted to terminate an already terminated context" +
                            "'{}'. Exception message: {}".format(self.identity, str(ex)))

    """
    BaseZMQ
    BaseAsyncZMQ
    BaseSyncZMQ
    BaseZMQClient
    SyncZMQClient
    AsyncZMQClient
    MessageMappedClientPool
    """

Functions

async_execute `async`

async_execute(client_id: str, 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) -> typing.Dict[str, typing.Any]

sends message and receives response.

Parameters:

Name	Type	Description	Default
`id`		instance name of the server	required
`operation`	`str`	unique str identifying a server side or `Thing` resource. These values corresponding to automatically extracted name from the object name or the URL_path prepended with the instance name.	required
`arguments`		if the operation invokes a method, arguments of that method.	required
`context`		see execution context definitions	required
`raise_client_side_exceptions`		raise exceptions from server on client side	required
`invokation_timeout`		server side timeout	required
`execution_timeout`		client side timeout, not the same as timeout passed to server, recommended to be None in general cases. Server side timeouts ensure start of execution of operations within specified timeouts and drops execution altogether if timeout occured. Client side timeouts only wait for message to come within the timeout, but do not gaurantee non-execution.	required

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

async def async_execute(self, 
                    client_id: str, 
                    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, 
                ) -> typing.Dict[str, typing.Any]:
    """
    sends message and receives response.

    Parameters
    ----------
    id: str
        instance name of the server
    operation: str
        unique str identifying a server side or ``Thing`` resource. These values corresponding 
        to automatically extracted name from the object name or the URL_path prepended with the instance name. 
    arguments: Dict[str, Any]
        if the operation invokes a method, arguments of that method. 
    context: Dict[str, Any]
        see execution context definitions
    raise_client_side_exceptions: bool, default False
        raise exceptions from server on client side
    invokation_timeout: float, default 5
        server side timeout
    execution_timeout: float, default None
        client side timeout, not the same as timeout passed to server, recommended to be None in general cases. 
        Server side timeouts ensure start of execution of operations within specified timeouts and 
        drops execution altogether if timeout occured. Client side timeouts only wait for message to come within 
        the timeout, but do not gaurantee non-execution.  
    """
    message_id = await self.async_send_request(
                                            client_id=client_id, 
                                            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(
                                        client_id=client_id, 
                                        message_id=message_id, 
                                    )

async_execute_in_all `async`

async_execute_in_all(objekt: str, operation: str, payload: SerializableData = SerializableNone, preserialized_payload: PreserializedData = PreserializedEmptyByte, ids: typing.Optional[typing.List[str]] = None, server_execution_context: ServerExecutionContext = default_server_execution_context, thing_execution_context: ThingExecutionContext = default_thing_execution_context) -> typing.Dict[str, typing.Any]

execute a specified operation in all Thing including eventloops

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

async def async_execute_in_all(self, 
                            objekt: str, 
                            operation: str, 
                            payload: SerializableData = SerializableNone, 
                            preserialized_payload: PreserializedData = PreserializedEmptyByte,
                            ids: typing.Optional[typing.List[str]] = None,
                            server_execution_context: ServerExecutionContext = default_server_execution_context,
                            thing_execution_context: ThingExecutionContext = default_thing_execution_context,
                        ) -> typing.Dict[str, typing.Any]:
    """
    execute a specified operation in all Thing including eventloops
    """
    if not ids:
        ids = self.pool.keys()
    gathered_replies = await asyncio.gather(*[
        self.async_execute(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 ids])
    replies = dict()
    for id, response in zip(ids, gathered_replies):
        replies[id] = response
    return replies  

async_execute_in_all_things `async`

async_execute_in_all_things(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) -> typing.Dict[str, typing.Any]

execute the same operation in all Things, eventloops are excluded.

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

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,        
                                ) -> typing.Dict[str, typing.Any]:
    """
    execute the same operation in all Things, eventloops are excluded. 
    """
    return await self.async_execute_in_all(
                    objekt=objekt, operation=operation, payload=payload, 
                    preserialized_payload=preserialized_payload,
                    ids=[id for id, client in self.pool.items()],
                    server_execution_context=server_execution_context,
                    thing_execution_context=thing_execution_context
                )

async_recv_response `async`

async_recv_response(client_id: str, message_id: bytes, timeout: float | int | None = None) -> typing.Dict[str, typing.Any]

Receive response for specified message ID.

Parameters:

Name	Type	Description	Default
`message_id`	`bytes`	the message id for which response needs to eb fetched	required
`raise_client_side_exceptions`		raise exceptions from server on client side	required
`timeout`	`float \| int \| None`	client side timeout, not the same as timeout passed to server, recommended to be None in general cases. Server side timeouts ensure start of execution of operations within specified timeouts and drops execution altogether if timeout occured. Client side timeouts only wait for message to come within the timeout, but do not gaurantee non-execution.	`None`

Returns:

Name	Type	Description
`response`	`(dict, Any)`	dictionary when plain response is False, any value returned from execution on the server side if plain response is True.

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\core\zmq\brokers.py

async def async_recv_response(self, 
                            client_id: str, 
                            message_id: bytes, 
                            timeout: float | int | None = None
                        ) -> typing.Dict[str, typing.Any]:
    """
    Receive response for specified message ID. 

    Parameters
    ----------
    message_id: bytes
        the message id for which response needs to eb fetched
    raise_client_side_exceptions: bool, default False
        raise exceptions from server on client side
    timeout: float, 
        client side timeout, not the same as timeout passed to server, recommended to be None in general cases. 
        Server side timeouts ensure start of execution of operations within specified timeouts and 
        drops execution altogether if timeout occured. Client side timeouts only wait for message to come within 
        the timeout, but do not gaurantee non-execution.  

    Returns
    -------
    response: dict, Any
        dictionary when plain response is False, any value returned from execution on the server side if plain response is
        True.

    Raises
    ------
    ValueError: 
        if supplied message id is not valid
    TimeoutError:
        if timeout is not None and response did not arrive
    """
    try:
        event = self.events_map[message_id]
    except KeyError:
        raise ValueError(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.debug(f'message_id {message_id} added to list of cancelled messages')
            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_send_request `async`

async_send_request(client_id: str, 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) -> bytes

Send operation to server with instance name. Replies are automatically polled & to be retrieved using async_recv_response()

Parameters:

Name	Type	Description	Default
`id`		instance name of the server	required
`operation`	`str`	unique str identifying a server side or `Thing` resource. These values corresponding to automatically extracted name from the object name or the URL_path prepended with the instance name.	required
`arguments`		if the operation invokes a method, arguments of that method.	required
`server_execution_context`	`ServerExecutionContext`	see execution context definitions	`default_server_execution_context`
`thing_execution_context`	`ThingExecutionContext`	see execution context definitions	`default_thing_execution_context`

Returns:

Name	Type	Description
`message_id`	`bytes`	created message ID

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

async def async_send_request(self, 
                        client_id: str, 
                        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
                    ) -> bytes:
    """
    Send operation to server with instance name. Replies are automatically polled & to be retrieved using 
    ``async_recv_response()``

    Parameters
    ----------
    id: str
        instance name of the server
    operation: str
        unique str identifying a server side or ``Thing`` resource. These values corresponding 
        to automatically extracted name from the object name or the URL_path prepended with the instance name. 
    arguments: Dict[str, Any]
        if the operation invokes a method, arguments of that method. 
    server_execution_context: Dict[str, Any]
        see execution context definitions
    thing_execution_context: Dict[str, Any]
        see execution context definitions

    Returns
    -------
    message_id: bytes
        created message 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

create_new

create_new(id: str, server_id: str, transport: str = 'IPC') -> None

Create new server with specified transport. other arguments are taken from pool specifications.

Parameters:

Name	Type	Description	Default
`id`	`str`	instance name of server	required
`transport`	`str`	transport implemented by ZMQ server	`'IPC'`

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

def create_new(self, id: str, server_id: str, transport: str = 'IPC') -> None:
    """
    Create new server with specified transport. other arguments are taken from pool specifications. 

    Parameters
    ----------
    id: str
        instance name of server 
    transport: str
        transport implemented by ZMQ server
    """
    if server_id not in self.pool.keys():
        client = AsyncZMQClient(
                    id=id,
                    server_id=server_id,
                    handshake=True, 
                    transport=transport, 
                    context=self.context, 
                    logger=self.logger
                )
        client._monitor_socket = client.socket.get_monitor_socket()
        self.poller.register(client._monitor_socket, zmq.POLLIN)
        self.pool[server_id] = client
    else: 
        raise ValueError(f"client for instance name '{server_id}' already present in pool")

get_client_id_from_thing_id

get_client_id_from_thing_id(thing_id: str) -> typing.Dict[str, AsyncZMQClient]

map of thing_id to client

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

def get_client_id_from_thing_id(self, thing_id: str) -> typing.Dict[str, AsyncZMQClient]:
    """
    map of thing_id to client
    """
    raise NotImplementedError("get_client_id_from_thing_id not implemented for MessageMappedZMQClientPool")

handshake

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

automatically called when handshake argument at init is True. When not automatically called, it is necessary to call this method before awaiting handshake_complete().

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

def handshake(self, timeout: int | None = 60000) -> None:
    """
    automatically called when handshake argument at init is True. When not automatically called, it is necessary
    to call this method before awaiting ``handshake_complete()``.
    """
    for client in self.pool.values():
        run_callable_somehow(client._handshake(timeout))

handshake_complete `async`

handshake_complete() -> None

wait for handshake to complete for all clients in the pool

Source code in 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

ping_all_servers `async`

ping_all_servers()

ping all servers connected to the client pool, calls ping() on Thing

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

async def ping_all_servers(self):
    """
    ping all servers connected to the client pool, calls ping() on Thing
    """
    return await self.async_execute_in_all() #operation='invokeAction', objekt=CommonRPC.PING)

poll_responses `async`

poll_responses() -> None

Poll for replies from server. Since the client is message mapped, this method should be independently started in the event loop. Sending message and retrieving a message mapped is still carried out by other methods.

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

async def poll_responses(self) -> None:
    """
    Poll for replies from server. Since the client is message mapped, this method should be independently started
    in the event loop. Sending message and retrieving a message mapped is still carried out by other methods.
    """
    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.id} disconnected." +
                                " Unregistering from poller temporarily until server comes back.")
                            break
                else:
                    message_id = response_message.id
                    self.logger.debug(f"received response from server '{response_message.sender_id}' with msg-ID '{message_id}'")
                    if message_id in self.cancelled_messages:
                        self.cancelled_messages.remove(message_id)
                        self.logger.debug(f"msg-ID '{message_id}' cancelled")
                        continue
                    event = self.events_map.get(message_id, None) 
                    final_data = response_message.body
                    # if len(response_message.pre_encoded_payload ) > 0 and response_message.payload:
                    #     final_data = tuple(response_message.body)
                    # elif len(response_message.pre_encoded_payload) > 0:
                    #     final_data = response_message.pre_encoded_payload
                    # else:
                    #     final_data = response_message.payload
                    if event:
                        event.set()
                    else:    
                        invalid_event_task = asyncio.create_task(self._resolve_response(message_id, final_data))
                        event_loop.call_soon(lambda: invalid_event_task)

register

register(client: AsyncZMQClient) -> None

Register a client with the pool.

Parameters:

Name	Type	Description	Default
`client`	`AsyncZMQClient`	client to be registered	required

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

def register(self, client: AsyncZMQClient) -> None:
    """
    Register a client with the pool. 

    Parameters
    ----------
    client: AsyncZMQClient
        client to be registered
    """
    if not isinstance(client, AsyncZMQClient):
        raise TypeError("registration possible for clients only subclass of AsyncZMQClient." +
                       f" Given type {type(client)}")
    self.pool[client.id] = client 
    self.poller.register(client.socket, zmq.POLLIN)
    self.poller.register(client._monitor_socket, zmq.POLLIN)

start_polling

start_polling() -> None

register the server message polling loop in the asyncio event loop.

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

def start_polling(self) -> None:
    """
    register the server message polling loop in the asyncio event loop. 
    """
    event_loop = asyncio.get_event_loop()
    event_loop.call_soon(lambda: asyncio.create_task(self.poll_responses()))

stop_polling

stop_polling()

stop polling for replies from server

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

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

hololinked.core.zmq.brokers.MessageMappedZMQClientPool

server_ids

id

client_type

handshake

poll_timeout

transport

context

deserialize_server_messages

**kwargs

Functions

async_execute async

id

operation

arguments

context

raise_client_side_exceptions

invokation_timeout

execution_timeout

async_execute_in_all async

async_execute_in_all_things async

async_recv_response async

message_id

raise_client_side_exceptions

timeout

async_send_request async

id

operation

arguments

server_execution_context

thing_execution_context

create_new

id

transport

get_client_id_from_thing_id

handshake

handshake_complete async

ping_all_servers async

poll_responses async

register

client

start_polling

stop_polling

`server_ids`

`id`

`client_type`

`handshake`

`poll_timeout`

`transport`

`context`

`deserialize_server_messages`

`**kwargs`

async_execute `async`

`id`

`operation`

`arguments`

`context`

`raise_client_side_exceptions`

`invokation_timeout`

`execution_timeout`

async_execute_in_all `async`

async_execute_in_all_things `async`

async_recv_response `async`

`message_id`

`raise_client_side_exceptions`

`timeout`

async_send_request `async`

`id`

`operation`

`arguments`

`server_execution_context`

`thing_execution_context`

`id`

`transport`

handshake_complete `async`

ping_all_servers `async`

poll_responses `async`

`client`