remove EventListener

deebot_client/events/event_bus.py

@@ -17,42 +17,18 @@
 T = TypeVar("T", bound=Event)
 
 
-class EventListener(Generic[T]):
-    """Object that allows event consumers to easily unsubscribe from event bus."""
-
-    def __init__(
-        self,
-        event_processing_data: "_EventProcessingData",
-        callback: Callable[[T], Coroutine[Any, Any, None]],
-    ) -> None:
-        self._event_processing_data: Final = event_processing_data
-        self.callback: Final = callback
-
-    def unsubscribe(self) -> None:
-        """Unsubscribe from event bus."""
-        self._event_processing_data.subscribers.remove(self)
-
-
 class _EventProcessingData(Generic[T]):
     """Data class, which holds all needed data per EventDto."""
 
     def __init__(self) -> None:
         super().__init__()
 
-        self._subscribers: Final[list[EventListener[T]]] = []
-        self._semaphore: Final = asyncio.Semaphore(1)
+        self.subscriber_callbacks: Final[
+            list[Callable[[T], Coroutine[Any, Any, None]]]
+        ] = []
+        self.semaphore: Final = asyncio.Semaphore(1)
         self.last_event: T | None = None
 
-    @property
-    def subscribers(self) -> list[EventListener[T]]:
-        """Return subscribers."""
-        return self._subscribers
-
-    @property
-    def semaphore(self) -> asyncio.Semaphore:
-        """Return semaphore."""
-        return self._semaphore
-
 
 class EventBus:
     """A very simple event bus system."""
@@ -70,7 +46,7 @@ def __init__(
     def has_subscribers(self, event: type[T]) -> bool:
         """Return True, if emitter has subscribers."""
         return (
-            len(self._event_processing_dict[event].subscribers) > 0
+            len(self._event_processing_dict[event].subscriber_callbacks) > 0
             if event in self._event_processing_dict
             else False
         )
@@ -79,23 +55,23 @@ def subscribe(
         self,
         event_type: type[T],
         callback: Callable[[T], Coroutine[Any, Any, None]],
-    ) -> EventListener[T]:
+    ) -> Callable[[], None]:
         """Subscribe to event."""
         event_processing_data = self._get_or_create_event_processing_data(event_type)
 
-        listener = EventListener(event_processing_data, callback)
-        event_processing_data.subscribers.append(listener)
+        def unsubscribe() -> None:
+            event_processing_data.subscriber_callbacks.remove(callback)
+
+        event_processing_data.subscriber_callbacks.append(callback)
 
         if event_processing_data.last_event:
             # Notify subscriber directly with the last event
-            create_task(
-                self._tasks, listener.callback(event_processing_data.last_event)
-            )
-        elif len(event_processing_data.subscribers) == 1:
+            create_task(self._tasks, callback(event_processing_data.last_event))
+        elif len(event_processing_data.subscriber_callbacks) == 1:
             # first subscriber therefore do refresh
             self.request_refresh(event_type)
 
-        return listener
+        return unsubscribe
 
     def notify(self, event: T) -> bool:
         """Notify subscriber with given event representation."""
@@ -126,10 +102,10 @@ def notify(self, event: T) -> bool:
             return False
 
         event_processing_data.last_event = event
-        if event_processing_data.subscribers:
+        if event_processing_data.subscriber_callbacks:
             _LOGGER.debug("Notify subscribers with %s", event)
-            for subscriber in event_processing_data.subscribers:
-                create_task(self._tasks, subscriber.callback(event))
+            for callback in event_processing_data.subscriber_callbacks:
+                create_task(self._tasks, callback(event))
             return True
 
         _LOGGER.debug("No subscribers... Discharging %s", event)

deebot_client/map.py

@@ -28,7 +28,7 @@
     PositionType,
     RoomsEvent,
 )
-from .events.event_bus import EventBus, EventListener
+from .events.event_bus import EventBus
 from .exceptions import MapError
 from .logging_filter import get_logger
 from .models import Room
@@ -143,38 +143,9 @@
         self._map_data: Final[MapData] = MapData()
         self._amount_rooms: int = 0
         self._last_image: LastImage | None = None
-        self._listeners: list[EventListener] = []
+        self._unsubscribers: list[Callable[[], None]] = []
         self._tasks: set[asyncio.Future[Any]] = set()
 
-        async def on_map_set(event: MapSetEvent) -> None:
-            if event.type == MapSetType.ROOMS:
-                self._amount_rooms = len(event.subsets)
-                for room_id, _ in self._map_data.rooms.copy().items():
-                    if room_id not in event.subsets:
-                        self._map_data.rooms.pop(room_id, None)
-            else:
-                for subset_id, subset in self._map_data.map_subsets.copy().items():
-                    if subset.type == event.type and subset_id not in event.subsets:
-                        self._map_data.map_subsets.pop(subset_id, None)
-
-        self._event_bus.subscribe(MapSetEvent, on_map_set)
-
-        async def on_map_subset(event: MapSubsetEvent) -> None:
-            if event.type == MapSetType.ROOMS and event.name:
-                room = Room(event.name, event.id, event.coordinates)
-                if self._map_data.rooms.get(event.id, None) != room:
-                    self._map_data.rooms[room.id] = room
-
-                    if len(self._map_data.rooms) == self._amount_rooms:
-                        self._event_bus.notify(
-                            RoomsEvent(list(self._map_data.rooms.values()))
-                        )
-
-            elif self._map_data.map_subsets.get(event.id, None) != event:
-                self._map_data.map_subsets[event.id] = event
-
-        self._event_bus.subscribe(MapSubsetEvent, on_map_subset)
-
     # ---------------------------- METHODS ----------------------------
 
     def _update_trace_points(self, data: str) -> None:
