Deprecate the openNotificationSubscription function

core/src/androidMain/kotlin/com/beepiz/bluetooth/gattcoroutines/CharacteristicNotificationsTracker.kt

@@ -0,0 +1,47 @@
+package com.beepiz.bluetooth.gattcoroutines
+
+import androidx.annotation.RequiresApi
+import kotlinx.coroutines.*
+import kotlinx.coroutines.sync.*
+import java.util.UUID
+
+@RequiresApi(18)
+internal class CharacteristicNotificationsTracker(
+    private val switch: suspend (characteristic: BGC, enable: Boolean) -> Unit
+) {
+
+    suspend fun keepNotificationsEnabled(characteristic: BGC) {
+        try {
+            countAndSwitchIfNeeded(characteristic, Operation.Add)
+            awaitCancellation()
+        } finally {
+            countAndSwitchIfNeeded(characteristic, Operation.Remove)
+        }
+    }
+
+    private enum class Operation { Add, Remove; }
+
+    private suspend fun countAndSwitchIfNeeded(
+        characteristic: BGC,
+        operation: Operation
+    ) = withContext(NonCancellable) {
+        mutex.withLock {
+            val uuid = characteristic.uuid
+            val previousCount = counts[uuid] ?: 0
+            counts[uuid] = when (operation) {
+                Operation.Add -> {
+                    if (previousCount == 0) switch(characteristic, true)
+                    previousCount + 1
+                }
+                Operation.Remove -> {
+                    if (previousCount == 1) switch(characteristic, false)
+                    previousCount - 1
+                }
+            }
+        }
+    }
+
+    private val mutex = Mutex()
+
+    private val counts = mutableMapOf<UUID, Int>()
+}

core/src/androidMain/kotlin/com/beepiz/bluetooth/gattcoroutines/GattConnection.kt

@@ -4,6 +4,7 @@ import android.Manifest
 import android.annotation.SuppressLint
 import android.bluetooth.BluetoothDevice
 import android.bluetooth.BluetoothGatt
+import android.bluetooth.BluetoothGattCharacteristic
 import android.bluetooth.BluetoothGattDescriptor
 import android.bluetooth.BluetoothGattService
 import androidx.annotation.RequiresApi
