Add keep-alive module

keep-alive/build.gradle

@@ -0,0 +1,35 @@
+plugins {
+    id 'com.android.library'
+    id 'org.jetbrains.kotlin.android'
+    id 'org.jmailen.kotlinter'
+    id 'com.hiya.jacoco-android'
+    id 'com.vanniktech.maven.publish'
+}
+
+jacoco {
+    toolVersion = "0.8.5"
+}
+
+jacocoAndroidUnitTestReport {
+    csv.enabled false
+    html.enabled true
+    xml.enabled true
+}
+
+mavenPublish {
+    useLegacyMode = true
+}
+
+android {
+    compileSdkVersion versions.compileSdk
+
+    defaultConfig {
+        minSdkVersion versions.minSdk
+    }
+}
+
+dependencies {
+    api project(':core')
+    testImplementation deps.kotlin.junit
+    testImplementation deps.mockk
+}

keep-alive/src/main/AndroidManifest.xml

@@ -0,0 +1,5 @@
+<!--
+  ~ Copyright 2020 JUUL Labs, Inc.
+  -->
+
+<manifest package="com.juul.able.keepalive" />

keep-alive/src/main/java/KeepAliveGatt.kt

@@ -0,0 +1,169 @@
+/*
+ * Copyright 2020 JUUL Labs, Inc.
+ */
+
+package com.juul.able.keepalive
+
+import android.bluetooth.BluetoothDevice
+import android.bluetooth.BluetoothGattCharacteristic
+import android.bluetooth.BluetoothGattDescriptor
+import android.bluetooth.BluetoothGattService
+import android.content.Context
+import com.juul.able.Able
+import com.juul.able.android.connectGatt
+import com.juul.able.device.ConnectGattResult.Failure
+import com.juul.able.device.ConnectGattResult.Success
+import com.juul.able.gatt.GattIo
+import com.juul.able.gatt.GattStatus
+import com.juul.able.gatt.OnCharacteristicChanged
+import com.juul.able.gatt.OnCharacteristicRead
+import com.juul.able.gatt.OnCharacteristicWrite
+import com.juul.able.gatt.OnDescriptorWrite
+import com.juul.able.gatt.OnMtuChanged
+import com.juul.able.gatt.OnReadRemoteRssi
+import com.juul.able.gatt.WriteType
+import kotlinx.coroutines.CoroutineName
+import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.FlowPreview
+import kotlinx.coroutines.Job
+import kotlinx.coroutines.NonCancellable
+import kotlinx.coroutines.SupervisorJob
+import kotlinx.coroutines.channels.BroadcastChannel
+import kotlinx.coroutines.channels.Channel
+import kotlinx.coroutines.channels.consume
+import kotlinx.coroutines.coroutineScope
+import kotlinx.coroutines.flow.Flow
+import kotlinx.coroutines.flow.asFlow
+import kotlinx.coroutines.flow.launchIn
+import kotlinx.coroutines.flow.onEach
+import kotlinx.coroutines.isActive
+import kotlinx.coroutines.launch
+import kotlinx.coroutines.withContext
+import kotlinx.coroutines.withTimeoutOrNull
+import java.util.UUID
+import kotlin.coroutines.CoroutineContext
+
+enum class State {
+    Connecting,
+    Connected,
+    Disconnecting,
+    Disconnected,
+}
+
+typealias ConnectAction = suspend GattIo.() -> Unit
+
+class NotReady(message: String) : IllegalStateException(message)
+
+fun CoroutineScope.keepAliveGatt(
+    androidContext: Context,
+    device: BluetoothDevice,
+    disconnectTimeoutMillis: Long,
+    onConnectAction: ConnectAction
+) = KeepAliveGatt(coroutineContext, androidContext, device, disconnectTimeoutMillis, onConnectAction)
+
+class KeepAliveGatt internal constructor(
+    coroutineContext: CoroutineContext,
+    androidContext: Context,
+    private val device: BluetoothDevice,
+    private val disconnectTimeoutMillis: Long,
+    private val onConnectAction: ConnectAction
+) : GattIo {
+
+    private val applicationContext = androidContext.applicationContext
+
+    init {
+        val parentJob = coroutineContext[Job]
+        CoroutineScope(coroutineContext + SupervisorJob(parentJob)).launch(
+            CoroutineName("KeepAliveGatt@$device")
+        ) {
+            while (isActive) spawnConnection()
+        }.apply {
+            invokeOnCompletion {
+                _onCharacteristicChanged.cancel()
+                _state.cancel()
+            }
+        }
+    }
+
+    @Volatile
+    private var _gatt: GattIo? = null
+
+    private val gatt: GattIo
+        inline get() = _gatt ?: throw NotReady(toString())
+
+    private suspend fun spawnConnection() {
+        _state.offer(State.Connecting)
+
+        val gatt = when (val result = device.connectGatt(applicationContext)) {
+            is Success -> result.gatt
+            is Failure -> {
+                _state.offer(State.Disconnected)
+                Able.error(result.cause) { "Failed to connect to $device" }
+                return
+            }
+        }
+
+        try {
+            coroutineScope {
+                gatt.onCharacteristicChanged
+                    .onEach(_onCharacteristicChanged::send)
+                    .launchIn(this)
+                onConnectAction.invoke(gatt)
+                _gatt = gatt
+                _state.offer(State.Connected)
+            }
+        } finally {
+            _state.offer(State.Disconnecting)
+            withContext(NonCancellable) {
+                withTimeoutOrNull(disconnectTimeoutMillis) {
+                    gatt.disconnect()
+                } ?: Able.warn { "Timed out waiting ${disconnectTimeoutMillis}ms for disconnect" }
+                _state.offer(State.Disconnected)
+            }
+        }
+    }
+
+    private val _state = BroadcastChannel<State>(Channel.CONFLATED)
+
+    @FlowPreview
+    val state: Flow<State> = _state.asFlow()
+
+    private val _onCharacteristicChanged = BroadcastChannel<OnCharacteristicChanged>(Channel.BUFFERED)
+
+    @FlowPreview
+    override val onCharacteristicChanged: Flow<OnCharacteristicChanged> =
+        _onCharacteristicChanged.asFlow()
+
+    override suspend fun discoverServices(): GattStatus = gatt.discoverServices()
+
+    override val services: List<BluetoothGattService> get() = gatt.services
+    override fun getService(uuid: UUID): BluetoothGattService? = gatt.getService(uuid)
+
+    override suspend fun requestMtu(mtu: Int): OnMtuChanged = gatt.requestMtu(mtu)
+
+    override suspend fun readCharacteristic(
+        characteristic: BluetoothGattCharacteristic
+    ): OnCharacteristicRead = gatt.readCharacteristic(characteristic)
+
+    override fun setCharacteristicNotification(
+        characteristic: BluetoothGattCharacteristic,
+        enable: Boolean
+    ): Boolean = gatt.setCharacteristicNotification(characteristic, enable)
+
+    override suspend fun writeCharacteristic(
+        characteristic: BluetoothGattCharacteristic,
+        value: ByteArray,
+        writeType: WriteType
+    ): OnCharacteristicWrite = gatt.writeCharacteristic(characteristic, value, writeType)
+
+    override suspend fun writeDescriptor(
+        descriptor: BluetoothGattDescriptor,
+        value: ByteArray
+    ): OnDescriptorWrite = gatt.writeDescriptor(descriptor, value)
+
+    override suspend fun readRemoteRssi(): OnReadRemoteRssi = gatt.readRemoteRssi()
+
+    // todo: Verify that this doesn't throw after _state is closed.
+    override fun toString() =
+        "KeepAliveGatt(device=$device, gatt=$_gatt, state=${_state.consume { poll() }})"
+}

