Have onConnectionStateChange backed by StateFlow (#66)

JuulLabs · May 21, 2020 · 5da3384 · 5da3384
1 parent 2c407fb
commit 5da3384
Show file tree

Hide file tree

Showing 4 changed files with 18 additions and 17 deletions.
diff --git a/core/src/main/java/gatt/CoroutinesGatt.kt b/core/src/main/java/gatt/CoroutinesGatt.kt
@@ -34,7 +34,7 @@ class CoroutinesGatt internal constructor(
 ) : Gatt {
 
     @FlowPreview
-    override val onConnectionStateChange = callback.onConnectionStateChange.asFlow()
+    override val onConnectionStateChange = callback.onConnectionStateChange
 
     @FlowPreview
     override val onCharacteristicChanged = callback.onCharacteristicChanged.asFlow()

diff --git a/core/src/main/java/gatt/GattCallback.kt b/core/src/main/java/gatt/GattCallback.kt
@@ -5,6 +5,7 @@
 package com.juul.able.gatt
 
 import android.bluetooth.BluetoothGatt
+import android.bluetooth.BluetoothGatt.GATT_SUCCESS
 import android.bluetooth.BluetoothGattCallback
 import android.bluetooth.BluetoothGattCharacteristic
 import android.bluetooth.BluetoothGattDescriptor
@@ -17,13 +18,19 @@ import kotlinx.coroutines.channels.BroadcastChannel
 import kotlinx.coroutines.channels.Channel
 import kotlinx.coroutines.channels.Channel.Factory.BUFFERED
 import kotlinx.coroutines.channels.Channel.Factory.CONFLATED
+import kotlinx.coroutines.flow.Flow
+import kotlinx.coroutines.flow.MutableStateFlow
+import kotlinx.coroutines.flow.filterNotNull
 import kotlinx.coroutines.runBlocking
 
 internal class GattCallback(
     private val dispatcher: ExecutorCoroutineDispatcher
 ) : BluetoothGattCallback() {
 
-    val onConnectionStateChange = BroadcastChannel<OnConnectionStateChange>(CONFLATED)
+    private val _onConnectionStateChange = MutableStateFlow<OnConnectionStateChange?>(null)
+    val onConnectionStateChange: Flow<OnConnectionStateChange> =
+        _onConnectionStateChange.filterNotNull()
+
     val onCharacteristicChanged = BroadcastChannel<OnCharacteristicChanged>(BUFFERED)
     val onResponse = Channel<Any>(CONFLATED)
 
@@ -38,9 +45,11 @@ internal class GattCallback(
         if (isClosed.compareAndSet(false, true)) {
             Able.verbose { "Closing GattCallback belonging to device ${gatt.device}" }
             onDisconnecting() // Duplicate call in case Android skips STATE_DISCONNECTING.
-            onConnectionStateChange.close()
             gatt.close()
 
+            _onConnectionStateChange.value =
+                OnConnectionStateChange(GATT_SUCCESS, STATE_DISCONNECTED)
+
             // todo: Remove when https://github.com/Kotlin/kotlinx.coroutines/issues/261 is fixed.
             dispatcher.close()
         }
@@ -53,7 +62,7 @@ internal class GattCallback(
     ) {
         val event = OnConnectionStateChange(status, newState)
         Able.debug { "← $event" }
-        onConnectionStateChange.offer(event)
+        _onConnectionStateChange.value = event
 
         when (newState) {
             STATE_DISCONNECTING -> onDisconnecting()

diff --git a/core/src/main/java/gatt/GattConnection.kt b/core/src/main/java/gatt/GattConnection.kt
@@ -12,7 +12,7 @@ import android.bluetooth.BluetoothProfile.STATE_DISCONNECTED
 import com.juul.able.Able
 import kotlinx.coroutines.FlowPreview
 import kotlinx.coroutines.flow.Flow
-import kotlinx.coroutines.flow.firstOrNull
+import kotlinx.coroutines.flow.first
 import kotlinx.coroutines.flow.onEach
 
 /**
@@ -60,18 +60,10 @@ internal suspend fun GattConnection.suspendUntilConnectionState(state: GattConne
         .onEach { event ->
             Able.verbose { "← Received $event while waiting for ${state.asGattConnectionStateString()}" }
             if (event.status != GATT_SUCCESS) throw GattStatusFailure(event)
+            if (event.newState == STATE_DISCONNECTED && state != STATE_DISCONNECTED) throw ConnectionLost()
         }
-        .firstOrNull { (_, newState) -> newState == state }
-        .also {
-            if (it == null) { // Upstream Channel closed due to STATE_DISCONNECTED.
-                if (state == STATE_DISCONNECTED) {
-                    Able.info { "Reached (implicit) STATE_DISCONNECTED" }
-                } else {
-                    throw ConnectionLost()
-                }
-            }
-        }
-        ?.also { (_, newState) ->
+        .first { (_, newState) -> newState == state }
+        .also { (_, newState) ->
             Able.info { "Reached ${newState.asGattConnectionStateString()}" }
         }
 }
diff --git a/gradle/dependencies.gradle b/gradle/dependencies.gradle
@@ -7,7 +7,7 @@ ext.versions = [
 ext.deps = [
     kotlin: [
         stdlib: "org.jetbrains.kotlin:kotlin-stdlib",
-        coroutines: "org.jetbrains.kotlinx:kotlinx-coroutines-core:1.3.5",
+        coroutines: "org.jetbrains.kotlinx:kotlinx-coroutines-core:1.3.7",
         junit: "org.jetbrains.kotlin:kotlin-test-junit",
     ],