@@ -227,23 +198,58 @@
 
     def enable(self) -> None:
         """Enable map."""
-        if self._listeners:
+        if self._unsubscribers:
             return
 
         create_task(self._tasks, self._execute_command(GetCachedMapInfo()))
 
+        async def on_map_set(event: MapSetEvent) -> None:
+            if event.type == MapSetType.ROOMS:
+                self._amount_rooms = len(event.subsets)
+                for room_id, _ in self._map_data.rooms.copy().items():
+                    if room_id not in event.subsets:
+                        self._map_data.rooms.pop(room_id, None)
+            else:
+                for subset_id, subset in self._map_data.map_subsets.copy().items():
+                    if subset.type == event.type and subset_id not in event.subsets:
+                        self._map_data.map_subsets.pop(subset_id, None)
+
+        self._unsubscribers.append(self._event_bus.subscribe(MapSetEvent, on_map_set))
+
+        async def on_map_subset(event: MapSubsetEvent) -> None:
+            if event.type == MapSetType.ROOMS and event.name:
+                room = Room(event.name, event.id, event.coordinates)
+                if self._map_data.rooms.get(event.id, None) != room:
+                    self._map_data.rooms[room.id] = room
+
+                    if len(self._map_data.rooms) == self._amount_rooms:
+                        self._event_bus.notify(
+                            RoomsEvent(list(self._map_data.rooms.values()))
+                        )
+
+            elif self._map_data.map_subsets.get(event.id, None) != event:
+                self._map_data.map_subsets[event.id] = event
+
+        self._unsubscribers.append(
+            self._event_bus.subscribe(MapSubsetEvent, on_map_subset)
+        )
+
         async def on_position(event: PositionsEvent) -> None:
             self._map_data.positions = event.positions
 
-        self._listeners.append(self._event_bus.subscribe(PositionsEvent, on_position))
+        self._unsubscribers.append(
+            self._event_bus.subscribe(PositionsEvent, on_position)
+        )
 
         async def on_map_trace(event: MapTraceEvent) -> None:
             if event.start == 0:
                 self._map_data.trace_values.clear()
 
             self._update_trace_points(event.data)
 
-        self._listeners.append(self._event_bus.subscribe(MapTraceEvent, on_map_trace))
+        self._unsubscribers.append(
+            self._event_bus.subscribe(MapTraceEvent, on_map_trace)
+        )
 
         async def on_major_map(event: MajorMapEvent) -> None:
             tasks = []
@@ -263,23 +269,27 @@
             if tasks:
                 await asyncio.gather(*tasks)
 
-        self._listeners.append(self._event_bus.subscribe(MajorMapEvent, on_major_map))
+        self._unsubscribers.append(
+            self._event_bus.subscribe(MajorMapEvent, on_major_map)
+        )
 
         async def on_minor_map(event: MinorMapEvent) -> None:
             self._map_data.map_pieces[event.index].update_points(event.value)
 
-        self._listeners.append(self._event_bus.subscribe(MinorMapEvent, on_minor_map))
+        self._unsubscribers.append(
+            self._event_bus.subscribe(MinorMapEvent, on_minor_map)
+        )
 
     def disable(self) -> None:
         """Disable map."""
-        listeners = self._listeners
-        self._listeners.clear()
-        for listener in listeners:
-            listener.unsubscribe()
+        unsubscribers = self._unsubscribers
+        self._unsubscribers.clear()
+        for unsubscribe in unsubscribers:
+            unsubscribe()
 
     def refresh(self) -> None:
         """Manually refresh map."""
-        if not self._listeners:
+        if not self._unsubscribers:
             raise MapError("Please enable the map first")
 
         # todo make it nice pylint: disable=fixme
@@ -289,7 +299,7 @@
 
     def get_base64_map(self, width: int | None = None) -> bytes:
         """Return map as base64 image string."""
-        if not self._listeners:
+        if not self._unsubscribers:
             raise MapError("Please enable the map first")
 
         if (

tests/conftest.py

@@ -1,13 +1,14 @@
 import logging
 from collections.abc import AsyncGenerator, Generator
-from unittest.mock import Mock
+from unittest.mock import AsyncMock, Mock
 
 import aiohttp
 import pytest
 from asyncio_mqtt import Client
 
 from deebot_client.api_client import ApiClient
 from deebot_client.authentication import Authenticator
+from deebot_client.events.event_bus import EventBus
 from deebot_client.models import Configuration, Credentials, DeviceInfo
 from deebot_client.mqtt_client import MqttClient, MqttConfiguration
 from deebot_client.vacuum_bot import VacuumBot
@@ -126,3 +127,13 @@ async def vacuum_bot(
     await bot.initialize(mqtt)
     yield bot
     await bot.teardown()
+
+
+@pytest.fixture
+def execute_mock() -> AsyncMock:
+    return AsyncMock()
+
+
+@pytest.fixture
+def event_bus(execute_mock: AsyncMock) -> EventBus:
+    return EventBus(execute_mock)