fix linter errors

bumble/transport/android_emulator.py

@@ -97,10 +97,13 @@ async def open_android_emulator_transport(spec: Optional[str]) -> Transport:
         raise ValueError('invalid mode')
 
     # Create the transport object
-    transport = PumpedTransport(
-        PumpedPacketSource(hci_device.read),
-        PumpedPacketSink(hci_device.write),
-        channel.close,
+    class EmulatorTransport(PumpedTransport):
+        async def close(self):
+            await super().close()
+            await channel.close()
+
+    transport = EmulatorTransport(
+        PumpedPacketSource(hci_device.read), PumpedPacketSink(hci_device.write)
     )
     transport.start()
 

bumble/transport/android_netsim.py

@@ -22,7 +22,7 @@ import grpc.aio
 import os
 import pathlib
 import sys
-from typing import Optional
+from typing import Dict, Optional
 
 from .common import (
     ParserSource,
@@ -74,14 +74,20 @@ def get_ini_dir() -> Optional[pathlib.Path]:
 
 
 # -----------------------------------------------------------------------------
-def find_grpc_port() -> int:
+def ini_file_name(instance_number: int) -> str:
+    suffix = f'_{instance_number}' if instance_number > 0 else ''
+    return f'netsim{suffix}.ini'
+
+
+# -----------------------------------------------------------------------------
+def find_grpc_port(instance_number: int) -> int:
     if not (ini_dir := get_ini_dir()):
         logger.debug('no known directory for .ini file')
         return 0
 
-    ini_file = ini_dir / 'netsim.ini'
+    ini_file = ini_dir / ini_file_name(instance_number)
+    logger.debug(f'Looking for .ini file at {ini_file}')
     if ini_file.is_file():
-        logger.debug(f'Found .ini file at {ini_file}')
         with open(ini_file, 'r') as ini_file_data:
             for line in ini_file_data.readlines():
                 if '=' in line:
@@ -90,12 +96,14 @@ def find_grpc_port() -> int:
                         logger.debug(f'gRPC port = {value}')
                         return int(value)
 
+        logger.debug('no grpc.port property found in .ini file')
+
     # Not found
     return 0
 
 
 # -----------------------------------------------------------------------------
-def publish_grpc_port(grpc_port) -> bool:
+def publish_grpc_port(grpc_port: int, instance_number: int = 0) -> bool:
     if not (ini_dir := get_ini_dir()):
         logger.debug('no known directory for .ini file')
         return False
@@ -104,7 +112,7 @@ def publish_grpc_port(grpc_port) -> bool:
         logger.debug('ini directory does not exist')
         return False
 
-    ini_file = ini_dir / 'netsim.ini'
+    ini_file = ini_dir / ini_file_name(instance_number)
     try:
         ini_file.write_text(f'grpc.port={grpc_port}\n')
         logger.debug(f"published gRPC port at {ini_file}")
@@ -122,14 +130,15 @@ def publish_grpc_port(grpc_port) -> bool:
 
 # -----------------------------------------------------------------------------
 async def open_android_netsim_controller_transport(
-    server_host: Optional[str], server_port: int
+    server_host: Optional[str], server_port: int, options: Dict[str, str]
 ) -> Transport:
     if not server_port:
         raise ValueError('invalid port')
     if server_host == '_' or not server_host:
         server_host = 'localhost'
 
-    if not publish_grpc_port(server_port):
+    instance_number = int(options.get('instance', "0"))
+    if not publish_grpc_port(server_port, instance_number):
         logger.warning("unable to publish gRPC port")
 
     class HciDevice:
@@ -186,15 +195,12 @@ async def open_android_netsim_controller_transport(
                 logger.debug(f'<<< PACKET: {data.hex()}')
                 self.on_data_received(data)
 
-        def send_packet(self, data):
-            async def send():
-                await self.context.write(
-                    PacketResponse(
-                        hci_packet=HCIPacket(packet_type=data[0], packet=data[1:])
-                    )
+        async def send_packet(self, data):
+            return await self.context.write(
+                PacketResponse(
+                    hci_packet=HCIPacket(packet_type=data[0], packet=data[1:])
                 )
-
-            self.loop.create_task(send())
+            )
 
         def terminate(self):
             self.task.cancel()
@@ -228,17 +234,17 @@ async def open_android_netsim_controller_transport(
                 logger.debug('gRPC server cancelled')
                 await self.grpc_server.stop(None)
 
-        def on_packet(self, packet):
+        async def send_packet(self, packet):
             if not self.device:
                 logger.debug('no device, dropping packet')
                 return
 
-            self.device.send_packet(packet)
+            return await self.device.send_packet(packet)
 
         async def StreamPackets(self, _request_iterator, context):
             logger.debug('StreamPackets request')
 
-            # Check that we won't already have a device
+            # Check that we don't already have a device
             if self.device:
                 logger.debug('busy, already serving a device')
                 return PacketResponse(error='Busy')
@@ -261,15 +267,15 @@ async def open_android_netsim_controller_transport(
     await server.start()
     asyncio.get_running_loop().create_task(server.serve())
 
-    class GrpcServerTransport(Transport):
-        async def close(self):
-            await super().close()
-
-    return GrpcServerTransport(server, server)
+    sink = PumpedPacketSink(server.send_packet)
+    sink.start()
+    return Transport(server, sink)
 
 
 # -----------------------------------------------------------------------------
-async def open_android_netsim_host_transport(server_host, server_port, options):
+async def open_android_netsim_host_transport(
+    server_host: Optional[str], server_port: int, options: Dict[str, str]
+):
     # Wrapper for I/O operations
     class HciDevice:
         def __init__(self, name, manufacturer, hci_device):
@@ -308,6 +314,7 @@ async def open_android_netsim_host_transport(server_host, server_port, options):
 
     name = options.get('name', DEFAULT_NAME)
     manufacturer = DEFAULT_MANUFACTURER
+    instance_number = int(options.get('instance', "0"))
 
     if server_host == '_' or not server_host:
         server_host = 'localhost'
@@ -315,7 +322,7 @@ async def open_android_netsim_host_transport(server_host, server_port, options):
     if not server_port:
         # Look for the gRPC config in a .ini file
         server_host = 'localhost'
-        server_port = find_grpc_port()
+        server_port = find_grpc_port(instance_number)
         if not server_port:
             raise RuntimeError('gRPC server port not found')
 
@@ -334,10 +341,14 @@ async def open_android_netsim_host_transport(server_host, server_port, options):
     await hci_device.start()
 
     # Create the transport object
-    transport = PumpedTransport(
+    class GrpcTransport(PumpedTransport):
+        async def close(self):
+            await super().close()
+            await channel.close()
+
+    transport = GrpcTransport(
         PumpedPacketSource(hci_device.read),
         PumpedPacketSink(hci_device.write),
-        channel.close,
     )
     transport.start()
 
@@ -345,7 +356,7 @@ async def open_android_netsim_host_transport(server_host, server_port, options):
 
 
 # -----------------------------------------------------------------------------
-async def open_android_netsim_transport(spec):
+async def open_android_netsim_transport(spec: Optional[str]) -> Transport:
     '''
     Open a transport connection as a client or server, implementing Android's `netsim`
     simulator protocol over gRPC.
@@ -359,6 +370,11 @@ async def open_android_netsim_transport(spec):
         to connect *to* a netsim server (netsim is the controller), or accept
         connections *as* a netsim-compatible server.
 
+      instance=<n>
+        Specifies an instance number, with <n> > 0. This is used to determine which
+        .init file to use. In `host` mode, it is ignored when the <host>:<port>
+        specifier is present, since in that case no .ini file is used.
+
     In `host` mode:
       The <host>:<port> part is optional. When not specified, the transport
       looks for a netsim .ini file, from which it will read the `grpc.backend.port`
@@ -387,14 +403,15 @@ async def open_android_netsim_transport(spec):
     params = spec.split(',') if spec else []
     if params and ':' in params[0]:
         # Explicit <host>:<port>
-        host, port = params[0].split(':')
+        host, port_str = params[0].split(':')
+        port = int(port_str)
         params_offset = 1
     else:
         host = None
         port = 0
         params_offset = 0
 
-    options = {}
+    options: Dict[str, str] = {}
     for param in params[params_offset:]:
         if '=' not in param:
             raise ValueError('invalid parameter, expected <name>=<value>')
@@ -407,6 +424,6 @@ async def open_android_netsim_transport(spec):
     if mode == 'controller':
         if host is None:
             raise ValueError('<host>:<port> missing')
-        return await open_android_netsim_controller_transport(host, port)
+        return await open_android_netsim_controller_transport(host, port, options)
 
     raise ValueError('invalid mode option')

bumble/transport/common.py

@@ -339,8 +339,9 @@ class PumpedPacketSource(ParserSource):
                 try:
                     packet = await self.receive_function()
                     self.parser.feed_data(packet)
-                except asyncio.exceptions.CancelledError:
+                except asyncio.CancelledError:
                     logger.debug('source pump task done')
+                    self.terminated.set_result(None)
                     break
                 except Exception as error:
                     logger.warning(f'exception while waiting for packet: {error}')
@@ -370,7 +371,7 @@ class PumpedPacketSink:
                 try:
                     packet = await self.packet_queue.get()
                     await self.send_function(packet)
-                except asyncio.exceptions.CancelledError:
+                except asyncio.CancelledError:
                     logger.debug('sink pump task done')
                     break
                 except Exception as error:
@@ -393,19 +394,13 @@ class PumpedTransport(Transport):
         self,
         source: PumpedPacketSource,
         sink: PumpedPacketSink,
-        close_function,
     ) -> None:
         super().__init__(source, sink)
-        self.close_function = close_function
 
     def start(self) -> None:
         self.source.start()
         self.sink.start()
 
-    async def close(self) -> None:
-        await super().close()
-        await self.close_function()
-
 
 # -----------------------------------------------------------------------------
 class SnoopingTransport(Transport):

bumble/transport/ws_client.py

@@ -38,10 +38,14 @@ async def open_ws_client_transport(spec: str) -> Transport:
 
     websocket = await websockets.client.connect(spec)
 
-    transport = PumpedTransport(
+    class WsTransport(PumpedTransport):
+        async def close(self):
+            await super().close()
+            await websocket.close()
+
+    transport = WsTransport(
         PumpedPacketSource(websocket.recv),
         PumpedPacketSink(websocket.send),
-        websocket.close,
     )
     transport.start()
     return transport