EventConsumer

hololinked.server.zmq.brokers.BaseEventConsumer

Bases: BaseZMQClient

Consumes events published at PUB sockets using SUB socket.

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

class BaseEventConsumer(BaseZMQClient):
    """Consumes events published at PUB sockets using SUB socket."""

    # the sync and async subclasses each narrow these to their own flavour
    _poller_lock: threading.Lock | asyncio.Lock

    def __init__(
        self,
        id: str,
        event_unique_identifier: str,
        access_point: str,
        context: zmq.Context | None = None,
        **kwargs,
    ) -> None:
        """
        Initialize the event consumer.

        Parameters
        ----------
        id: str
            unique identity for the consumer
        event_unique_identifier: str
            unique identifier of the event registered at the PUB socket
        access_point: str
            socket address of the event publisher (`EventPublisher`), properly qualified with transport method
        context: zmq.Context
            ZMQ context to use, if None, a global context is used.
        **kwargs:
            additional arguments:

            - `logger`: `logging.Logger`, logger instance to use. If None, a default
            - `poll_timeout`: `int`, socket polling timeout in milliseconds greater than 0.
            - `server_id`: `str`, id of the PUB socket server, usually not necessary as `access_point` is sufficient.

        Raises
        ------
        TypeError
            if the consumer is not subclassed by either `BaseSyncZMQ` or `BaseAsyncZMQ`
        """
        if isinstance(self, BaseSyncZMQ):
            self.context = context or global_config.zmq_context()
            self.poller = zmq.Poller()
            socket_class = zmq.Socket
            self._poller_lock = threading.Lock()
        elif isinstance(self, BaseAsyncZMQ):
            self.context = context or global_config.zmq_context()
            self.poller = zmq.asyncio.Poller()
            socket_class = zmq.asyncio.Socket
            self._poller_lock = asyncio.Lock()
        else:
            raise TypeError("BaseEventConsumer must be subclassed by either BaseSyncZMQ or BaseAsyncZMQ")
        super().__init__(id=id, server_id=kwargs.get("server_id", None), **kwargs)  # ty: ignore[invalid-argument-type]
        logger = kwargs.get("logger", None)
        if not logger:
            logger = structlog.get_logger().bind(
                component="broker",
                impl=self.__class__.__name__,
                id=id,
                event_id=event_unique_identifier,
            )
        self.logger = logger  # type: structlog.stdlib.BoundLogger
        self.create_socket(
            server_id=id,
            socket_id=id,
            node_type="client",
            context=self.context,  # ty: ignore[invalid-argument-type]
            socket_type=zmq.SocketType.SUB,
            access_point=access_point,
            **kwargs,
        )
        self.event_unique_identifier = bytes(event_unique_identifier, encoding="utf-8")
        short_uuid = uuid_hex()
        self.interruptor = self.context.socket(zmq.SocketType.PAIR, socket_class=socket_class)
        self.interruptor.setsockopt_string(zmq.IDENTITY, f"interrupting-server-{short_uuid}")
        self.interrupting_peer = self.context.socket(zmq.SocketType.PAIR, socket_class=socket_class)
        self.interrupting_peer.setsockopt_string(zmq.IDENTITY, f"interrupting-client-{short_uuid}")
        self.interruptor.bind(f"inproc://{self.id}-{short_uuid}/interruption")
        self.interrupting_peer.connect(f"inproc://{self.id}-{short_uuid}/interruption")
        self._stop = False

    def subscribe(self) -> None:
        """Subscribe to the event at the PUB socket."""
        self.socket.setsockopt(zmq.SUBSCRIBE, self.event_unique_identifier)
        # pair sockets cannot be polled unforunately, so we use router
        # if self.socket in self.poller._map:
        #     self.poller.unregister(self.socket)
        # if self.interruptor in self.poller._map:
        #     self.poller.unregister(self.interruptor)
        self.poller.register(self.socket, zmq.POLLIN)
        self.poller.register(self.interruptor, zmq.POLLIN)
        self._stop = False

    def stop_polling(self) -> None:
        """Stop polling for events ending the `receive()` method."""
        self._stop = True

    @property
    def interrupt_message(self) -> EventMessage:
        """
        Craft an interrupt message to be sent to the interruptor socket, if `stop_polling()` is not sufficient as the poll timeout is infinite.

        Used internally by `interrupt()` method.
        """
        return EventMessage.craft_from_arguments(
            event_id=f"{self.id}/interrupting-server",
            sender_id=self.id,
            payload=SerializableData("INTERRUPT", content_type="application/json"),
        )

    def exit(self):
        self.stop_polling()
        BaseZMQ.exit(self)
        if self.socket.closed:
            return  # __del__ reaching a consumer that its own listener already exited
        for socket in (self.socket, self.interruptor):
            try:
                self.poller.unregister(socket)
            except KeyError:
                pass  # never registered - subscribe() was not called
            except Exception as ex:  # noqa: BLE001
                self.logger.warning(f"could not unregister socket from poller for event consumer - {str(ex)}")
        for socket in (self.socket, self.interruptor, self.interrupting_peer):
            try:
                socket.close(0)
            except Exception as ex:  # noqa: BLE001
                self.logger.warning(f"could not terminate socket of event consumer - {str(ex)}")
        self.logger.info(f"terminated event consuming socket {self.socket_address}")

