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

add sink method for lost transports

apps/pandora_server.py

@@ -1,8 +1,10 @@
 import asyncio
 import click
 import logging
+import json
 
 from bumble.pandora import PandoraDevice, serve
+from typing import Dict, Any
 
 BUMBLE_SERVER_GRPC_PORT = 7999
 ROOTCANAL_PORT_CUTTLEFISH = 7300
@@ -18,13 +20,30 @@ ROOTCANAL_PORT_CUTTLEFISH = 7300
     help='HCI transport',
     default=f'tcp-client:127.0.0.1:<rootcanal-port>',
 )
-def main(grpc_port: int, rootcanal_port: int, transport: str) -> None:
+@click.option(
+    '--config',
+    help='Bumble json configuration file',
+)
+def main(grpc_port: int, rootcanal_port: int, transport: str, config: str) -> None:
     if '<rootcanal-port>' in transport:
         transport = transport.replace('<rootcanal-port>', str(rootcanal_port))
-    device = PandoraDevice({'transport': transport})
+
+    bumble_config = retrieve_config(config)
+    if 'transport' not in bumble_config.keys():
+        bumble_config.update({'transport': transport})
+    device = PandoraDevice(bumble_config)
+
     logging.basicConfig(level=logging.DEBUG)
     asyncio.run(serve(device, port=grpc_port))
 
 
+def retrieve_config(config: str) -> Dict[str, Any]:
+    if not config:
+        return {}
+
+    with open(config, 'r') as f:
+        return json.load(f)
+
+
 if __name__ == '__main__':
     main()  # pylint: disable=no-value-for-parameter

apps/speaker/speaker.py

@@ -228,10 +228,11 @@ class FfplayOutput(QueuedOutput):
     subprocess: Optional[asyncio.subprocess.Process]
     ffplay_task: Optional[asyncio.Task]
 
-    def __init__(self) -> None:
-        super().__init__(AacAudioExtractor())
+    def __init__(self, codec: str) -> None:
+        super().__init__(AudioExtractor.create(codec))
         self.subprocess = None
         self.ffplay_task = None
+        self.codec = codec
 
     async def start(self):
         if self.started:
@@ -240,7 +241,7 @@ class FfplayOutput(QueuedOutput):
         await super().start()
 
         self.subprocess = await asyncio.create_subprocess_shell(
-            'ffplay -acodec aac pipe:0',
+            f'ffplay -f {self.codec} pipe:0',
             stdin=asyncio.subprocess.PIPE,
             stdout=asyncio.subprocess.PIPE,
             stderr=asyncio.subprocess.PIPE,
@@ -419,7 +420,7 @@ class Speaker:
         self.outputs = []
         for output in outputs:
             if output == '@ffplay':
-                self.outputs.append(FfplayOutput())
+                self.outputs.append(FfplayOutput(codec))
                 continue
 
             # Default to FileOutput
@@ -708,17 +709,6 @@ def speaker(
 ):
     """Run the speaker."""
 
-    # ffplay only works with AAC for now
-    if codec != 'aac' and '@ffplay' in output:
-        print(
-            color(
-                f'{codec} not supported with @ffplay output, '
-                '@ffplay output will be skipped',
-                'yellow',
-            )
-        )
-        output = list(filter(lambda x: x != '@ffplay', output))
-
     if '@ffplay' in output:
         # Check if ffplay is installed
         try:

bumble/controller.py

@@ -188,6 +188,8 @@ class Controller:
         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 @@ class Controller:
         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 collections
 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 @@ from .hci import (
     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 @@ class Host(AbortableEventEmitter):
                 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 @@ class Host(AbortableEventEmitter):
         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 @@ HCI_PACKET_INFO = {
 }
 
 
+# -----------------------------------------------------------------------------
+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 @@ class PacketPump:
 
 # -----------------------------------------------------------------------------
 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 @@ class PacketParser:
 
 # -----------------------------------------------------------------------------
 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 @@ class PacketReader:
 
 # -----------------------------------------------------------------------------
 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 @@ class AsyncPacketReader:
 
 # -----------------------------------------------------------------------------
 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,6 +223,9 @@ 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()
@@ -223,7 +233,19 @@ class ParserSource:
     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(