keep-alive/src/test/java/KeepAliveGattTest.kt

@@ -0,0 +1,142 @@
+/*
+ * Copyright 2020 JUUL Labs, Inc.
+ */
+
+package com.juul.able.keepalive.test
+
+import android.bluetooth.BluetoothDevice
+import android.bluetooth.BluetoothGatt
+import android.bluetooth.BluetoothGattCallback
+import com.juul.able.Able
+import com.juul.able.android.connectGatt
+import com.juul.able.device.ConnectGattResult
+import com.juul.able.gatt.Gatt
+import com.juul.able.keepalive.KeepAliveGatt
+import com.juul.able.keepalive.State.Connected
+import com.juul.able.keepalive.State.Connecting
+import com.juul.able.keepalive.keepAliveGatt
+import com.juul.able.logger.Logger
+import io.mockk.coEvery
+import io.mockk.coVerify
+import io.mockk.every
+import io.mockk.mockk
+import io.mockk.mockkStatic
+import io.mockk.slot
+import io.mockk.unmockkStatic
+import io.mockk.verify
+import kotlinx.coroutines.CoroutineStart.LAZY
+import kotlinx.coroutines.cancelAndJoin
+import kotlinx.coroutines.coroutineScope
+import kotlinx.coroutines.delay
+import kotlinx.coroutines.flow.first
+import kotlinx.coroutines.flow.firstOrNull
+import kotlinx.coroutines.flow.flow
+import kotlinx.coroutines.launch
+import kotlinx.coroutines.runBlocking
+import kotlinx.coroutines.yield
+import java.util.concurrent.atomic.AtomicReference
+import kotlin.test.BeforeTest
+import kotlin.test.Test
+import kotlin.test.assertEquals
+import kotlin.test.assertFalse
+import kotlin.test.assertNull
+
+private const val DISCONNECT_TIMEOUT = 5_000L
+
+class KeepAliveGattTest {
+
+    @BeforeTest
+    fun setup() {
+        Able.logger = object : Logger {
+            override fun isLoggable(priority: Int): Boolean = true
+            override fun log(priority: Int, throwable: Throwable?, message: String) {
+                println("[$priority] $message")
+                throwable?.printStackTrace()
+            }
+        }
+    }
+
+    @Test
+    fun `When parent scope is lazy, connect occurs when parent is started`() = runBlocking {
+        val callbackSlot = slot<BluetoothGattCallback>()
+        lateinit var bluetoothGatt: BluetoothGatt
+        val bluetoothDevice = mockk<BluetoothDevice> {
+            bluetoothGatt = createBluetoothGatt(this@mockk)
+            every { connectGatt(any(), false, capture(callbackSlot)) } returns bluetoothGatt
+            every { this@mockk.toString() } returns "00:11:22:33:FF:EE"
+        }
+
+        val gatt = AtomicReference<KeepAliveGatt>()
+        val job = launch(start = LAZY) {
+            gatt.set(keepAliveGatt(mockk(relaxed = true), bluetoothDevice, DISCONNECT_TIMEOUT) {})
+        }
+
+        delay(500L)
+        assertFalse(callbackSlot.isCaptured)
+        assertNull(gatt.get())
+
+        job.start()
+
+        // Wait for `CoroutinesGatt` to spin up and provide us with the `GattCallback`.
+        while (!callbackSlot.isCaptured) yield()
+
+        assertEquals(
+            expected = Connecting,
+            actual = gatt.get().state.firstOrNull()
+        )
+    }
+
+    @Test
+    fun `When Coroutine is cancelled, Gatt is disconnected`() = runBlocking {
+        val bluetoothDevice = mockk<BluetoothDevice> {
+            every { this@mockk.toString() } returns "00:11:22:33:FF:EE"
+        }
+        val gatt = mockk<Gatt> {
+            every { onCharacteristicChanged } returns flow { delay(Long.MAX_VALUE) }
+            coEvery { disconnect() } returns Unit
+        }
+
+        mockkStatic("com.juul.able.android.BluetoothDeviceKt")
+        try {
+            coEvery { bluetoothDevice.connectGatt(any()) } returns ConnectGattResult.Success(gatt)
+
+            coroutineScope {
+                val result = AtomicReference<KeepAliveGatt>()
+                val job = launch {
+                    result.set(keepAliveGatt(
+                        androidContext = mockk(relaxed = true),
+                        device = bluetoothDevice,
+                        disconnectTimeoutMillis = DISCONNECT_TIMEOUT,
+                        onConnectAction = {}
+                    ))
+                }
+
+                val keepAliveGatt = result.awaitNonNull()
+                keepAliveGatt.state.first { it == Connected }
+                job.cancel()
+            }
+
+            coVerify {
+                gatt.disconnect()
+            }
+        } finally {
+            unmockkStatic("com.juul.able.android.BluetoothDeviceKt")
+        }
+    }
+}
+
+private fun createBluetoothGatt(
+    bluetoothDevice: BluetoothDevice
+): BluetoothGatt = mockk {
+    every { device } returns bluetoothDevice
+    every { close() } returns Unit
+}
+
+private suspend fun <T> AtomicReference<T>.awaitNonNull(): T {
+    var value: T?
+    do {
+        yield()
+        value = get()
+    } while (value == null)
+    return value
+}

settings.gradle

@@ -19,3 +19,4 @@ include ':core'
 include ':processor'
 include ':throw'
 include ':timber-logger'
+include ':keep-alive'