refactor(concurrent): Introduce SequentialJob to manage service setup (#3983)

Signed-off-by: James Rich <[email protected]>

Author: jamesarich

app/build.gradle.kts

@@ -254,6 +254,7 @@ dependencies {
     androidTestImplementation(libs.hilt.android.testing)
 
     testImplementation(libs.junit)
+    testImplementation(libs.kotlinx.coroutines.test)
 
     dokkaPlugin(libs.dokka.android.documentation.plugin)
 }

app/src/main/java/com/geeksville/mesh/MeshServiceClient.kt

@@ -25,12 +25,11 @@ import androidx.lifecycle.LifecycleOwner
 import androidx.lifecycle.lifecycleScope
 import com.geeksville.mesh.android.BindFailedException
 import com.geeksville.mesh.android.ServiceClient
-import com.geeksville.mesh.concurrent.handledLaunch
+import com.geeksville.mesh.concurrent.SequentialJob
 import com.geeksville.mesh.service.MeshService
 import com.geeksville.mesh.service.startService
 import dagger.hilt.android.qualifiers.ActivityContext
 import dagger.hilt.android.scopes.ActivityScoped
-import kotlinx.coroutines.Job
 import org.meshtastic.core.service.IMeshService
 import org.meshtastic.core.service.ServiceRepository
 import timber.log.Timber
@@ -43,14 +42,12 @@ class MeshServiceClient
 constructor(
     @ActivityContext private val context: Context,
     private val serviceRepository: ServiceRepository,
+    private val serviceSetupJob: SequentialJob,
 ) : ServiceClient<IMeshService>(IMeshService.Stub::asInterface),
     DefaultLifecycleObserver {
 
     private val lifecycleOwner: LifecycleOwner = context as LifecycleOwner
 
-    // TODO Inject this for ease of testing
-    private var serviceSetupJob: Job? = null
-
     init {
         Timber.d("Adding self as LifecycleObserver for $lifecycleOwner")
         lifecycleOwner.lifecycle.addObserver(this)
@@ -59,16 +56,14 @@ constructor(
     // region ServiceClient overrides
 
     override fun onConnected(service: IMeshService) {
-        serviceSetupJob?.cancel()
-        serviceSetupJob =
-            lifecycleOwner.lifecycleScope.handledLaunch {
-                serviceRepository.setMeshService(service)
-                Timber.d("connected to mesh service, connectionState=${serviceRepository.connectionState.value}")
-            }
+        serviceSetupJob.launch(lifecycleOwner.lifecycleScope) {
+            serviceRepository.setMeshService(service)
+            Timber.d("connected to mesh service, connectionState=${serviceRepository.connectionState.value}")
+        }
     }
 
     override fun onDisconnected() {
-        serviceSetupJob?.cancel()
+        serviceSetupJob.cancel()
         serviceRepository.setMeshService(null)
     }
 

app/src/main/java/com/geeksville/mesh/concurrent/SequentialJob.kt

@@ -0,0 +1,51 @@
+/*
+ * Copyright (c) 2025 Meshtastic LLC
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this program.  If not, see <https://www.gnu.org/licenses/>.
+ */
+
+package com.geeksville.mesh.concurrent
+
+import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.Job
+import java.util.concurrent.atomic.AtomicReference
+import javax.inject.Inject
+
+/**
+ * A helper class that manages a single [Job].
+ *
+ * When a new job is launched, the previous one is cancelled. This is useful for ensuring that only one operation of a
+ * certain type is running at a time.
+ */
+class SequentialJob @Inject constructor() {
+    private val job = AtomicReference<Job?>(null)
+
+    /**
+     * Cancels the previous job (if any) and launches a new one in the given [scope].
+     *
+     * The new job uses [handledLaunch] to ensure exceptions are reported.
+     */
+    fun launch(scope: CoroutineScope, block: suspend CoroutineScope.() -> Unit) {
+        cancel()
+        val newJob = scope.handledLaunch(block = block)
+        job.set(newJob)
+
+        newJob.invokeOnCompletion { job.compareAndSet(newJob, null) }
+    }
+
+    /** Cancels the current job. */
+    fun cancel() {
+        job.getAndSet(null)?.cancel()
+    }
+}

app/src/test/java/com/geeksville/mesh/concurrent/SequentialJobTest.kt

@@ -0,0 +1,81 @@
+/*
+ * Copyright (c) 2025 Meshtastic LLC
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this program.  If not, see <https://www.gnu.org/licenses/>.
+ */
+
+package com.geeksville.mesh.concurrent
+
+import kotlinx.coroutines.ExperimentalCoroutinesApi
+import kotlinx.coroutines.delay
+import kotlinx.coroutines.test.advanceTimeBy
+import kotlinx.coroutines.test.runTest
+import org.junit.Assert.assertTrue
+import org.junit.Test
+
+@ExperimentalCoroutinesApi
+class SequentialJobTest {
+
+    private val sequentialJob = SequentialJob()
+
+    @Test
+    fun `launch cancels previous job`() = runTest {
+        var job1Active = false
+        var job1Cancelled = false
+
+        // Launch first job
+        sequentialJob.launch(this) {
+            try {
+                job1Active = true
+                delay(1000)
+            } finally {
+                job1Cancelled = true
+            }
+        }
+
+        advanceTimeBy(100)
+        assertTrue("Job 1 should be active", job1Active)
+
+        // Launch second job
+        sequentialJob.launch(this) {
+            // Do nothing
+        }
+
+        advanceTimeBy(100)
+        assertTrue("Job 1 should be cancelled", job1Cancelled)
+    }
+
+    @Test
+    fun `cancel stops the job`() = runTest {
+        var jobActive = false
+        var jobCancelled = false
+
+        sequentialJob.launch(this) {
+            try {
+                jobActive = true
+                delay(1000)
+            } finally {
+                jobCancelled = true
+            }
+        }
+
+        advanceTimeBy(100)
+        assertTrue("Job should be active", jobActive)
+
+        sequentialJob.cancel()
+
+        advanceTimeBy(100)
+        assertTrue("Job should be cancelled", jobCancelled)
+    }
+}

gradle/libs.versions.toml

@@ -111,6 +111,7 @@ truth = { module = "com.google.truth:truth", version = "1.4.5" }
 kotlin-gradlePlugin = { module = "org.jetbrains.kotlin:kotlin-gradle-plugin", version.ref = "kotlin" }
 kotlinx-collections-immutable = { module = "org.jetbrains.kotlinx:kotlinx-collections-immutable", version = "0.4.0" }
 kotlinx-coroutines-core = { module = "org.jetbrains.kotlinx:kotlinx-coroutines-core", version.ref = "kotlinx-coroutines-android" }
+kotlinx-coroutines-test = { module = "org.jetbrains.kotlinx:kotlinx-coroutines-test", version.ref = "kotlinx-coroutines-android" }
 kotlinx-serialization-core = { module = "org.jetbrains.kotlinx:kotlinx-serialization-core", version.ref = "kotlinx-serialization" }
 kotlinx-serialization-json = { module = "org.jetbrains.kotlinx:kotlinx-serialization-json", version.ref = "kotlinx-serialization" }