Merge pull request #244 from google/gbg/hci-source-termination-mode

add sink method for lost transports

bumble/controller.py

@@ -188,6 +188,8 @@ def __init__(
         if link:
             link.add_controller(self)
 
+        self.terminated = asyncio.get_running_loop().create_future()
+
     @property
     def host(self):
         return self.hci_sink
@@ -288,10 +290,9 @@ def send_hci_packet(self, packet):
         if self.host:
             self.host.on_packet(packet.to_bytes())
 
-    # This method allow the controller to emulate the same API as a transport source
+    # This method allows the controller to emulate the same API as a transport source
     async def wait_for_termination(self):
-        # For now, just wait forever
-        await asyncio.get_running_loop().create_future()
+        await self.terminated
 
     ############################################################
     # Link connections

bumble/host.py

@@ -20,13 +20,13 @@
 import logging
 import struct
 
+from typing import Optional
+
 from bumble.colors import color
 from bumble.l2cap import L2CAP_PDU
 from bumble.snoop import Snooper
 from bumble import drivers
 
-from typing import Optional
-
 from .hci import (
     Address,
     HCI_ACL_DATA_PACKET,
@@ -63,16 +63,15 @@
     HCI_Read_Local_Version_Information_Command,
     HCI_Reset_Command,
     HCI_Set_Event_Mask_Command,
-    map_null_terminated_utf8_string,
 )
 from .core import (
     BT_BR_EDR_TRANSPORT,
-    BT_CENTRAL_ROLE,
     BT_LE_TRANSPORT,
     ConnectionPHY,
     ConnectionParameters,
 )
 from .utils import AbortableEventEmitter
+from .transport.common import TransportLostError
 
 
 # -----------------------------------------------------------------------------
@@ -349,7 +348,7 @@ async def send_command(self, command, check_result=False):
                 return response
             except Exception as error:
                 logger.warning(
-                    f'{color("!!! Exception while sending HCI packet:", "red")} {error}'
+                    f'{color("!!! Exception while sending command:", "red")} {error}'
                 )
                 raise error
             finally:
@@ -455,6 +454,13 @@ def on_packet(self, packet):
         else:
             logger.debug('reset not done, ignoring packet from controller')
 
+    def on_transport_lost(self):
+        # Called by the source when the transport has been lost.
+        if self.pending_response:
+            self.pending_response.set_exception(TransportLostError('transport lost'))
+
+        self.emit('flush')
+
     def on_hci_packet(self, packet):
         logger.debug(f'{color("### CONTROLLER -> HOST", "green")}: {packet}')
 

bumble/transport/common.py

@@ -44,11 +44,18 @@
 }
 
 
+# -----------------------------------------------------------------------------
+class TransportLostError(Exception):
+    """
+    The Transport has been lost/disconnected.
+    """
+
+
 # -----------------------------------------------------------------------------
 class PacketPump:
-    '''
-    Pump HCI packets from a reader to a sink
-    '''
+    """
+    Pump HCI packets from a reader to a sink.
+    """
 
     def __init__(self, reader, sink):
         self.reader = reader
@@ -68,10 +75,10 @@ async def run(self):
 
 # -----------------------------------------------------------------------------
 class PacketParser:
-    '''
+    """
     In-line parser that accepts data and emits 'on_packet' when a full packet has been
-    parsed
-    '''
+    parsed.
+    """
 
     # pylint: disable=attribute-defined-outside-init
 
@@ -134,9 +141,9 @@ def set_packet_sink(self, sink):
 
 # -----------------------------------------------------------------------------
 class PacketReader:
-    '''
-    Reader that reads HCI packets from a sync source
-    '''
+    """
+    Reader that reads HCI packets from a sync source.
+    """
 
     def __init__(self, source):
         self.source = source
@@ -169,9 +176,9 @@ def next_packet(self):
 
 # -----------------------------------------------------------------------------
 class AsyncPacketReader:
-    '''
-    Reader that reads HCI packets from an async source
-    '''
+    """
+    Reader that reads HCI packets from an async source.
+    """
 
     def __init__(self, source):
         self.source = source
@@ -198,9 +205,9 @@ async def next_packet(self):
 
 # -----------------------------------------------------------------------------
 class AsyncPipeSink:
-    '''
-    Sink that forwards packets asynchronously to another sink
-    '''
+    """
+    Sink that forwards packets asynchronously to another sink.
+    """
 
     def __init__(self, sink):
         self.sink = sink
@@ -216,14 +223,29 @@ class ParserSource:
     Base class designed to be subclassed by transport-specific source classes
     """
 
+    terminated: asyncio.Future
+    parser: PacketParser
+
     def __init__(self):
         self.parser = PacketParser()
         self.terminated = asyncio.get_running_loop().create_future()
 
     def set_packet_sink(self, sink):
         self.parser.set_packet_sink(sink)
 
+    def on_transport_lost(self):
+        self.terminated.set_result(None)
+        if self.parser.sink:
+            try:
+                self.parser.sink.on_transport_lost()
+            except AttributeError:
+                pass
+
     async def wait_for_termination(self):
+        """
+        Convenience method for backward compatibility. Prefer using the `terminated`
+        attribute instead.
+        """
         return await self.terminated
 
     def close(self):

bumble/transport/tcp_client.py

@@ -39,7 +39,7 @@ async def open_tcp_client_transport(spec):
     class TcpPacketSource(StreamPacketSource):
         def connection_lost(self, exc):
             logger.debug(f'connection lost: {exc}')
-            self.terminated.set_result(exc)
+            self.on_transport_lost()
 
     remote_host, remote_port = spec.split(':')
     tcp_transport, packet_source = await asyncio.get_running_loop().create_connection(