hololinked.client.abstractions.ConsumedThingEvent

class ConsumedThingEvent:  # noqa: D101 # Dont add class doc otherwise it will conflict with __doc__ in slots
    # event subscription

    if TYPE_CHECKING:
        # These are declared as __slots__ in the protocol-specific mixin subclasses.
        # Annotated here only so type checkers resolve them on the base type.
        __name__: str
        __qualname__: str
        __doc__: str | None

    def __init__(
        self,
        resource: EventAffordance | PropertyAffordance,
        logger: structlog.stdlib.BoundLogger,
        owner_inst: Any,
    ) -> None:
        """
        Initialize a consumed thing event.

        Parameters
        ----------
        resource: EventAffordance | PropertyAffordance
            dataclass object representing the event or an observable property
        logger: structlog.stdlib.BoundLogger
            logger instance
        owner_inst: Any
            the parent object that owns this event
        """
        from . import ObjectProxy  # noqa: F401

        self.resource = resource
        self.logger = logger
        self.owner_inst = owner_inst  # type: ObjectProxy
        self._subscribed = {}
        # self._sync_callbacks = []
        # self._async_callbacks = []

    def subscribe(
        self,
        callbacks: list[Callable] | Callable,
        asynch: bool = False,
        concurrent: bool = False,
        deserialize: bool = True,
        # create_new_connection: bool = False,
    ) -> None:
        """
        Subscribe to the event.

        Parameters
        ----------
        callbacks: list[Callable] | Callable
            callback or list of callbacks to add
        asynch: bool
            whether to start an async(-io task) event listener instead of a threaded listener
        concurrent: bool
            - asyncio - if `True`, each callback is scheduled as a separate task, if `False` they are awaited sequentially.
            - threading - if `True`, each callback is called in a separate thread, if `False` they are called sequentially.
        deserialize: bool
            if `False`, event payload is passed to the callbacks as raw bytes, if `True` it is deserialized

        Raises
        ------
        ValueError
            if no form is found for the event subscription
        """
        op = Operations.observeproperty if isinstance(self.resource, PropertyAffordance) else Operations.subscribeevent
        form = self.resource.retrieve_form(op, None)
        cbs = (
            callbacks
            if isinstance(callbacks, list)
            else list(callbacks)
            if isinstance(callbacks, tuple)
            else [callbacks]
        )
        # if not create_new_connection:
        #   see tag v0.3.2 for logic
        if form is None:
            raise ValueError(f"No form found for {op} operation for {self.resource.name}")
        if asynch:
            get_current_async_loop().call_soon(
                lambda: asyncio.create_task(self.async_listen(form, cbs, concurrent, deserialize))  # type: ignore
            )
        else:
            _thread = threading.Thread(
                target=self.listen,
                args=(form, cbs, concurrent, deserialize),
                daemon=True,
            )
            _thread.start()

    def unsubscribe(self) -> None:
        """Unsubscribe from the event."""
        self._subscribed.clear()
        # self._sync_callbacks.clear()
        # self._async_callbacks.clear()

    def listen(self, form: Form, callbacks: list[Callable], concurrent: bool = True, deserialize: bool = True) -> None:
        """
        Listen to events and call the callbacks in threaded mode.

        This method needs to be invoked by the `subscribe()` method. Use `async_listen()` for asyncio mode.

        Parameters
        ----------
        form: Form
            form to use for event subscription
        callbacks: list[Callable]
            list of callbacks to call on event
        concurrent: bool
            whether to run each callback concurrently in a separate thread
        deserialize: bool
            whether to deserialize the event payload before passing to callbacks
        """
        raise NotImplementedError("implement listen per protocol")

    async def async_listen(
        self,
        form: Form,
        callbacks: list[Callable],
        concurrent: bool = True,
        deserialize: bool = True,
    ) -> None:
        """
        Listen to events and call the callbacks.

        This method needs to be invoked by the `subscribe()` method in asyncio mode. Use `listen()` for threaded mode.

        Parameters
        ----------
        form: Form
            form to use for event subscription
        callbacks: list[Callable]
            list of callbacks to call on event
        concurrent: bool
            whether to run each callback concurrently as separate tasks
        deserialize: bool
            whether to deserialize the event payload before passing to callbacks
        """
        raise NotImplementedError("implement async_listen per protocol")

    def schedule_callbacks(self, callbacks: list[Callable], event_data: Any, concurrent: bool = False) -> None:
        """
        Schedule the callbacks to be called with the event data.

        Parameters
        ----------
        callbacks: list[Callable]
            list of callbacks to call
        event_data: Any
            event data to pass to the callbacks
        concurrent: bool
            whether to run each callback in a separate thread concurrently
        """
        for cb in callbacks:
            try:
                if not concurrent:
                    cb(event_data)
                else:
                    threading.Thread(target=cb, args=(event_data,)).start()
            except Exception as ex:
                self.logger.error(f"Error occurred in callback {cb} - {ex}", exc_info=True)

    async def async_schedule_callbacks(self, callbacks, event_data: Any, concurrent: bool = False) -> None:
        """
        Async schedule the callbacks to be called with the event data.

        Parameters
        ----------
        callbacks: list[Callable]
            list of callbacks to call
        event_data: Any
            event data to pass to the callbacks
        concurrent: bool
            whether to run each callback in a separate asyncio task concurrently
        """
        loop = get_current_async_loop()
        for cb in callbacks:
            try:
                if concurrent:
                    if asyncio.iscoroutinefunction(cb):
                        loop.create_task(cb(event_data))
                    else:
                        loop.run_in_executor(None, cb, event_data)
                elif asyncio.iscoroutinefunction(cb):
                    await cb(event_data)
                else:
                    cb(event_data)
            except Exception as ex:
                self.logger.error(f"Error occurred in callback {cb} - {ex}", exc_info=True)

    def add_callbacks(self, callbacks: list[Callable] | Callable, asynch: bool = False) -> None:
        """
        Add callbacks to the event.

        Parameters
        ----------
        *callbacks: list[Callable] | Callable
            callback or list of callbacks to add
        """
        # for logic, see tag v0.3.2
        raise NotImplementedError(
            "cannot add callbacks currrently to reuse event subscription."
            + " Unsubscribe and resubscribe with new callbacks"
        )