Merge pull request hbldh#193 from hbldh/release/0.6.2

Version 0.6.2

Better cleanup of Bluez notifications (merges hbldh#154 and fixes hbldh#130 )
Fix for read_gatt_char in Core Bluetooth (fixes hbldh#177)
Fix for is_disconnected in Core Bluetooth (merges hbldh#187 and fixes hbldh#185)
Added disconnection_callback functionality for Core Bluetooth (merges hbldh#186 and fixes hbldh#184)
Documentation fixes
Added requirements.txt

HISTORY.rst

@@ -2,6 +2,16 @@
 History
 =======
 
+0.6.2 (2020-05-15)
+------------------
+
+* Better cleanup of Bluez notifications (#154)
+* Fix for ``read_gatt_char`` in Core Bluetooth (#177)
+* Fix for ``is_disconnected`` in Core Bluetooth (#187 & #185)
+* Added ``disconnection_callback`` functionality for Core Bluetooth (#184 & #186)
+* Documentation fixes
+* Added ``requirements.txt``
+
 0.6.1 (2020-03-09)
 ------------------
 

bleak/__version__.py

@@ -1,3 +1,3 @@
 # -*- coding: utf-8 -*-
 
-__version__ = "0.6.1"
+__version__ = "0.6.2"

bleak/backends/bluezdbus/__init__.py

@@ -0,0 +1,28 @@
+from asyncio import AbstractEventLoop
+
+from twisted.internet.asyncioreactor import AsyncioSelectorReactor
+
+_reactors = {}
+
+
+def get_reactor(loop: AbstractEventLoop):
+    """Helper factory to get a Twisted reactor for the provided loop.
+
+    Since the AsyncioSelectorReactor on POSIX systems leaks file descriptors
+    even if stopped and presumably cleaned up, we lazily initialize them and
+    cache them for each loop. In a normal use case you will only work on one
+    event loop anyway, but in the case someone has different loops, this
+    construct still works without leaking resources.
+
+    Args:
+        loop (asyncio.events.AbstractEventLoop): The event loop to use.
+
+    Returns:
+           A :py:class:`twisted.internet.asnycioreactor.AsyncioSelectorReactor`
+           running on the provided asyncio event loop.
+
+    """
+    if loop not in _reactors:
+        _reactors[loop] = AsyncioSelectorReactor(loop)
+
+    return _reactors[loop]

bleak/backends/bluezdbus/characteristic.py

@@ -64,7 +64,9 @@ def descriptors(self) -> List:
         """List of descriptors for this service"""
         return self.__descriptors
 
-    def get_descriptor(self, _uuid: Union[str, UUID]) -> Union[BleakGATTDescriptor, None]:
+    def get_descriptor(
+        self, _uuid: Union[str, UUID]
+    ) -> Union[BleakGATTDescriptor, None]:
         """Get a descriptor by UUID"""
         try:
             return next(filter(lambda x: x.uuid == _uuid, self.descriptors))

bleak/backends/bluezdbus/client.py

@@ -6,21 +6,20 @@
 import subprocess
 import uuid
 from asyncio import Future
+from asyncio.events import AbstractEventLoop
 from functools import wraps, partial
 from typing import Callable, Any, Union
 
 from bleak.backends.service import BleakGATTServiceCollection
 from bleak.exc import BleakError
 from bleak.backends.client import BaseBleakClient
-from bleak.backends.bluezdbus import defs, signals, utils
+from bleak.backends.bluezdbus import defs, signals, utils, get_reactor
 from bleak.backends.bluezdbus.discovery import discover
 from bleak.backends.bluezdbus.utils import get_device_object_path, get_managed_objects
 from bleak.backends.bluezdbus.service import BleakGATTServiceBlueZDBus
 from bleak.backends.bluezdbus.characteristic import BleakGATTCharacteristicBlueZDBus
 from bleak.backends.bluezdbus.descriptor import BleakGATTDescriptorBlueZDBus
 
-from twisted.internet.asyncioreactor import AsyncioSelectorReactor
-from twisted.internet.error import ReactorNotRunning
 from txdbus.client import connect as txdbus_connect
 from txdbus.error import RemoteError
 
@@ -110,7 +109,7 @@ async def connect(self, **kwargs) -> bool:
         timeout = kwargs.get("timeout", self._timeout)
         await discover(timeout=timeout, device=self.device, loop=self.loop)
 
-        self._reactor = AsyncioSelectorReactor(self.loop)
+        self._reactor = get_reactor(self.loop)
 
         # Create system bus
         self._bus = await txdbus_connect(self._reactor, busAddress="system").asFuture(
@@ -143,11 +142,13 @@ def _services_resolved_callback(message):
                 destination="org.bluez",
             ).asFuture(self.loop)
         except RemoteError as e:
+            await self._cleanup_all()
             raise BleakError(str(e))
 
         if await self.is_connected():
             logger.debug("Connection successful.")
         else:
+            await self._cleanup_all()
             raise BleakError(
                 "Connection to {0} was not successful!".format(self.address)
             )
@@ -156,6 +157,7 @@ def _services_resolved_callback(message):
         await self.get_services()
         properties = await self._get_device_properties()
         if not properties.get("Connected"):
+            await self._cleanup_all()
             raise BleakError("Connection failed!")
 
         await self._bus.delMatch(rule_id).asFuture(self.loop)
@@ -164,22 +166,51 @@ def _services_resolved_callback(message):
         )
         return True
 
-    async def _cleanup(self) -> None:
+    async def _cleanup_notifications(self) -> None:
+        """
+        Remove all pending notifications of the client. This method is used to
+        free the DBus matches that have been established.
+        """
         for rule_name, rule_id in self._rules.items():
             logger.debug("Removing rule {0}, ID: {1}".format(rule_name, rule_id))
             try:
                 await self._bus.delMatch(rule_id).asFuture(self.loop)
             except Exception as e:
-                logger.error("Could not remove rule {0} ({1}): {2}".format(rule_id, rule_name, e))
+                logger.error(
+                    "Could not remove rule {0} ({1}): {2}".format(rule_id, rule_name, e)
+                )
         self._rules = {}
 
         for _uuid in list(self._subscriptions):
             try:
                 await self.stop_notify(_uuid)
             except Exception as e:
-                logger.error("Could not remove notifications on characteristic {0}: {1}".format(_uuid, e))
+                logger.error(
+                    "Could not remove notifications on characteristic {0}: {1}".format(
+                        _uuid, e
+                    )
+                )
         self._subscriptions = []
 
+    async def _cleanup_dbus_resources(self) -> None:
+        """
+        Free the resources allocated for both the DBus bus and the Twisted
+        reactor. Use this method upon final disconnection.
+        """
+        # Try to disconnect the System Bus.
+        try:
+            self._bus.disconnect()
+        except Exception as e:
+            logger.error("Attempt to disconnect system bus failed: {0}".format(e))
+
+    async def _cleanup_all(self) -> None:
+        """
+        Free all the allocated resource in DBus and Twisted. Use this method to
+        eventually cleanup all otherwise leaked resources.
+        """
+        await self._cleanup_notifications()
+        await self._cleanup_dbus_resources()
+
     async def disconnect(self) -> bool:
         """Disconnect from the specified GATT server.
 
@@ -190,7 +221,7 @@ async def disconnect(self) -> bool:
         logger.debug("Disconnecting from BLE device...")
 
         # Remove all residual notifications.
-        await self._cleanup()
+        await self._cleanup_notifications()
 
         # Try to disconnect the actual device/peripheral
         try:
@@ -206,21 +237,7 @@ async def disconnect(self) -> bool:
         # See if it has been disconnected.
         is_disconnected = not await self.is_connected()
 
-        # Try to disconnect the System Bus.
-        try:
-            self._bus.disconnect()
-        except Exception as e:
-            logger.error("Attempt to disconnect system bus failed: {0}".format(e))
-
-        # Stop the Twisted reactor holding the connection to the DBus system.
-        try:
-            self._reactor.stop()
-        except Exception as e:
-            # I think Bleak will always end up here, but I want to call stop just in case...
-            logger.debug("Attempt to stop Twisted reactor failed: {0}".format(e))
-        finally:
-            self._bus = None
-            self._reactor = None
+        await self._cleanup_dbus_resources()
 
         return is_disconnected
 
@@ -345,14 +362,14 @@ async def read_gatt_char(self, _uuid: Union[str, uuid.UUID], **kwargs) -> bytear
                     )
                 )
                 return value
-            if str(_uuid) == '00002a00-0000-1000-8000-00805f9b34fb' and (
+            if str(_uuid) == "00002a00-0000-1000-8000-00805f9b34fb" and (
                 self._bluez_version[0] == 5 and self._bluez_version[1] >= 48
             ):
                 props = await self._get_device_properties(
                     interface=defs.DEVICE_INTERFACE
                 )
                 # Simulate regular characteristics read to be consistent over all platforms.
-                value = bytearray(props.get("Name", "").encode('ascii'))
+                value = bytearray(props.get("Name", "").encode("ascii"))
                 logger.debug(
                     "Read Device Name {0} | {1}: {2}".format(
                         _uuid, self._device_path, value
@@ -508,22 +525,23 @@ async def write_gatt_descriptor(self, handle: int, data: bytearray) -> None:
             raise BleakError("Descriptor with handle {0} was not found!".format(handle))
         await self._bus.callRemote(
             descriptor.path,
-            'WriteValue',
+            "WriteValue",
             interface=defs.GATT_DESCRIPTOR_INTERFACE,
             destination=defs.BLUEZ_SERVICE,
-            signature='aya{sv}',
-            body=[data, {'type': 'command'}],
-            returnSignature='',
+            signature="aya{sv}",
+            body=[data, {"type": "command"}],
+            returnSignature="",
         ).asFuture(self.loop)
 
         logger.debug(
-            "Write Descriptor {0} | {1}: {2}".format(
-                handle, descriptor.path, data
-            )
+            "Write Descriptor {0} | {1}: {2}".format(handle, descriptor.path, data)
         )
 
     async def start_notify(
-        self, _uuid: Union[str, uuid.UUID], callback: Callable[[str, Any], Any], **kwargs
+        self,
+        _uuid: Union[str, uuid.UUID],
+        callback: Callable[[str, Any], Any],
+        **kwargs
     ) -> None:
         """Activate notifications/indications on a characteristic.
 
@@ -703,9 +721,11 @@ def _properties_changed_callback(self, message):
                 ):
                     logger.debug("Device {} disconnected.".format(self.address))
 
-                    task = self.loop.create_task(self._cleanup())
+                    task = self.loop.create_task(self._cleanup_all())
                     if self._disconnected_callback is not None:
-                        task.add_done_callback(partial(self._disconnected_callback, self))
+                        task.add_done_callback(
+                            partial(self._disconnected_callback, self)
+                        )
 
 
 def _data_notification_wrapper(func, char_map):

bleak/backends/bluezdbus/discovery.py

@@ -4,12 +4,10 @@
 import logging
 
 from bleak.backends.device import BLEDevice
-from bleak.backends.bluezdbus import defs
+from bleak.backends.bluezdbus import defs, get_reactor
 from bleak.backends.bluezdbus.utils import validate_mac_address
 
 from txdbus import client
-from twisted.internet.asyncioreactor import AsyncioSelectorReactor
-from twisted.internet.error import ReactorNotRunning
 
 logger = logging.getLogger(__name__)
 
@@ -80,7 +78,7 @@ async def discover(timeout=5.0, loop=None, **kwargs):
     cached_devices = {}
     devices = {}
     rules = list()
-    reactor = AsyncioSelectorReactor(loop)
+    reactor = get_reactor(loop)
 
     # Discovery filters
     filters = kwargs.get("filters", {})
@@ -235,10 +233,4 @@ def parse_msg(message):
     except Exception as e:
         logger.error("Attempt to disconnect system bus failed: {0}".format(e))
 
-    try:
-        reactor.stop()
-    except ReactorNotRunning:
-        # I think Bleak will always end up here, but I want to call stop just in case...
-        pass
-
     return discovered_devices

bleak/backends/bluezdbus/scanner.py

@@ -6,15 +6,12 @@
 from functools import wraps
 from typing import Callable, Any, Union, List
 
-
 from bleak.backends.scanner import BaseBleakScanner
 from bleak.backends.device import BLEDevice
-from bleak.backends.bluezdbus import defs
+from bleak.backends.bluezdbus import defs, get_reactor
 from bleak.backends.bluezdbus.utils import validate_mac_address
 
 from txdbus import client
-from twisted.internet.asyncioreactor import AsyncioSelectorReactor
-from twisted.internet.error import ReactorNotRunning
 
 logger = logging.getLogger(__name__)
 _here = pathlib.Path(__file__).parent
@@ -68,6 +65,7 @@ class BleakScannerBlueZDBus(BaseBleakScanner):
     Keyword Args:
 
     """
+
     def __init__(self, loop: AbstractEventLoop = None, **kwargs):
         super(BleakScannerBlueZDBus, self).__init__(loop, **kwargs)
 
@@ -89,7 +87,7 @@ def __init__(self, loop: AbstractEventLoop = None, **kwargs):
         self._callback = None
 
     async def start(self):
-        self._reactor = AsyncioSelectorReactor(self.loop)
+        self._reactor = get_reactor(self.loop)
         self._bus = await client.connect(self._reactor, "system").asFuture(self.loop)
 
         # Add signal listeners
@@ -163,11 +161,6 @@ async def stop(self):
         except Exception as e:
             logger.error("Attempt to disconnect system bus failed: {0}".format(e))
 
-        try:
-            self._reactor.stop()
-        except ReactorNotRunning:
-            pass
-
         self._bus = None
         self._reactor = None
 
@@ -238,7 +231,9 @@ def parse_msg(self, message):
             if msg_path not in self._devices and msg_path in self._cached_devices:
                 self._devices[msg_path] = self._cached_devices[msg_path]
             self._devices[msg_path] = (
-                {**self._devices[msg_path], **changed} if msg_path in self._devices else changed
+                {**self._devices[msg_path], **changed}
+                if msg_path in self._devices
+                else changed
             )
         elif (
             message.member == "InterfacesRemoved"

bleak/backends/bluezdbus/signals.py

@@ -19,7 +19,7 @@ def listen_properties_changed(bus, loop, callback):
         callback,
         interface=PROPERTIES_INTERFACE,
         member="PropertiesChanged",
-        path_namespace="/org/bluez"
+        path_namespace="/org/bluez",
     ).asFuture(loop)
 
 
@@ -39,7 +39,7 @@ def listen_interfaces_added(bus, loop, callback):
         callback,
         interface=OBJECT_MANAGER_INTERFACE,
         member="InterfacesAdded",
-        path_namespace="/org/bluez"
+        path_namespace="/org/bluez",
     ).asFuture(loop)
 
 
@@ -59,5 +59,5 @@ def listen_interfaces_removed(bus, loop, callback):
         callback,
         interface=OBJECT_MANAGER_INTERFACE,
         member="InterfacesRemoved",
-        path_namespace="/org/bluez"
+        path_namespace="/org/bluez",
     ).asFuture(loop)

bleak/backends/client.py

@@ -17,6 +17,10 @@ class BaseBleakClient(abc.ABC):
     """The Client Interface for Bleak Backend implementations to implement.
 
     The documentation of this interface should thus be safe to use as a reference for your implementation.
+
+    Keyword Args:
+        timeout (float): Timeout for required ``discover`` call. Defaults to 2.0.
+
     """
 
     def __init__(self, address, loop=None, **kwargs):