Attributes

interrupt_message property

interrupt_message: EventMessage

Craft an interrupt message to be sent to the interruptor socket, if stop_polling() is not sufficient as the poll timeout is infinite.

Used internally by interrupt() method.

Functions

__init__

__init__(id: str, event_unique_identifier: str, access_point: str, context: Context | None = None, **kwargs) -> None

Initialize the event consumer.

Parameters:

Name	Type	Description	Default
id	str	unique identity for the consumer	required
event_unique_identifier	str	unique identifier of the event registered at the PUB socket	required
access_point	str	socket address of the event publisher (EventPublisher), properly qualified with transport method	required
context	Context | None	ZMQ context to use, if None, a global context is used.	None
**kwargs		additional arguments: logger: logging.Logger, logger instance to use. If None, a default poll_timeout: int, socket polling timeout in milliseconds greater than 0. server_id: str, id of the PUB socket server, usually not necessary as access_point is sufficient.	{}

Raises:

Type	Description
TypeError	if the consumer is not subclassed by either BaseSyncZMQ or BaseAsyncZMQ

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

def __init__(
    self,
    id: str,
    event_unique_identifier: str,
    access_point: str,
    context: zmq.Context | None = None,
    **kwargs,
) -> None:
    """
    Initialize the event consumer.

    Parameters
    ----------
    id: str
        unique identity for the consumer
    event_unique_identifier: str
        unique identifier of the event registered at the PUB socket
    access_point: str
        socket address of the event publisher (`EventPublisher`), properly qualified with transport method
    context: zmq.Context
        ZMQ context to use, if None, a global context is used.
    **kwargs:
        additional arguments:

        - `logger`: `logging.Logger`, logger instance to use. If None, a default
        - `poll_timeout`: `int`, socket polling timeout in milliseconds greater than 0.
        - `server_id`: `str`, id of the PUB socket server, usually not necessary as `access_point` is sufficient.

    Raises
    ------
    TypeError
        if the consumer is not subclassed by either `BaseSyncZMQ` or `BaseAsyncZMQ`
    """
    if isinstance(self, BaseSyncZMQ):
        self.context = context or global_config.zmq_context()
        self.poller = zmq.Poller()
        socket_class = zmq.Socket
        self._poller_lock = threading.Lock()
    elif isinstance(self, BaseAsyncZMQ):
        self.context = context or global_config.zmq_context()
        self.poller = zmq.asyncio.Poller()
        socket_class = zmq.asyncio.Socket
        self._poller_lock = asyncio.Lock()
    else:
        raise TypeError("BaseEventConsumer must be subclassed by either BaseSyncZMQ or BaseAsyncZMQ")
    super().__init__(id=id, server_id=kwargs.get("server_id", None), **kwargs)  # ty: ignore[invalid-argument-type]
    logger = kwargs.get("logger", None)
    if not logger:
        logger = structlog.get_logger().bind(
            component="broker",
            impl=self.__class__.__name__,
            id=id,
            event_id=event_unique_identifier,
        )
    self.logger = logger  # type: structlog.stdlib.BoundLogger
    self.create_socket(
        server_id=id,
        socket_id=id,
        node_type="client",
        context=self.context,  # ty: ignore[invalid-argument-type]
        socket_type=zmq.SocketType.SUB,
        access_point=access_point,
        **kwargs,
    )
    self.event_unique_identifier = bytes(event_unique_identifier, encoding="utf-8")
    short_uuid = uuid_hex()
    self.interruptor = self.context.socket(zmq.SocketType.PAIR, socket_class=socket_class)
    self.interruptor.setsockopt_string(zmq.IDENTITY, f"interrupting-server-{short_uuid}")
    self.interrupting_peer = self.context.socket(zmq.SocketType.PAIR, socket_class=socket_class)
    self.interrupting_peer.setsockopt_string(zmq.IDENTITY, f"interrupting-client-{short_uuid}")
    self.interruptor.bind(f"inproc://{self.id}-{short_uuid}/interruption")
    self.interrupting_peer.connect(f"inproc://{self.id}-{short_uuid}/interruption")
    self._stop = False

