bluezdbus/scanner: Detect AdvertisementMonitor not registering

Author: bojanpotocnik

bleak/backends/bluezdbus/advertisement_monitor.py

@@ -34,7 +34,7 @@ class OrPattern(NamedTuple):
 OrPatternLike = Union[OrPattern, Tuple[int, AdvertisementDataType, bytes]]
 
 
-ReleasedCallback = Callable[[], None]
+StatusCallback = Callable[[bool], None]
 
 
 class AdvertisementMonitor(ServiceInterface):
@@ -52,28 +52,30 @@ class AdvertisementMonitor(ServiceInterface):
     """
 
     def __init__(
-        self, or_patterns: Iterable[OrPatternLike], released_callback: ReleasedCallback
+        self, or_patterns: Iterable[OrPatternLike], status_callback: StatusCallback
     ):
         """
         Args:
             or_patterns:
                 List of or patterns that will be returned by the ``Patterns`` property.
-            released_callback:
-                A callback that is called when the D-bus "Release" method is called.
+            status_callback:
+                A callback that is called with argument ``True`` when the D-bus "Activate"
+                method is called, or with ``False`` when "Release" is called.
         """
         super().__init__(defs.ADVERTISEMENT_MONITOR_INTERFACE)
         # dbus_fast marshaling requires list instead of tuple
         self._or_patterns = [list(p) for p in or_patterns]
-        self._released_callback = released_callback
+        self._status_callback = status_callback
 
     @method()
     def Release(self):
         logger.debug("Release")
-        self._released_callback()
+        self._status_callback(False)
 
     @method()
     def Activate(self):
         logger.debug("Activate")
+        self._status_callback(True)
 
     # REVISIT: mypy is broke, so we have to add redundant @no_type_check
     # https://github.com/python/mypy/issues/6583

bleak/backends/bluezdbus/manager.py

@@ -9,6 +9,7 @@
 import asyncio
 import logging
 import os
+import sys
 from typing import (
     Any,
     Callable,
@@ -24,6 +25,11 @@
 )
 from weakref import WeakKeyDictionary
 
+if sys.version_info < (3, 11):
+    from async_timeout import timeout as async_timeout
+else:
+    from asyncio import timeout as async_timeout
+
 from dbus_fast import BusType, Message, MessageType, Variant, unpack_variants
 from dbus_fast.aio.message_bus import MessageBus
 
@@ -441,7 +447,6 @@ async def passive_scan(
         filters: List[OrPatternLike],
         advertisement_callback: AdvertisementCallback,
         device_removed_callback: DeviceRemovedCallback,
-        discovery_stopped_callback: DiscoveryStoppedCallback,
     ) -> Callable[[], Coroutine]:
         """
         Configures the advertisement data filters and starts scanning.
@@ -472,12 +477,28 @@ async def passive_scan(
             )
             self._device_removed_callbacks.append(device_removed_callback_and_state)
 
+            # Once a monitoring job is activated by BlueZ, the client can expect to get notified
+            # on the targeted advertisements no matter if there is an ongoing discovery session
+            # (started/stopped with org.bluez.Adapter1.StartDiscovery/StopDiscovery, as done when
+            # using active scanning).
+            # That is why discovery_stopped_callback is not added to self._discovery_stopped_callbacks
+            # at this point, as org.bluez.Adapter1.Discovering status is not relevant for passive
+            # scanning.
+
+            # If advertisement monitor is released before the scanning is stopped, it means that the
+            # kernel does not support passive scanning and error was returned when trying to execute
+            # MGMT command "Add Adv Patterns Monitor" (see https://github.com/hbldh/bleak/issues/1136).
+            # Otherwise, monitor will be activated and start to receive advertisement packets.
+            monitor_processed = asyncio.Queue()
+
             try:
-                monitor = AdvertisementMonitor(filters, discovery_stopped_callback)
+                monitor = AdvertisementMonitor(filters, monitor_processed.put_nowait)
 
                 # this should be a unique path to allow multiple python interpreters
                 # running bleak and multiple scanners within a single interpreter
-                monitor_path = f"/org/bleak/{os.getpid()}/{id(monitor)}"
+                monitor_path = (
+                    f"/org/bleak/{os.getpid()}/{type(monitor).__name__}_{id(monitor)}"
+                )
 
                 reply = await self._bus.call(
                     Message(
@@ -529,14 +550,32 @@ async def stop():
                         )
                         assert_reply(reply)
 
-                return stop
+                try:
+                    # Advertising Monitor will be "immediately" activated or released
+                    async with async_timeout(1):
+                        if await monitor_processed.get():
+                            # Advertising Monitor has been activated
+                            return stop
+
+                except asyncio.TimeoutError:
+                    pass
+
+                # Do not call await stop() here as the bus is already locked
 
             except BaseException:
                 # if starting scanning failed, don't leak the callbacks
                 self._advertisement_callbacks.remove(callback_and_state)
                 self._device_removed_callbacks.remove(device_removed_callback_and_state)
                 raise
 
+        # Reaching here means that the Advertising Monitor has not been successfully activated
+        await stop()
+
+        raise BleakNoPassiveScanError(
+            "Advertising Monitor (required for passive scanning) is not supported by this kernel"
+            " (Linux kernel >= 5.10 is required)"
+        )
+
     def add_device_watcher(
         self,
         device_path: str,

bleak/backends/bluezdbus/scanner.py

@@ -186,7 +186,6 @@ async def start(self):
                 self._or_patterns,
                 self._handle_advertising_data,
                 self._handle_device_removed,
-                self.handle_early_stop,
             )
         else:
             self._stop = await manager.active_scan(