Only listen on data type changes tha have active reminders (#47)

app/src/main/kotlin/de/timklge/karooreminder/Extensions.kt

@@ -1,5 +1,11 @@
 package de.timklge.karooreminder
 
+import android.content.Context
+import android.util.Log
+import de.timklge.karooreminder.KarooReminderExtension.Companion.TAG
+import de.timklge.karooreminder.screens.Reminder
+import de.timklge.karooreminder.screens.defaultReminders
+import de.timklge.karooreminder.screens.preferencesKey
 import io.hammerhead.karooext.KarooSystemService
 import io.hammerhead.karooext.models.OnStreamState
 import io.hammerhead.karooext.models.RideState
@@ -9,6 +15,8 @@ import kotlinx.coroutines.channels.awaitClose
 import kotlinx.coroutines.channels.trySendBlocking
 import kotlinx.coroutines.flow.Flow
 import kotlinx.coroutines.flow.callbackFlow
+import kotlinx.coroutines.flow.map
+import kotlinx.serialization.json.Json
 
 fun KarooSystemService.streamDataFlow(dataTypeId: String): Flow<StreamState> {
     return callbackFlow {
@@ -41,4 +49,17 @@ fun KarooSystemService.streamUserProfile(): Flow<UserProfile> {
             removeConsumer(listenerId)
         }
     }
+}
+
+fun Context.streamPreferences(): Flow<MutableList<Reminder>> {
+    return applicationContext.dataStore.data.map { remindersJson ->
+        try {
+            Json.decodeFromString<MutableList<Reminder>>(
+                remindersJson[preferencesKey] ?: defaultReminders
+            )
+        } catch(e: Throwable){
+            Log.e(TAG,"Failed to read preferences", e)
+            mutableListOf()
+        }
+    }
 }

app/src/main/kotlin/de/timklge/karooreminder/KarooReminderExtension.kt

@@ -119,8 +119,6 @@ class KarooReminderExtension : KarooExtension("karoo-reminder", BuildConfig.VERS
 
     private lateinit var karooSystem: KarooSystemService
 
-    private var jobs: MutableSet<Job> = mutableSetOf()
-
     data class DisplayedReminder(val beepPattern: ReminderBeepPattern, val trigger: ReminderTrigger, val alert: InRideAlert)
 
     private var reminderChannel = Channel<DisplayedReminder>(2, BufferOverflow.DROP_OLDEST)
@@ -163,6 +161,9 @@ class KarooReminderExtension : KarooExtension("karoo-reminder", BuildConfig.VERS
         }
     }
 
+    private lateinit var receiveJob: Job
+    private lateinit var triggerStreamJob: Job
+
     override fun onCreate() {
         super.onCreate()
 
@@ -170,10 +171,9 @@ class KarooReminderExtension : KarooExtension("karoo-reminder", BuildConfig.VERS
 
         karooSystem = KarooSystemService(applicationContext)
 
-        val receiveJob = CoroutineScope(Dispatchers.IO).launch {
+        receiveJob = CoroutineScope(Dispatchers.IO).launch {
             receiverWorker()
         }
-        jobs.add(receiveJob)
 
         karooSystem.connect { connected ->
             if (connected) {
@@ -181,75 +181,88 @@ class KarooReminderExtension : KarooExtension("karoo-reminder", BuildConfig.VERS
             }
         }
 
-        val distanceJob = startIntervalJob(ReminderTrigger.DISTANCE) {
+        val triggerJobs = mutableSetOf<Job>()
-            karooSystem.streamDataFlow(DataType.Type.DISTANCE)
+
-                .mapNotNull { (it as? StreamState.Streaming)?.dataPoint?.singleValue  }
+        triggerStreamJob = CoroutineScope(Dispatchers.IO).launch {
-                .combine(karooSystem.streamUserProfile()) { distance, profile -> distance to profile }
+            streamPreferences().collect { reminders ->
-                .map { (distance, profile) ->
+                triggerJobs.forEach { it.cancel() }
-                    if (profile.preferredUnit.distance == UserProfile.PreferredUnit.UnitType.IMPERIAL){
+                triggerJobs.clear()
-                        (distance / 1609.344).toInt()
+
-                    } else {
+                if (reminders.any { it.trigger == ReminderTrigger.DISTANCE }){
-                        (distance / 1000.0).toInt()
+                    val distanceJob = startIntervalJob(ReminderTrigger.DISTANCE) {
+                        karooSystem.streamDataFlow(DataType.Type.DISTANCE)
+                            .mapNotNull { (it as? StreamState.Streaming)?.dataPoint?.singleValue  }
+                            .combine(karooSystem.streamUserProfile()) { distance, profile -> distance to profile }
+                            .map { (distance, profile) ->
+                                if (profile.preferredUnit.distance == UserProfile.PreferredUnit.UnitType.IMPERIAL){
+                                    (distance / 1609.344).toInt()
+                                } else {
+                                    (distance / 1000.0).toInt()
+                                }
+                            }
+                            .distinctUntilChanged()
+                            .filterNot { it == 0 }
+                    }
+                    triggerJobs.add(distanceJob)
+                }
+
+                if (reminders.any { it.trigger == ReminderTrigger.ELAPSED_TIME }){
+                    val elapsedTimeJob = startIntervalJob(ReminderTrigger.ELAPSED_TIME) {
+                        karooSystem.streamDataFlow(DataType.Type.ELAPSED_TIME)
+                            .mapNotNull { (it as? StreamState.Streaming)?.dataPoint?.singleValue  }
+                            .map { (it / 1000 / 60).toInt() }
+                            .distinctUntilChanged()
+                            .filterNot { it == 0 }
+                    }
+                    triggerJobs.add(elapsedTimeJob)
+                }
+
+                if (reminders.any { it.trigger == ReminderTrigger.ENERGY_OUTPUT }){
+                    val energyOutputJob = startIntervalJob(ReminderTrigger.ENERGY_OUTPUT) {
+                        karooSystem.streamDataFlow(DataType.Type.ENERGY_OUTPUT)
+                            .mapNotNull { (it as? StreamState.Streaming)?.dataPoint?.singleValue  }
+                            .map { it.toInt() }
+                            .distinctUntilChanged()
+                            .filterNot { it == 0 }
+                            .allIntermediateInts()
+                    }
+                    triggerJobs.add(energyOutputJob)
+                }
+
+                val intervalTriggers = setOf(
+                    ReminderTrigger.POWER_LIMIT_MAXIMUM_EXCEEDED,
+                    ReminderTrigger.POWER_LIMIT_MINIMUM_EXCEEDED,
+                    ReminderTrigger.HR_LIMIT_MAXIMUM_EXCEEDED,
+                    ReminderTrigger.HR_LIMIT_MINIMUM_EXCEEDED,
+                    ReminderTrigger.SPEED_LIMIT_MAXIMUM_EXCEEDED,
+                    ReminderTrigger.SPEED_LIMIT_MINIMUM_EXCEEDED,
+                    ReminderTrigger.CADENCE_LIMIT_MAXIMUM_EXCEEDED,
+                    ReminderTrigger.CADENCE_LIMIT_MINIMUM_EXCEEDED,
+                    ReminderTrigger.AMBIENT_TEMPERATURE_LIMIT_MAXIMUM_EXCEEDED,
+                    ReminderTrigger.AMBIENT_TEMPERATURE_LIMIT_MINIMUM_EXCEEDED,
+                    ReminderTrigger.GRADIENT_LIMIT_MAXIMUM_EXCEEDED,
+                    ReminderTrigger.GRADIENT_LIMIT_MINIMUM_EXCEEDED,
+                    ReminderTrigger.FRONT_TIRE_PRESSURE_LIMIT_MAXIMUM_EXCEEDED,
+                    ReminderTrigger.FRONT_TIRE_PRESSURE_LIMIT_MINIMUM_EXCEEDED,
+                    ReminderTrigger.REAR_TIRE_PRESSURE_LIMIT_MAXIMUM_EXCEEDED,
+                    ReminderTrigger.REAR_TIRE_PRESSURE_LIMIT_MINIMUM_EXCEEDED,
+                    ReminderTrigger.CORE_TEMPERATURE_LIMIT_MAXIMUM_EXCEEDED,
+                    ReminderTrigger.CORE_TEMPERATURE_LIMIT_MINIMUM_EXCEEDED
+                )
+
+                intervalTriggers.forEach { trigger ->
+                    if (reminders.any { it.trigger == trigger }){
+                        val job = startRangeExceededJob(trigger)
+                        triggerJobs.add(job)
                     }
                 }
-                .distinctUntilChanged()
+            }
-                .filterNot { it == 0 }
         }
-        jobs.add(distanceJob)
-
-        jobs.addAll(listOf(
-            startRangeExceededJob(ReminderTrigger.POWER_LIMIT_MAXIMUM_EXCEEDED),
-            startRangeExceededJob(ReminderTrigger.HR_LIMIT_MAXIMUM_EXCEEDED),
-            startRangeExceededJob(ReminderTrigger.POWER_LIMIT_MINIMUM_EXCEEDED),
-            startRangeExceededJob(ReminderTrigger.HR_LIMIT_MINIMUM_EXCEEDED),
-            startRangeExceededJob(ReminderTrigger.SPEED_LIMIT_MAXIMUM_EXCEEDED),
-            startRangeExceededJob(ReminderTrigger.SPEED_LIMIT_MINIMUM_EXCEEDED),
-            startRangeExceededJob(ReminderTrigger.CADENCE_LIMIT_MAXIMUM_EXCEEDED),
-            startRangeExceededJob(ReminderTrigger.CADENCE_LIMIT_MINIMUM_EXCEEDED),
-            startRangeExceededJob(ReminderTrigger.CORE_TEMPERATURE_LIMIT_MAXIMUM_EXCEEDED),
-            startRangeExceededJob(ReminderTrigger.CORE_TEMPERATURE_LIMIT_MINIMUM_EXCEEDED),
-            startRangeExceededJob(ReminderTrigger.AMBIENT_TEMPERATURE_LIMIT_MAXIMUM_EXCEEDED),
-            startRangeExceededJob(ReminderTrigger.AMBIENT_TEMPERATURE_LIMIT_MINIMUM_EXCEEDED),
-            startRangeExceededJob(ReminderTrigger.FRONT_TIRE_PRESSURE_LIMIT_MAXIMUM_EXCEEDED),
-            startRangeExceededJob(ReminderTrigger.FRONT_TIRE_PRESSURE_LIMIT_MINIMUM_EXCEEDED),
-            startRangeExceededJob(ReminderTrigger.REAR_TIRE_PRESSURE_LIMIT_MAXIMUM_EXCEEDED),
-            startRangeExceededJob(ReminderTrigger.REAR_TIRE_PRESSURE_LIMIT_MINIMUM_EXCEEDED),
-            startRangeExceededJob(ReminderTrigger.GRADIENT_LIMIT_MAXIMUM_EXCEEDED),
-            startRangeExceededJob(ReminderTrigger.GRADIENT_LIMIT_MINIMUM_EXCEEDED)
-        ))
-
-        val elapsedTimeJob = startIntervalJob(ReminderTrigger.ELAPSED_TIME) {
-            karooSystem.streamDataFlow(DataType.Type.ELAPSED_TIME)
-                .mapNotNull { (it as? StreamState.Streaming)?.dataPoint?.singleValue  }
-                .map { (it / 1000 / 60).toInt() }
-                .distinctUntilChanged()
-                .filterNot { it == 0 }
-        }
-        jobs.add(elapsedTimeJob)
-
-        val energyOutputJob = startIntervalJob(ReminderTrigger.ENERGY_OUTPUT) {
-            karooSystem.streamDataFlow(DataType.Type.ENERGY_OUTPUT)
-                .mapNotNull { (it as? StreamState.Streaming)?.dataPoint?.singleValue  }
-                .map { it.toInt() }
-                .distinctUntilChanged()
-                .filterNot { it == 0 }
-                .allIntermediateInts()
-        }
-        jobs.add(energyOutputJob)
     }
 
     private fun startIntervalJob(trigger: ReminderTrigger, flow: () -> Flow<Int>): Job {
         return CoroutineScope(Dispatchers.IO).launch {
-            val preferences = applicationContext.dataStore.data.map { remindersJson ->
+            val preferences = streamPreferences()
-                try {
-                    Json.decodeFromString<MutableList<Reminder>>(
-                        remindersJson[preferencesKey] ?: defaultReminders
-                    )
-                } catch(e: Throwable){
-                    Log.e(TAG,"Failed to read preferences", e)
-                    mutableListOf()
-                }
-            }
 
             flow()
                 .filterNot { it == 0 }
@@ -280,16 +293,7 @@ class KarooReminderExtension : KarooExtension("karoo-reminder", BuildConfig.VERS
 
     private fun startRangeExceededJob(triggerType: ReminderTrigger): Job {
         return CoroutineScope(Dispatchers.IO).launch {
-            val preferences = applicationContext.dataStore.data.map { remindersJson ->
+            val preferences = streamPreferences()
-                try {
-                    Json.decodeFromString<MutableList<Reminder>>(
-                        remindersJson[preferencesKey] ?: defaultReminders
-                    )
-                } catch (e: Throwable) {
-                    Log.e(TAG, "Failed to read preferences", e)
-                    mutableListOf()
-                }
-            }
 
             val dataType = when (triggerType) {
                 ReminderTrigger.HR_LIMIT_MAXIMUM_EXCEEDED, ReminderTrigger.HR_LIMIT_MINIMUM_EXCEEDED -> DataType.Type.HEART_RATE
@@ -426,8 +430,8 @@ class KarooReminderExtension : KarooExtension("karoo-reminder", BuildConfig.VERS
     }
 
     override fun onDestroy() {
-        jobs.forEach { job -> job.cancel() }
+        receiveJob.cancel()
-        jobs.clear()
+        triggerStreamJob.cancel()
 
         karooSystem.disconnect()
         super.onDestroy()