subscribe

subscribe() -> None

Subscribe to the event at the PUB socket.

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

def subscribe(self) -> None:
    """Subscribe to the event at the PUB socket."""
    self.socket.setsockopt(zmq.SUBSCRIBE, self.event_unique_identifier)
    # pair sockets cannot be polled unforunately, so we use router
    # if self.socket in self.poller._map:
    #     self.poller.unregister(self.socket)
    # if self.interruptor in self.poller._map:
    #     self.poller.unregister(self.interruptor)
    self.poller.register(self.socket, zmq.POLLIN)
    self.poller.register(self.interruptor, zmq.POLLIN)
    self._stop = False

stop_polling

stop_polling() -> None

Stop polling for events ending the receive() method.

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

def stop_polling(self) -> None:
    """Stop polling for events ending the `receive()` method."""
    self._stop = True

hololinked.server.zmq.brokers.EventConsumer

Bases: BaseEventConsumer, BaseSyncZMQ

Sync Event Consumer to used outside of async loops.

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

class EventConsumer(BaseEventConsumer, BaseSyncZMQ):
    """Sync Event Consumer to used outside of async loops."""

    _poller_lock: threading.Lock
    poller: zmq.Poller

    def receive(self, timeout: float | None = 1000, raise_interrupt_as_exception: bool = False) -> EventMessage | None:
        """
        Receive event with given timeout.

        Parameters
        ----------
        timeout: float, int, None
            timeout in milliseconds, None for blocking
        raise_interrupt_as_exception: bool
            if True, raises BreakLoop exception when interrupted, otherwise returns None

        Returns
        -------
        event_message: EventMessage | None
            the received event, or None if polling was interrupted or timed out

        Raises
        ------
        BreakLoop
            if polling was stopped or interrupted and `raise_interrupt_as_exception` is True
        zmq.ZMQError
            if the poll or the receive failed for a reason other than the sockets being closed
        """
        while not self._stop:
            if not self._poller_lock.acquire(timeout=timeout / 1000 if timeout else -1):
                continue
            try:
                if self._stop or self.socket.closed:
                    break
                sockets = self.poller.poll(timeout)  # ty: ignore[invalid-argument-type]  # list[tuple[zmq.Socket, int]]
                if len(sockets) > 1:
                    # if there is an interrupt message as well as an event,
                    # give preference to interrupt message.
                    if sockets[0][0] == self.interruptor:
                        sockets = [sockets[0]]  # we still need the socket, poll event  tuple
                    elif sockets[1][0] == self.interruptor:
                        sockets = [sockets[1]]
                for socket, _ in sockets:
                    try:
                        raw_message = socket.recv_multipart(zmq.NOBLOCK)
                        message = EventMessage(raw_message)
                        if socket == self.interruptor:
                            if message.payload.deserialize() == "INTERRUPT":
                                self.stop_polling()
                                break
                        return message
                    except zmq.Again:
                        pass
                    # if not self.handled_default_message_types(event_message):
            except zmq.ZMQError as ex:
                # the sockets or the context were closed under us, which is a teardown elsewhere and not
                # an error here - anything else is a real failure and belongs to the caller
                if ex.errno not in (zmq.ETERM, zmq.ENOTSOCK):
                    raise
                self.stop_polling()
                break
            finally:
                self._poller_lock.release()
        if raise_interrupt_as_exception:
            raise BreakLoop("event consumer interrupted")
        return None

    def interrupt(self):
        """
        Interrupts the event consumer.

        Generally should be used for exiting this object if there is no poll period/infinite polling.
        Otherwise please use stop_polling().
        """
        self.interrupting_peer.send_multipart(self.interrupt_message.byte_array)

    def exit(self) -> None:
        """
        Stop polling and close the sockets, waiting out a `receive()` already in flight.

        The poller lock is what makes closing safe from a thread other than the polling one - without it,
        `poll()` is left holding sockets that no longer exist.
        """
        self.stop_polling()
        acquired = self._poller_lock.acquire(timeout=2 * self.poll_timeout / 1000)
        try:
            super().exit()
        finally:
            if acquired:
                self._poller_lock.release()

Functions

receive

receive(timeout: float | None = 1000, raise_interrupt_as_exception: bool = False) -> EventMessage | None

