hololinked.server.zmq.brokers.EventPublisher

Bases: BaseZMQServer, BaseSyncZMQ

Event publisher for broadcasting messages to all connected clients. Implements PUB-SUB pattern.

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

class EventPublisher(BaseZMQServer, BaseSyncZMQ):
    """Event publisher for broadcasting messages to all connected clients. Implements PUB-SUB pattern."""

    _standard_address_suffix = "/event-publisher"
    _standard_address_suffix_filename_replacement = "event-publisher"

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

        Parameters
        ----------
        id: str
            unique identifier of the publisher
        context: zmq.Context
            ZMQ context to use, if None, a global context is used.
        access_point: Enum | str, default ZMQ_TRANSPORTS.IPC
            access point for the publisher to bind to, usually `INPROC`
        """
        super().__init__(id=id, **kwargs)
        self.create_socket(
            server_id=id,
            socket_id=id,
            node_type="server",
            context=context,
            access_point=access_point,
            socket_type=zmq.SocketType.PUB,
            **kwargs,
        )
        self._send_lock = threading.Lock()

    def publish(self, event, data: Any) -> None:
        """
        Encode one event and put it on the PUB socket.

        Parameters
        ----------
        event: `RegisteredEvent`
            the event being published, as the bus registered it
        data: Any
            its payload, unencoded. `bytes` bypass serialization and travel as the preserialized frame.
        """
        # uncomment for type definitions
        # from ...core.eventloop import RegisteredEvent
        # assert isinstance(event, RegisteredEvent), "event must be an instance of RegisteredEvent"

        try:
            self._send_lock.acquire()
            serializer = Serializers.for_object(
                event.owner.id,
                event.owner.__class__.__name__,
                event.descriptor.name,
            )
            content_type_if_no_serializer = Serializers.get_content_type_for_object(
                event.owner.id,
                event.owner.__class__.__name__,
                event.descriptor.name,
            )
            if not isinstance(data, bytes):
                payload = SerializableData(data, serializer=serializer)
                preserialized_payload = PreserializedEmptyByte
            else:
                payload = SerializableNone
                preserialized_payload = PreserializedData(data, content_type=content_type_if_no_serializer)

            event_message = EventMessage.craft_from_arguments(
                event.unique_identifier,
                self.id,
                payload=payload,
                preserialized_payload=preserialized_payload,
            )
            self.socket.send_multipart(event_message.byte_array)
            self.logger.debug(f"published event with unique identifier {event.unique_identifier}")
        finally:
            try:
                self._send_lock.release()
            except Exception as ex:
                self.logger.warning(f"could not release publish lock for event publisher - {str(ex)}")

    def exit(self):
        # the send lock is what keeps the close off a socket that another thread is publishing on
        acquired = self._send_lock.acquire(timeout=5)
        try:
            BaseZMQ.exit(self)
            self.socket.close(0)
            self.logger.info("terminated event publishing socket")
        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)}"
            )
        finally:
            if acquired:
                self._send_lock.release()

Functions

__init__

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

Initialize the publisher.

Parameters:

Name	Type	Description	Default
id	str	unique identifier of the publisher	required
context	Context | None	ZMQ context to use, if None, a global context is used.	None
access_point	str	access point for the publisher to bind to, usually INPROC	IPC

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

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

    Parameters
    ----------
    id: str
        unique identifier of the publisher
    context: zmq.Context
        ZMQ context to use, if None, a global context is used.
    access_point: Enum | str, default ZMQ_TRANSPORTS.IPC
        access point for the publisher to bind to, usually `INPROC`
    """
    super().__init__(id=id, **kwargs)
    self.create_socket(
        server_id=id,
        socket_id=id,
        node_type="server",
        context=context,
        access_point=access_point,
        socket_type=zmq.SocketType.PUB,
        **kwargs,
    )
    self._send_lock = threading.Lock()

publish

publish(event, data: Any) -> None

Encode one event and put it on the PUB socket.

Parameters:

Name	Type	Description	Default
event		the event being published, as the bus registered it	required
data	Any	its payload, unencoded. bytes bypass serialization and travel as the preserialized frame.	required

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

def publish(self, event, data: Any) -> None:
    """
    Encode one event and put it on the PUB socket.

    Parameters
    ----------
    event: `RegisteredEvent`
        the event being published, as the bus registered it
    data: Any
        its payload, unencoded. `bytes` bypass serialization and travel as the preserialized frame.
    """
    # uncomment for type definitions
    # from ...core.eventloop import RegisteredEvent
    # assert isinstance(event, RegisteredEvent), "event must be an instance of RegisteredEvent"

    try:
        self._send_lock.acquire()
        serializer = Serializers.for_object(
            event.owner.id,
            event.owner.__class__.__name__,
            event.descriptor.name,
        )
        content_type_if_no_serializer = Serializers.get_content_type_for_object(
            event.owner.id,
            event.owner.__class__.__name__,
            event.descriptor.name,
        )
        if not isinstance(data, bytes):
            payload = SerializableData(data, serializer=serializer)
            preserialized_payload = PreserializedEmptyByte
        else:
            payload = SerializableNone
            preserialized_payload = PreserializedData(data, content_type=content_type_if_no_serializer)

        event_message = EventMessage.craft_from_arguments(
            event.unique_identifier,
            self.id,
            payload=payload,
            preserialized_payload=preserialized_payload,
        )
        self.socket.send_multipart(event_message.byte_array)
        self.logger.debug(f"published event with unique identifier {event.unique_identifier}")
    finally:
        try:
            self._send_lock.release()
        except Exception as ex:
            self.logger.warning(f"could not release publish lock for event publisher - {str(ex)}")