Fix BLE companion transport reliability

android/app/src/main/java/cloud/molberg/picar/BleClient.kt

@@ -7,6 +7,7 @@ import android.bluetooth.le.ScanFilter
 import android.bluetooth.le.ScanResult
 import android.bluetooth.le.ScanSettings
 import android.content.Context
+import android.os.Build
 import android.os.ParcelUuid
 import android.os.Handler
 import android.os.Looper
@@ -33,6 +34,7 @@ class BleClient(private val context: Context) {
     private val incoming = StringBuilder()
     private val pending = ConcurrentHashMap<String, CompletableDeferred<JSONObject>>()
     private val writes = ArrayDeque<ByteArray>()
+    private var writeInProgress = false
     private val handler = Handler(Looper.getMainLooper())
     private var stopped = true
 
@@ -49,6 +51,7 @@ class BleClient(private val context: Context) {
         handler.removeCallbacksAndMessages(null)
         bluetoothManager.adapter?.bluetoothLeScanner?.stopScan(scanCallback)
         gatt?.close(); gatt = null; rx = null
+        resetWrites()
         state.value = ConnectionState.STOPPED
     }
 
@@ -77,49 +80,119 @@ class BleClient(private val context: Context) {
 
     private val gattCallback = object : BluetoothGattCallback() {
         override fun onConnectionStateChange(gatt: BluetoothGatt, status: Int, newState: Int) {
-            if (newState == BluetoothProfile.STATE_CONNECTED) { this@BleClient.gatt = gatt; gatt.requestMtu(247); gatt.discoverServices() }
+            if (status == BluetoothGatt.GATT_SUCCESS && newState == BluetoothProfile.STATE_CONNECTED) {
+                this@BleClient.gatt = gatt
+                // Android GATT operations are serialized. Starting service
+                // discovery while the MTU request is still pending can leave
+                // the client in CONNECTING forever on some phones.
+                if (!gatt.requestMtu(247)) gatt.discoverServices()
+            }
             else {
-                rx = null; gatt.close(); state.value = ConnectionState.DISCONNECTED
+                rx = null; gatt.close(); resetWrites(); state.value = ConnectionState.DISCONNECTED
                 if (!stopped) handler.postDelayed({ connect() }, 3_000)
             }
         }
+
+        override fun onMtuChanged(gatt: BluetoothGatt, mtu: Int, status: Int) {
+            gatt.discoverServices()
+        }
+
         override fun onServicesDiscovered(gatt: BluetoothGatt, status: Int) {
-            val service = gatt.getService(SERVICE_UUID) ?: return
+            val service = gatt.getService(SERVICE_UUID)
-            rx = service.getCharacteristic(RX_UUID)
+            val nextRx = service?.getCharacteristic(RX_UUID)
-            val tx = service.getCharacteristic(TX_UUID) ?: return
+            val tx = service?.getCharacteristic(TX_UUID)
-            gatt.setCharacteristicNotification(tx, true)
+            val descriptor = tx?.getDescriptor(CCC_UUID)
-            tx.getDescriptor(CCC_UUID)?.let { descriptor ->
+            if (status != BluetoothGatt.GATT_SUCCESS || nextRx == null || tx == null || descriptor == null) {
-                descriptor.value = BluetoothGattDescriptor.ENABLE_NOTIFICATION_VALUE
+                gatt.disconnect()
+                return
+            }
+            rx = nextRx
+            if (!gatt.setCharacteristicNotification(tx, true)) {
+                gatt.disconnect()
+                return
+            }
+            val descriptorStarted = if (Build.VERSION.SDK_INT >= 33) {
+                gatt.writeDescriptor(descriptor, BluetoothGattDescriptor.ENABLE_INDICATION_VALUE) == BluetoothStatusCodes.SUCCESS
+            } else {
+                descriptor.value = BluetoothGattDescriptor.ENABLE_INDICATION_VALUE
                 gatt.writeDescriptor(descriptor)
             }
-            state.value = ConnectionState.CONNECTED
+            if (!descriptorStarted) gatt.disconnect()
         }
+
+        override fun onDescriptorWrite(gatt: BluetoothGatt, descriptor: BluetoothGattDescriptor, status: Int) {
+            if (descriptor.uuid == CCC_UUID && status == BluetoothGatt.GATT_SUCCESS) {
+                state.value = ConnectionState.CONNECTED
+            } else {
+                gatt.disconnect()
+            }
+        }
+        @Deprecated("Used on Android 12 and earlier")
         override fun onCharacteristicChanged(gatt: BluetoothGatt, characteristic: BluetoothGattCharacteristic) {
-            incoming.append(characteristic.value.toString(Charsets.UTF_8))
+            handleIncoming(characteristic.value)
-            while (incoming.contains("\n")) {
+        }
-                val index = incoming.indexOf("\n")
+
-                val line = incoming.substring(0, index); incoming.delete(0, index + 1)
+        override fun onCharacteristicChanged(gatt: BluetoothGatt, characteristic: BluetoothGattCharacteristic, value: ByteArray) {
-                runCatching { JSONObject(line) }.getOrNull()?.let { response ->
+            handleIncoming(value)
-                    pending.remove(response.optString("id"))?.let { deferred ->
+        }
-                        if (response.optBoolean("ok")) deferred.complete(response.getJSONObject("data"))
+        override fun onCharacteristicWrite(gatt: BluetoothGatt, characteristic: BluetoothGattCharacteristic, status: Int) {
-                        else deferred.completeExceptionally(IllegalStateException(response.optJSONObject("error")?.optString("message") ?: "Bluetooth request failed"))
+            if (status == BluetoothGatt.GATT_SUCCESS) writeNext() else {
-                    }
+                synchronized(this@BleClient) { writeInProgress = false }
+                gatt.disconnect()
+            }
+        }
+    }
+
+    private fun handleIncoming(value: ByteArray) {
+        incoming.append(value.toString(Charsets.UTF_8))
+        while (incoming.contains("\n")) {
+            val index = incoming.indexOf("\n")
+            val line = incoming.substring(0, index); incoming.delete(0, index + 1)
+            runCatching { JSONObject(line) }.getOrNull()?.let { response ->
+                pending.remove(response.optString("id"))?.let { deferred ->
+                    if (response.optBoolean("ok")) deferred.complete(response.getJSONObject("data"))
+                    else deferred.completeExceptionally(IllegalStateException(response.optJSONObject("error")?.optString("message") ?: "Bluetooth request failed"))
                 }
             }
         }
-        override fun onCharacteristicWrite(gatt: BluetoothGatt, characteristic: BluetoothGattCharacteristic, status: Int) { writeNext() }
     }
 
     @Synchronized private fun enqueue(bytes: ByteArray) {
-        bytes.asList().chunked(180).forEach { chunk -> writes.add(chunk.toByteArray()) }
+        val chunks = bytes.asList().chunked(180)
-        if (writes.size == bytes.asList().chunked(180).size) writeNext()
+        chunks.forEach { chunk -> writes.add(chunk.toByteArray()) }
+        if (!writeInProgress) {
+            writeInProgress = true
+            writeNext()
+        }
     }
 
     @Synchronized private fun writeNext() {
-        val characteristic = rx ?: return
+        val characteristic = rx
-        val next = writes.removeFirstOrNull() ?: return
+        val currentGatt = gatt
-        characteristic.writeType = BluetoothGattCharacteristic.WRITE_TYPE_DEFAULT
+        if (characteristic == null || currentGatt == null) {
-        characteristic.value = next
+            writeInProgress = false
-        gatt?.writeCharacteristic(characteristic)
+            return
+        }
+        val next = writes.removeFirstOrNull()
+        if (next == null) {
+            writeInProgress = false
+            return
+        }
+        val started = if (Build.VERSION.SDK_INT >= 33) {
+            currentGatt.writeCharacteristic(characteristic, next, BluetoothGattCharacteristic.WRITE_TYPE_DEFAULT) == BluetoothStatusCodes.SUCCESS
+        } else {
+            characteristic.writeType = BluetoothGattCharacteristic.WRITE_TYPE_DEFAULT
+            characteristic.value = next
+            currentGatt.writeCharacteristic(characteristic)
+        }
+        if (!started) {
+            writeInProgress = false
+            currentGatt.disconnect()
+        }
+    }
+
+    @Synchronized private fun resetWrites() {
+        writes.clear()
+        writeInProgress = false
     }
 }

scripts/pi-car-companion-bluetooth

@@ -21,8 +21,6 @@ DBUS_PROPERTIES = "org.freedesktop.DBus.Properties"
 GATT_MANAGER = "org.bluez.GattManager1"
 GATT_SERVICE = "org.bluez.GattService1"
 GATT_CHARACTERISTIC = "org.bluez.GattCharacteristic1"
-AD_MANAGER = "org.bluez.LEAdvertisingManager1"
-ADVERTISEMENT = "org.bluez.LEAdvertisement1"
 
 SERVICE_UUID = "6c7a0001-7c6d-4f74-9bb0-c5a4e5efc001"
 RX_UUID = "6c7a0002-7c6d-4f74-9bb0-c5a4e5efc001"
@@ -103,19 +101,30 @@ class Characteristic(dbus.service.Object):
 
 class TxCharacteristic(Characteristic):
     def __init__(self, bus, service):
-        super().__init__(bus, 1, TX_UUID, ["notify"], service)
+        super().__init__(bus, 1, TX_UUID, ["indicate"], service)
         self.notifying = False
         self.queue = deque()
         self.sending = False
 
-    @dbus.service.method(GATT_CHARACTERISTIC, in_signature="a{sv}")
+    @dbus.service.method(GATT_CHARACTERISTIC)
-    def StartNotify(self, _options):
+    def StartNotify(self):
         self.notifying = True
+        if self.queue and not self.sending:
+            self.sending = True
+            GLib.idle_add(self._send_next)
 
-    @dbus.service.method(GATT_CHARACTERISTIC, in_signature="a{sv}")
+    @dbus.service.method(GATT_CHARACTERISTIC)
-    def StopNotify(self, _options):
+    def StopNotify(self):
         self.notifying = False
         self.queue.clear()
+        self.sending = False
+
+    @dbus.service.method(GATT_CHARACTERISTIC)
+    def Confirm(self):
+        if self.queue:
+            GLib.idle_add(self._send_next)
+        else:
+            self.sending = False
 
     def send(self, message):
         encoded = (json.dumps(message, separators=(",", ":")) + "\n").encode("utf-8")
@@ -123,7 +132,7 @@ class TxCharacteristic(Characteristic):
             self.queue.append(encoded[offset:offset + NOTIFY_CHUNK_BYTES])
         if not self.sending:
             self.sending = True
-            GLib.timeout_add(20, self._send_next)
+            GLib.idle_add(self._send_next)
 
     def _send_next(self):
         if not self.notifying or not self.queue:
@@ -132,7 +141,7 @@ class TxCharacteristic(Characteristic):
         chunk = self.queue.popleft()
         value = dbus.Array([dbus.Byte(byte) for byte in chunk], signature="y")
         self.PropertiesChanged(GATT_CHARACTERISTIC, {"Value": value}, [])
-        return bool(self.queue)
+        return False
 
 
 class RxCharacteristic(Characteristic):
@@ -216,26 +225,10 @@ def call_api(method, path, body, token):
         raise ProtocolError("PI_SERVICE_UNAVAILABLE", "The Pi companion service is unavailable") from error
 
 
-class Advertisement(dbus.service.Object):
-    def __init__(self, bus):
-        self.path = "/org/pi_car_companion/advertisement0"
-        super().__init__(bus, self.path)
-
-    @dbus.service.method(DBUS_PROPERTIES, in_signature="s", out_signature="a{sv}")
-    def GetAll(self, interface):
-        if interface != ADVERTISEMENT:
-            raise InvalidArgs()
-        return {"Type": "peripheral", "ServiceUUIDs": dbus.Array([SERVICE_UUID], signature="s"), "LocalName": "Pi Car"}
-
-    @dbus.service.method(ADVERTISEMENT)
-    def Release(self):
-        pass
-
-
 def find_adapter(bus):
     objects = dbus.Interface(bus.get_object(BLUEZ, "/"), DBUS_OM).GetManagedObjects()
     for path, interfaces in objects.items():
-        if GATT_MANAGER in interfaces and AD_MANAGER in interfaces:
+        if GATT_MANAGER in interfaces:
             return path
     raise RuntimeError("No Bluetooth LE adapter with GATT server support was found")
 
@@ -253,10 +246,22 @@ def main():
     service.add_characteristic(rx)
     service.add_characteristic(tx)
     application.add_service(service)
-    advertisement = Advertisement(bus)
+    loop = GLib.MainLoop()
-    dbus.Interface(adapter, GATT_MANAGER).RegisterApplication(application.path, {}, reply_handler=lambda: None, error_handler=lambda error: (_ for _ in ()).throw(error))
+    registration_errors = []
-    dbus.Interface(adapter, AD_MANAGER).RegisterAdvertisement(advertisement.path, {}, reply_handler=lambda: None, error_handler=lambda error: (_ for _ in ()).throw(error))
+
-    GLib.MainLoop().run()
+    def registration_failed(error):
+        registration_errors.append(error)
+        loop.quit()
+
+    dbus.Interface(adapter, GATT_MANAGER).RegisterApplication(
+        application.path,
+        {},
+        reply_handler=lambda: None,
+        error_handler=registration_failed,
+    )
+    loop.run()
+    if registration_errors:
+        raise registration_errors[0]
 
 
 if __name__ == "__main__":

systemd/pi-car-companion-bluetooth.service

@@ -7,6 +7,7 @@ Wants=pi-car-companion.service
 [Service]
 Type=simple
 ExecStart=/usr/local/libexec/pi-car-companion-bluetooth
+ExecStartPost=-/usr/bin/timeout 10 /usr/bin/btmgmt add-adv -c -u 6c7a0001-7c6d-4f74-9bb0-c5a4e5efc001 1
 Restart=on-failure
 RestartSec=5s
 NoNewPrivileges=true
@@ -18,7 +19,7 @@ ProtectKernelModules=true
 ProtectControlGroups=true
 RestrictRealtime=true
 LockPersonality=true
-RestrictAddressFamilies=AF_UNIX AF_INET AF_INET6
+RestrictAddressFamilies=AF_UNIX AF_INET AF_INET6 AF_BLUETOOTH
 
 [Install]
 WantedBy=multi-user.target