Receive event with given timeout.

Parameters:

Name	Type	Description	Default
timeout	float | None	timeout in milliseconds, None for blocking	1000
raise_interrupt_as_exception	bool	if True, raises BreakLoop exception when interrupted, otherwise returns None	False

Returns:

Name	Type	Description
event_message	EventMessage | None	the received event, or None if polling was interrupted or timed out

Raises:

Type	Description
BreakLoop	if polling was stopped or interrupted and raise_interrupt_as_exception is True
ZMQError	if the poll or the receive failed for a reason other than the sockets being closed

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

def receive(self, timeout: float | None = 1000, raise_interrupt_as_exception: bool = False) -> EventMessage | None:
    """
    Receive event with given timeout.

    Parameters
    ----------
    timeout: float, int, None
        timeout in milliseconds, None for blocking
    raise_interrupt_as_exception: bool
        if True, raises BreakLoop exception when interrupted, otherwise returns None

    Returns
    -------
    event_message: EventMessage | None
        the received event, or None if polling was interrupted or timed out

    Raises
    ------
    BreakLoop
        if polling was stopped or interrupted and `raise_interrupt_as_exception` is True
    zmq.ZMQError
        if the poll or the receive failed for a reason other than the sockets being closed
    """
    while not self._stop:
        if not self._poller_lock.acquire(timeout=timeout / 1000 if timeout else -1):
            continue
        try:
            if self._stop or self.socket.closed:
                break
            sockets = self.poller.poll(timeout)  # ty: ignore[invalid-argument-type]  # list[tuple[zmq.Socket, int]]
            if len(sockets) > 1:
                # if there is an interrupt message as well as an event,
                # give preference to interrupt message.
                if sockets[0][0] == self.interruptor:
                    sockets = [sockets[0]]  # we still need the socket, poll event  tuple
                elif sockets[1][0] == self.interruptor:
                    sockets = [sockets[1]]
            for socket, _ in sockets:
                try:
                    raw_message = socket.recv_multipart(zmq.NOBLOCK)
                    message = EventMessage(raw_message)
                    if socket == self.interruptor:
                        if message.payload.deserialize() == "INTERRUPT":
                            self.stop_polling()
                            break
                    return message
                except zmq.Again:
                    pass
                # if not self.handled_default_message_types(event_message):
        except zmq.ZMQError as ex:
            # the sockets or the context were closed under us, which is a teardown elsewhere and not
            # an error here - anything else is a real failure and belongs to the caller
            if ex.errno not in (zmq.ETERM, zmq.ENOTSOCK):
                raise
            self.stop_polling()
            break
        finally:
            self._poller_lock.release()
    if raise_interrupt_as_exception:
        raise BreakLoop("event consumer interrupted")
    return None

interrupt

interrupt()

Interrupts the event consumer.

Generally should be used for exiting this object if there is no poll period/infinite polling. Otherwise please use stop_polling().

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

def interrupt(self):
    """
    Interrupts the event consumer.

    Generally should be used for exiting this object if there is no poll period/infinite polling.
    Otherwise please use stop_polling().
    """
    self.interrupting_peer.send_multipart(self.interrupt_message.byte_array)

hololinked.server.zmq.brokers.AsyncEventConsumer

Bases: BaseEventConsumer, BaseAsyncZMQ

Async Event Consumer to be used inside async loops.

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