@@ -159,20 +160,35 @@ interface GattConnection {
     fun setCharacteristicNotificationsEnabled(characteristic: BGC, enable: Boolean)
 
     /**
-     * Enables notifications for that [characteristic] client-side (you still need to enable it
-     * on the remote device) and returns a [ReceiveChannel] that will get the notifications for that
-     * characteristic only.
+     * Returns a [Flow] of [BluetoothGattCharacteristic] that forwards the notifications sent by
+     * the remote device for the passed characteristic.
      *
-     * By default, [disableNotificationsOnChannelClose] is true, and will cause the notifications
-     * to be disabled client-side when the channel is closed/consumed.
+     * If [enableLocallyIfNeeded] is true, which is the default, when at least one `Flow`
+     * returned by this function for this [characteristic] is being collected,
+     * notifications will be enabled client-side for it.
      *
      * **IMPORTANT**: On most BLE devices, you'll need to enable notifications on the remote device
      * too. You can do so with the [setCharacteristicNotificationsEnabledOnRemoteDevice] function.
      *
      * You can enable notifications on the remote device before or after calling this function, both
-     * ways, notifications will be able to arrive once enabling on remote device completes.
+     * ways, notifications will be able to arrive once enabling on the remote device completes.
      */
     @RequiresPermission(Manifest.permission.BLUETOOTH_CONNECT)
+    fun notifications(
+        characteristic: BGC,
+        enableLocallyIfNeeded: Boolean = true
+    ): Flow<BGC>
+
+    /**
+     * Channel version of the [notifications] function. Will be removed in a future release.
+     */
+    @Deprecated(
+        message = "Use the notifications instead",
+        replaceWith = ReplaceWith(
+            "notifications(characteristic, enableLocallyIfNeeded = disableNotificationsOnChannelClose)"
+        )
+    )
+    @RequiresPermission(Manifest.permission.BLUETOOTH_CONNECT)
     fun openNotificationSubscription(
         characteristic: BGC,
         disableNotificationsOnChannelClose: Boolean = true
@@ -252,20 +268,20 @@ interface GattConnection {
      * Receives all characteristic update notifications.
      *
      * If you need to get the notifications of only a specific characteristic, you may want to use
-     * the [openNotificationSubscription] function instead.
-     *
-     * Since version 0.4.0, in the default implementation, this channel is backed by a
-     * [BroadcastChannel], which means you can have multiple consumers as each time you get this
-     * property, a new subscription is opened.
+     * the [allNotifications] function instead.
      *
      * For characteristic notifications to come in this channel, you need to enable it on
      * client-side (using [setCharacteristicNotificationsEnabled]), and they also need to be enabled
      * on the remote device (you can enable it with the
      * [setCharacteristicNotificationsEnabledOnRemoteDevice] function).
+     *
+     * Note that if you don't call [setCharacteristicNotificationsEnabled], this [Flow] will still
+     * emit any notifications that might be incoming if there is a [Flow] returned by the
+     * [notifications] function that is being collected.
      */
-    val notifications: Flow<BGC>
+    val allNotifications: Flow<BGC>
 
-    @Deprecated("Use notifications which returns a Flow", ReplaceWith("notifications"))
+    @Deprecated("Use notifications which returns a Flow", ReplaceWith("allNotifications"))
     val notifyChannel: ReceiveChannel<BGC>
 
     data class StateChange internal constructor(val status: Int, val newState: Int)

core/src/androidMain/kotlin/com/beepiz/bluetooth/gattcoroutines/GattConnectionImpl.kt

@@ -17,7 +17,6 @@ import kotlinx.coroutines.sync.withLock
 import splitties.bitflags.hasFlag
 import splitties.init.appCtx
 import java.util.UUID
-import kotlin.coroutines.CoroutineContext
 import android.Manifest.permission.BLUETOOTH_CONNECT as BluetoothConnectPermission
 
 @RequiresApi(18)
@@ -31,9 +30,8 @@ internal class GattConnectionImpl
 constructor(
     override val bluetoothDevice: BluetoothDevice,
     private val connectionSettings: GattConnection.ConnectionSettings
-) : GattConnection, CoroutineScope {
-    private val job = Job()
-    override val coroutineContext: CoroutineContext = Dispatchers.Main + job
+) : GattConnection {
+    private val coroutineScope = CoroutineScope(Dispatchers.Main)
 
     init {
         require(bluetoothDevice.type != BluetoothDevice.DEVICE_TYPE_CLASSIC) {
@@ -73,15 +71,15 @@ constructor(
     @Suppress("OverridingDeprecatedMember")
     @OptIn(ExperimentalCoroutinesApi::class)
     override val stateChangeChannel: ReceiveChannel<GattConnection.StateChange>
-        get() = produce { stateChangesMutableFlow.collect { send(it) } }
+        get() = coroutineScope.produce { stateChangesMutableFlow.collect { send(it) } }
 
-    override val notifications: Flow<BGC>
+    override val allNotifications: Flow<BGC>
         get() = characteristicChangedFlow
 
     @Suppress("OverridingDeprecatedMember")
     @OptIn(ExperimentalCoroutinesApi::class)
     override val notifyChannel: ReceiveChannel<BGC>
-        get() = produce { notifications.collect { send(it) } }
+        get() = coroutineScope.produce { allNotifications.collect { send(it) } }
 
     private var bluetoothGatt: BG? = null
     private fun requireGatt(): BG = bluetoothGatt ?: error("Call connect() first!")
@@ -150,7 +148,7 @@ constructor(
         reliableWriteChannel.close(cause)
         readDescChannel.close(cause)
         writeDescChannel.close(cause)
-        job.cancel()
+        coroutineScope.cancel()
     }
 
     @RequiresPermission(BluetoothConnectPermission)
@@ -172,8 +170,9 @@ constructor(
 
     @RequiresPermission(BluetoothConnectPermission)
     override fun setCharacteristicNotificationsEnabled(characteristic: BGC, enable: Boolean) {
-        requireGatt().setCharacteristicNotification(characteristic, enable)
-            .checkOperationInitiationSucceeded()
+        requireGatt().setCharacteristicNotification(characteristic, enable).also {
+            if (enable) it.checkOperationInitiationSucceeded()
+        }
     }
 
     @RequiresPermission(BluetoothConnectPermission)
@@ -194,6 +193,24 @@ constructor(
         writeDescriptor(descriptor)
     }
 
+    @RequiresPermission(BluetoothConnectPermission)
+    override fun notifications(
+        characteristic: BGC,
+        enableLocallyIfNeeded: Boolean
+    ): Flow<BGC> {
+        require(characteristic.properties.hasFlag(BGC.PROPERTY_NOTIFY)) {
+            "This characteristic doesn't support notification or doesn't come from discoverServices()."
+        }
+        return callbackFlow {
+            if (enableLocallyIfNeeded) launch {
+                notificationsTracker.keepNotificationsEnabled(characteristic)
+            }
+            characteristicChangedFlow.collect {
+                if (it.uuid == characteristic.uuid) send(it)
+            }
+        }
+    }
+
     @OptIn(ExperimentalCoroutinesApi::class)
     @RequiresPermission(BluetoothConnectPermission)
     override fun openNotificationSubscription(
@@ -203,16 +220,12 @@ constructor(
         require(characteristic.properties.hasFlag(BGC.PROPERTY_NOTIFY)) {
             "This characteristic doesn't support notification or doesn't come from discoverServices()."
         }
-        return produce {
-            setCharacteristicNotificationsEnabled(characteristic, enable = true)
-            try {
-                characteristicChangedFlow.collect {
-                    if (it.uuid == characteristic.uuid) send(it)
-                }
-            } finally {
-                if (disableNotificationsOnChannelClose) {
-                    setCharacteristicNotificationsEnabled(characteristic, enable = false)
-                }
+        return coroutineScope.produce {
+            notifications(
+                characteristic = characteristic,
+                enableLocallyIfNeeded = disableNotificationsOnChannelClose
+            ).collect {
+                send(it)
             }
         }
     }
@@ -299,7 +312,7 @@ constructor(
         }
 
         override fun onCharacteristicChanged(gatt: BG, characteristic: BGC) {
-            launch { characteristicChangedFlow.emit(characteristic) }
+            coroutineScope.launch { characteristicChangedFlow.emit(characteristic) }
         }
 
         override fun onDescriptorRead(gatt: BG, descriptor: BGD, status: Int) {
@@ -361,11 +374,17 @@ constructor(
      * status is not success.
      */
     private fun <E> SendChannel<GattResponse<E>>.launchAndSendResponse(e: E, status: Int) {
-        launch {
+        coroutineScope.launch {
             send(GattResponse(e, status))
         }
     }
 
+    private val notificationsTracker =
+        CharacteristicNotificationsTracker { characteristic, enable ->
+            @Suppress("MissingPermission") // Invoked through methods that have the permission.
+            setCharacteristicNotificationsEnabled(characteristic, enable)
+        }
+
     @Suppress("NOTHING_TO_INLINE")
     private inline fun checkNotClosed() {
         if (isClosed) throw ConnectionClosedException(closedException)
@@ -377,7 +396,7 @@ constructor(
 
     init {
         val closeOnDisconnect = connectionSettings.allowAutoConnect.not()
-        if (closeOnDisconnect) launch {
+        if (closeOnDisconnect) coroutineScope.launch {
             stateChanges.collect { stateChange ->
                 if (stateChange.newState == BluetoothProfile.STATE_DISCONNECTED) {
                     val cause =