Event Stream Service

EventStreamService — owns the anyio memory channel for event routing.

EventStreamService

Bases: Resource

Owns the anyio memory channel that routes events from producers to the bus.

This is a Resource (no background task). It creates the send/receive streams at construction time and tears them down on shutdown.

Source code in src/hassette/core/event_stream_service.py

class EventStreamService(Resource):
    """Owns the anyio memory channel that routes events from producers to the bus.

    This is a Resource (no background task). It creates the send/receive streams
    at construction time and tears them down on shutdown.
    """

    _send_stream: "MemoryObjectSendStream[Event[Any]]"
    _receive_stream: "MemoryObjectReceiveStream[Event[Any]]"

    def __init__(self, hassette: "Hassette", *, parent: "Resource | None" = None) -> None:
        super().__init__(hassette, parent=parent)
        buffer_size = hassette.config.hassette_event_buffer_size
        self._send_stream, self._receive_stream = create_memory_object_stream["Event[Any]"](buffer_size)

    async def on_initialize(self) -> None:
        """Signal readiness — streams are created synchronously in __init__."""
        self.mark_ready(reason="EventStreamService initialized")

    @property
    def config_log_level(self) -> LOG_LEVEL_TYPE:
        return self.hassette.config.logging.bus_service

    @property
    def receive_stream(self) -> "MemoryObjectReceiveStream[Event[Any]]":
        """The receive end of the event stream, for BusService to clone."""
        return self._receive_stream

    async def send_event(self, event: "Event[Any]") -> None:
        """Send an event to the bus via the memory channel."""
        await self._send_stream.send(event)

    @property
    def event_streams_closed(self) -> bool:
        """Check if both streams are closed."""
        return self._send_stream._closed and self._receive_stream._closed  # pyright: ignore[reportPrivateUsage]

    async def close_streams(self) -> None:
        """Close both streams. Called by Hassette._on_children_stopped() after all children have stopped.

        Idempotent — safe to call multiple times (Hassette.shutdown() finally block
        calls this as a fallback on timeout paths).
        """
        if self.event_streams_closed:
            return
        await self._send_stream.aclose()
        await self._receive_stream.aclose()

receive_stream: MemoryObjectReceiveStream[Event[Any]] property

The receive end of the event stream, for BusService to clone.

event_streams_closed: bool property

Check if both streams are closed.

on_initialize() -> None async

Signal readiness — streams are created synchronously in init.

Source code in src/hassette/core/event_stream_service.py

async def on_initialize(self) -> None:
    """Signal readiness — streams are created synchronously in __init__."""
    self.mark_ready(reason="EventStreamService initialized")

send_event(event: Event[Any]) -> None async

Send an event to the bus via the memory channel.

Source code in src/hassette/core/event_stream_service.py

async def send_event(self, event: "Event[Any]") -> None:
    """Send an event to the bus via the memory channel."""
    await self._send_stream.send(event)

close_streams() -> None async

Close both streams. Called by Hassette._on_children_stopped() after all children have stopped.

Idempotent — safe to call multiple times (Hassette.shutdown() finally block calls this as a fallback on timeout paths).

Source code in src/hassette/core/event_stream_service.py

async def close_streams(self) -> None:
    """Close both streams. Called by Hassette._on_children_stopped() after all children have stopped.

    Idempotent — safe to call multiple times (Hassette.shutdown() finally block
    calls this as a fallback on timeout paths).
    """
    if self.event_streams_closed:
        return
    await self._send_stream.aclose()
    await self._receive_stream.aclose()