class AsyncEventConsumer(BaseEventConsumer, BaseAsyncZMQ):
    """Async Event Consumer to be used inside async loops."""

    _poller_lock: asyncio.Lock
    poller: zmq.asyncio.Poller

    async def receive(
        self,
        timeout: float | None = 1000,
        raise_interrupt_as_exception: bool = False,
    ) -> EventMessage | None:
        """
        Receive event with given timeout.

        Parameters
        ----------
        timeout: float, int, None
            timeout in milliseconds, None for blocking
        raise_interrupt_as_exception: bool
            if True, raises BreakLoop exception when interrupted, otherwise returns None

        Returns
        -------
        event_message: EventMessage | None
            the received event, or None if polling was interrupted or timed out

        Raises
        ------
        BreakLoop
            if polling was stopped or interrupted and `raise_interrupt_as_exception` is True
        zmq.ZMQError
            if the poll or the receive failed for a reason other than the sockets being closed
        """
        while not self._stop:
            try:
                await asyncio.wait_for(
                    self._poller_lock.acquire(),
                    timeout=timeout / 1000 if timeout else None,
                )
            except TimeoutError:
                continue
            try:
                if self._stop or self.socket.closed:
                    break
                sockets = await self.poller.poll(timeout)
                if len(sockets) > 1:
                    # if there is an interrupt message as well as an event,
                    # give preference to interrupt message.
                    if sockets[0][0] == self.interruptor:
                        sockets = [sockets[0]]
                    elif sockets[1][0] == self.interruptor:
                        sockets = [sockets[1]]
                for socket, _ in sockets:
                    try:
                        raw_message = await socket.recv_multipart(zmq.NOBLOCK)
                        message = EventMessage(raw_message)
                        if socket == self.interruptor:
                            if message.payload.deserialize() == "INTERRUPT":
                                self.stop_polling()
                                break
                        return message
                    except zmq.Again:
                        pass
            except zmq.ZMQError as ex:
                # see the note in the sync consumer - a closed socket or context is a teardown, not a failure
                if ex.errno not in (zmq.ETERM, zmq.ENOTSOCK):
                    raise
                self.stop_polling()
                break
            finally:
                self._poller_lock.release()
        if raise_interrupt_as_exception:
            raise BreakLoop("event consumer interrupted")
        return None

    async def interrupt(self):
        """
        Interrupts the event consumer.

        Generally should be used for exiting this object if there is no poll period/infinite polling.
        Otherwise please use stop_polling().
        """
        await self.interrupting_peer.send_multipart(self.interrupt_message.byte_array)

Functions

receive async

receive(timeout: float | None = 1000, raise_interrupt_as_exception: bool = False) -> EventMessage | None

Receive event with given timeout.

Parameters:

Name	Type	Description	Default
timeout	float | None	timeout in milliseconds, None for blocking	1000
raise_interrupt_as_exception	bool	if True, raises BreakLoop exception when interrupted, otherwise returns None	False

Returns:

Name	Type	Description
event_message	EventMessage | None	the received event, or None if polling was interrupted or timed out

Raises:

Type	Description
BreakLoop	if polling was stopped or interrupted and raise_interrupt_as_exception is True
ZMQError	if the poll or the receive failed for a reason other than the sockets being closed

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

async def receive(
    self,
    timeout: float | None = 1000,
    raise_interrupt_as_exception: bool = False,
) -> EventMessage | None:
    """
    Receive event with given timeout.

    Parameters
    ----------
    timeout: float, int, None
        timeout in milliseconds, None for blocking
    raise_interrupt_as_exception: bool
        if True, raises BreakLoop exception when interrupted, otherwise returns None

    Returns
    -------
    event_message: EventMessage | None
        the received event, or None if polling was interrupted or timed out

    Raises
    ------
    BreakLoop
        if polling was stopped or interrupted and `raise_interrupt_as_exception` is True
    zmq.ZMQError
        if the poll or the receive failed for a reason other than the sockets being closed
    """
    while not self._stop:
        try:
            await asyncio.wait_for(
                self._poller_lock.acquire(),
                timeout=timeout / 1000 if timeout else None,
            )
        except TimeoutError:
            continue
        try:
            if self._stop or self.socket.closed:
                break
            sockets = await self.poller.poll(timeout)
            if len(sockets) > 1:
                # if there is an interrupt message as well as an event,
                # give preference to interrupt message.
                if sockets[0][0] == self.interruptor:
                    sockets = [sockets[0]]
                elif sockets[1][0] == self.interruptor:
                    sockets = [sockets[1]]
            for socket, _ in sockets:
                try:
                    raw_message = await socket.recv_multipart(zmq.NOBLOCK)
                    message = EventMessage(raw_message)
                    if socket == self.interruptor:
                        if message.payload.deserialize() == "INTERRUPT":
                            self.stop_polling()
                            break
                    return message
                except zmq.Again:
                    pass
        except zmq.ZMQError as ex:
            # see the note in the sync consumer - a closed socket or context is a teardown, not a failure
            if ex.errno not in (zmq.ETERM, zmq.ENOTSOCK):
                raise
            self.stop_polling()
            break
        finally:
            self._poller_lock.release()
    if raise_interrupt_as_exception:
        raise BreakLoop("event consumer interrupted")
    return None

interrupt async

interrupt()

Interrupts the event consumer.

Generally should be used for exiting this object if there is no poll period/infinite polling. Otherwise please use stop_polling().

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

async def interrupt(self):
    """
    Interrupts the event consumer.

    Generally should be used for exiting this object if there is no poll period/infinite polling.
    Otherwise please use stop_polling().
    """
    await self.interrupting_peer.send_multipart(self.interrupt_message.byte_array)