Bluez notify read fix

CHANGELOG.rst

@@ -18,6 +18,7 @@ Added
 Changed
 -------
 * Raise new ``BleakBluetoothNotAvailableError`` when Bluetooth is not supported, turned off or permission is denied.
+* Use AcquireNotify rather than StartNotify for Linux backend on supported characteristics
 
 Fixed
 -----

bleak/backends/bluezdbus/client.py

@@ -103,6 +103,7 @@ def __init__(
         self._disconnect_monitor_event: Optional[asyncio.Event] = None
         # map of characteristic D-Bus object path to notification callback
         self._notification_callbacks: dict[str, NotifyCallback] = {}
+        self._notification_fds: dict[str, int] = {}
 
         # used to override mtu_size property
         self._mtu_size: Optional[int] = None
@@ -205,6 +206,17 @@ def on_value_changed(char_path: str, value: bytes) -> None:
                     stack.callback(local_disconnect_monitor_event.set)
 
                     async def disconnect_device() -> None:
+                        # Clean up open notification file descriptors
+                        loop = asyncio.get_running_loop()
+                        for fd in self._notification_fds.items():
+                            try:
+                                loop.remove_reader(fd)
+                                os.close(fd)
+                            except Exception as e:
+                                logger.error(
+                                    "Failed to remove file descriptor %d: %d", fd, e
+                                )
+
                         # Calling Disconnect cancels any pending connect request. Also,
                         # if connection was successful but _get_services() raises (e.g.
                         # because task was cancelled), then we still need to disconnect
@@ -904,6 +916,29 @@ async def write_gatt_descriptor(
             "Write Descriptor %s | %s: %s", descriptor.handle, descriptor.obj[0], data
         )
 
+    def _read_notify_fd(self, fd, callback):
+        loop = asyncio.get_running_loop()
+
+        def on_data():
+            try:
+                data = os.read(fd, 1024)
+                if not data:
+                    raise RuntimeError("Unexpected EOF on notification file handle")
+                callback(bytes(data))
+            except Exception as e:
+                logger.error(
+                    "AcquireNotify: Read error on fd %d: %s. Notifications have been stopped.",
+                    fd,
+                    e,
+                )
+                try:
+                    loop.remove_reader(fd)
+                    os.close(fd)
+                except OSError:
+                    pass
+
+        loop.add_reader(fd, on_data)
+
     @override
     async def start_notify(
         self,
@@ -914,20 +949,44 @@ async def start_notify(
         """
         Activate notifications/indications on a characteristic.
         """
-        self._notification_callbacks[characteristic.obj[0]] = callback
-
         assert self._bus is not None
 
-        reply = await self._bus.call(
-            Message(
-                destination=defs.BLUEZ_SERVICE,
-                path=characteristic.obj[0],
-                interface=defs.GATT_CHARACTERISTIC_INTERFACE,
-                member="StartNotify",
+        # If using StartNotify and calling a read on the same
+        # characteristic, BlueZ will return the response as
+        # both a notification and read, duplicating the message.
+        # Using AcquireNotify on supported characteristics avoids this.
+        # However, using the preferred AcquireNotify requires that devices
+        # correctly indicate "notify" and/or "indicate" properties. If they
+        # don't, we fall back to StartNotify.
+        if "NotifyAcquired" in characteristic.obj[1]:
+            reply = await self._bus.call(
+                Message(
+                    destination=defs.BLUEZ_SERVICE,
+                    path=characteristic.obj[0],
+                    interface=defs.GATT_CHARACTERISTIC_INTERFACE,
+                    member="AcquireNotify",
+                    body=[{}],
+                    signature="a{sv}",
+                )
             )
-        )
-        assert reply
-        assert_reply(reply)
+            assert reply
+            assert_reply(reply)
+
+            unix_fd = reply.unix_fds[0]
+            self._notification_fds[characteristic.obj[0]] = unix_fd
+            self._read_notify_fd(unix_fd, callback)
+        else:
+            self._notification_callbacks[characteristic.obj[0]] = callback
+            reply = await self._bus.call(
+                Message(
+                    destination=defs.BLUEZ_SERVICE,
+                    path=characteristic.obj[0],
+                    interface=defs.GATT_CHARACTERISTIC_INTERFACE,
+                    member="StartNotify",
+                )
+            )
+            assert reply
+            assert_reply(reply)
 
     @override
     async def stop_notify(self, characteristic: BleakGATTCharacteristic) -> None:
@@ -942,15 +1001,24 @@ async def stop_notify(self, characteristic: BleakGATTCharacteristic) -> None:
 
         assert self._bus is not None
 
-        reply = await self._bus.call(
-            Message(
-                destination=defs.BLUEZ_SERVICE,
-                path=characteristic.obj[0],
-                interface=defs.GATT_CHARACTERISTIC_INTERFACE,
-                member="StopNotify",
+        if "NotifyAcquired" in characteristic.obj[1]:
+            fd = self._notification_fds.pop(characteristic.obj[0])
+            if fd:
+                loop = asyncio.get_running_loop()
+                try:
+                    loop.remove_reader(fd)
+                    os.close(fd)
+                except Exception as e:
+                    logger.error("Failed to remove file descriptor %d: %d", fd, e)
+        else:
+            reply = await self._bus.call(
+                Message(
+                    destination=defs.BLUEZ_SERVICE,
+                    path=characteristic.obj[0],
+                    interface=defs.GATT_CHARACTERISTIC_INTERFACE,
+                    member="StopNotify",
+                )
             )
-        )
-        assert reply
-        assert_reply(reply)
-
-        self._notification_callbacks.pop(characteristic.obj[0], None)
+            assert reply
+            assert_reply(reply)
+            self._notification_callbacks.pop(characteristic.obj[0], None)