chore: Improve PropertyListener for Wildcard Subscribes

BREAKING CHANGE: - Method PropertyListener#onPropertyChanged(String, Field, DataEntry) was removed - Method PropertyListener#onPropertyChanged(List<EntryUpdate>) was added - Previous DataEntry can be retrieved via EntryUpdate#entry - Subscribing to a branch will initially send out an update containing a list of all leafs
eclipse-kuksa · Jan 31, 2024 · bb92476 · bb92476
1 parent de30bd5
commit bb92476
Show file tree

Hide file tree

Showing 13 changed files with 157 additions and 87 deletions.
diff --git a/app/src/main/kotlin/org/eclipse/kuksa/testapp/KuksaDataBrokerActivity.kt b/app/src/main/kotlin/org/eclipse/kuksa/testapp/KuksaDataBrokerActivity.kt
@@ -40,7 +40,6 @@ import org.eclipse.kuksa.model.Property
 import org.eclipse.kuksa.proto.v1.KuksaValV1
 import org.eclipse.kuksa.proto.v1.KuksaValV1.GetResponse
 import org.eclipse.kuksa.proto.v1.Types
-import org.eclipse.kuksa.proto.v1.Types.DataEntry
 import org.eclipse.kuksa.proto.v1.Types.Datapoint
 import org.eclipse.kuksa.proto.v1.Types.Field
 import org.eclipse.kuksa.testapp.databroker.DataBrokerEngine
@@ -94,9 +93,9 @@ class KuksaDataBrokerActivity : ComponentActivity() {
     }
 
     private val propertyListener = object : PropertyListener {
-        override fun onPropertyChanged(vssPath: String, field: Field, updatedValue: DataEntry) {
-            Log.d(TAG, "onPropertyChanged path: vssPath = $vssPath, field = $field, changedValue = $updatedValue")
-            outputViewModel.addOutputEntry("Updated value: $updatedValue")
+        override fun onPropertyChanged(entryUpdates: List<KuksaValV1.EntryUpdate>) {
+            Log.d(TAG, "onPropertyChanged() called with: updatedValues = $entryUpdates")
+            outputViewModel.addOutputEntry("Updated value: $entryUpdates")
         }
 
         override fun onError(throwable: Throwable) {

diff --git a/gradle.properties b/gradle.properties
@@ -6,6 +6,7 @@ android.useAndroidX=true
 kotlin.code.style=official
 
 org.gradle.jvmargs=-Xmx2048m
+kotlin.daemon.jvmargs=-Xmx2048m
 
 # When using compose + ksp the incremental compiler should be disabled:  https://issuetracker.google.com/issues/207185051
 ksp.incremental=false
diff --git a/kuksa-sdk/src/main/kotlin/org/eclipse/kuksa/DataBrokerTransporter.kt b/kuksa-sdk/src/main/kotlin/org/eclipse/kuksa/DataBrokerTransporter.kt
@@ -149,12 +149,8 @@ internal class DataBrokerTransporter(
         val subscription = Subscription(vssPath, field, cancellableContext)
         val streamObserver = object : StreamObserver<SubscribeResponse> {
             override fun onNext(value: SubscribeResponse) {
-                for (entryUpdate in value.updatesList) {
-                    val entry = entryUpdate.entry
-
-                    subscription.listeners.forEach { observer ->
-                        observer.onPropertyChanged(entry.path, field, entry)
-                    }
+                subscription.listeners.forEach { observer ->
+                    observer.onPropertyChanged(value.updatesList)
                 }
 
                 subscription.lastSubscribeResponse = value

diff --git a/kuksa-sdk/src/main/kotlin/org/eclipse/kuksa/PropertyListener.kt b/kuksa-sdk/src/main/kotlin/org/eclipse/kuksa/PropertyListener.kt
@@ -20,18 +20,19 @@
 package org.eclipse.kuksa
 
 import org.eclipse.kuksa.pattern.listener.Listener
-import org.eclipse.kuksa.proto.v1.Types.DataEntry
-import org.eclipse.kuksa.proto.v1.Types.Field
+import org.eclipse.kuksa.proto.v1.KuksaValV1
 import org.eclipse.kuksa.vsscore.model.VssSpecification
 
 /**
- * The Listener is used to notify about changes to subscribed properties.
+ * The Listener is used to notify about changes to subscribed properties. When registering the listener to
+ * Vehicle.ADAS.ABS this listener will also be notified about changes of sub-properties e.g. Vehicle.ADAS.ABS.IsEnabled
+ * or Vehicle.ADAS.ABS.IsEngaged.
  */
 interface PropertyListener : Listener {
     /**
-     * Will be triggered with the [updatedValue] when the underlying [field] of the [vssPath] changed it's value.
+     * Will be triggered with a list of [entryUpdates] of the corresponding field.
      */
-    fun onPropertyChanged(vssPath: String, field: Field, updatedValue: DataEntry)
+    fun onPropertyChanged(entryUpdates: List<KuksaValV1.EntryUpdate>)
 
     /**
      * Will be triggered when an error happens during subscription and forwards the [throwable].

diff --git a/kuksa-sdk/src/main/kotlin/org/eclipse/kuksa/subscription/DataBrokerSubscriber.kt b/kuksa-sdk/src/main/kotlin/org/eclipse/kuksa/subscription/DataBrokerSubscriber.kt
@@ -29,7 +29,6 @@ import org.eclipse.kuksa.proto.v1.Types
 import org.eclipse.kuksa.proto.v1.Types.Field
 import org.eclipse.kuksa.vsscore.model.VssProperty
 import org.eclipse.kuksa.vsscore.model.VssSpecification
-import org.eclipse.kuksa.vsscore.model.vssProperties
 
 /**
  * Creates [Subscription]s to the DataBroker to get notified about changes on the underlying vssPaths and fields.
@@ -93,10 +92,7 @@ internal class DataBrokerSubscriber(private val dataBrokerTransporter: DataBroke
     ) {
         val vssPath = specification.vssPath
 
-        val leafProperties = specification.vssProperties
-        val childPropertiesPaths = leafProperties.map { it.vssPath }
-
-        val specificationPropertyListener = SpecificationPropertyListener(specification, childPropertiesPaths, listener)
+        val specificationPropertyListener = SpecificationPropertyListener(specification, listener)
         subscribe(vssPath, field, specificationPropertyListener)
     }
 
@@ -113,10 +109,7 @@ internal class DataBrokerSubscriber(private val dataBrokerTransporter: DataBroke
     ) {
         val vssPath = specification.vssPath
 
-        val leafProperties = specification.vssProperties
-        val childPropertiesPaths = leafProperties.map { it.vssPath }
-
-        val specificationPropertyListener = SpecificationPropertyListener(specification, childPropertiesPaths, listener)
+        val specificationPropertyListener = SpecificationPropertyListener(specification, listener)
         unsubscribe(vssPath, field, specificationPropertyListener)
     }
 

diff --git a/kuksa-sdk/src/main/kotlin/org/eclipse/kuksa/subscription/SpecificationPropertyListener.kt b/kuksa-sdk/src/main/kotlin/org/eclipse/kuksa/subscription/SpecificationPropertyListener.kt
@@ -19,48 +19,27 @@
 
 package org.eclipse.kuksa.subscription
 
-import android.util.Log
-import kotlinx.coroutines.Dispatchers
-import kotlinx.coroutines.ExperimentalCoroutinesApi
-import kotlinx.coroutines.runBlocking
 import org.eclipse.kuksa.PropertyListener
 import org.eclipse.kuksa.VssSpecificationListener
-import org.eclipse.kuksa.extension.TAG
 import org.eclipse.kuksa.extension.copy
-import org.eclipse.kuksa.proto.v1.Types
+import org.eclipse.kuksa.proto.v1.KuksaValV1
 import org.eclipse.kuksa.vsscore.model.VssSpecification
 
 internal class SpecificationPropertyListener<T : VssSpecification>(
     specification: T,
-    vssPaths: Collection<String>,
     private val listener: VssSpecificationListener<T>,
 ) : PropertyListener {
-    // Reduces the load on the observer for big VssSpecifications. We wait for the initial update
-    // of all VssProperties before notifying the observer about the first batch
-    private val initialSubscriptionUpdates = vssPaths.associateWith { false }.toMutableMap()
-
     // This is currently needed because we get multiple subscribe responses for every heir. Otherwise we
     // would override the last heir value with every new response.
     private var updatedVssSpecification: T = specification
 
-    // Multiple onPropertyChanged updates from different threads may be called. The updatedVssSpecification must be
-    // in sync however. Calling the .copy in a blocking context is necessary for this.
-    @OptIn(ExperimentalCoroutinesApi::class)
-    private val specificationUpdateContext = Dispatchers.IO.limitedParallelism(1)
-
-    override fun onPropertyChanged(vssPath: String, field: Types.Field, updatedValue: Types.DataEntry) {
-        Log.d(TAG, "Update from subscribed property: $vssPath - $field: ${updatedValue.value}")
-
-        runBlocking(specificationUpdateContext) {
-            updatedVssSpecification = updatedVssSpecification.copy(vssPath, updatedValue.value)
+    override fun onPropertyChanged(entryUpdates: List<KuksaValV1.EntryUpdate>) {
+        entryUpdates.forEach { entryUpdate ->
+            val dataEntry = entryUpdate.entry
+            updatedVssSpecification = updatedVssSpecification.copy(dataEntry.path, dataEntry.value)
         }
 
-        initialSubscriptionUpdates[vssPath] = true
-        val isInitialSubscriptionComplete = initialSubscriptionUpdates.values.all { it }
-        if (isInitialSubscriptionComplete) {
-            Log.d(TAG, "Update for subscribed specification complete: ${updatedVssSpecification.vssPath}")
-            listener.onSpecificationChanged(updatedVssSpecification)
-        }
+        listener.onSpecificationChanged(updatedVssSpecification)
     }
 
     override fun onError(throwable: Throwable) {

diff --git a/kuksa-sdk/src/main/kotlin/org/eclipse/kuksa/subscription/Subscription.kt b/kuksa-sdk/src/main/kotlin/org/eclipse/kuksa/subscription/Subscription.kt
@@ -48,10 +48,7 @@ internal class Subscription(
             } else {
                 val lastSubscribeResponse = lastSubscribeResponse ?: return@MultiListener
 
-                for (entryUpdate in lastSubscribeResponse.updatesList) {
-                    val entry = entryUpdate.entry
-                    observer.onPropertyChanged(entry.path, field, entry)
-                }
+                observer.onPropertyChanged(lastSubscribeResponse.updatesList)
             }
         },
     )

diff --git a/kuksa-sdk/src/test/kotlin/org/eclipse/kuksa/DataBrokerConnectionTest.kt b/kuksa-sdk/src/test/kotlin/org/eclipse/kuksa/DataBrokerConnectionTest.kt
@@ -32,6 +32,7 @@ import io.mockk.verify
 import kotlinx.coroutines.runBlocking
 import org.eclipse.kuksa.databroker.DataBrokerConnectorProvider
 import org.eclipse.kuksa.model.Property
+import org.eclipse.kuksa.proto.v1.KuksaValV1
 import org.eclipse.kuksa.proto.v1.Types
 import org.eclipse.kuksa.proto.v1.Types.Datapoint
 import org.eclipse.kuksa.test.kotest.Integration
@@ -48,17 +49,23 @@ class DataBrokerConnectionTest : BehaviorSpec({
         val dataBrokerConnection = connectToDataBrokerBlocking()
 
         and("A Property with a valid VSS Path") {
+            val vssPath = "Vehicle.Acceleration.Lateral"
             val fields = listOf(Types.Field.FIELD_VALUE)
-            val property = Property("Vehicle.Acceleration.Lateral", fields)
+            val property = Property(vssPath, fields)
 
             `when`("Subscribing to the Property") {
                 val propertyListener = mockk<PropertyListener>(relaxed = true)
                 dataBrokerConnection.subscribe(property, propertyListener)
 
                 then("The #onPropertyChanged method is triggered") {
+                    val capturingSlot = slot<List<KuksaValV1.EntryUpdate>>()
                     verify(timeout = 100L) {
-                        propertyListener.onPropertyChanged(any(), any(), any())
+                        propertyListener.onPropertyChanged(capture(capturingSlot))
                     }
+
+                    val entryUpdates = capturingSlot.captured
+                    entryUpdates.size shouldBe 1
+                    entryUpdates[0].entry.path shouldBe vssPath
                 }
 
                 `when`("The observed Property changes") {
@@ -70,14 +77,14 @@ class DataBrokerConnectionTest : BehaviorSpec({
                     dataBrokerConnection.update(property, datapoint)
 
                     then("The #onPropertyChanged callback is triggered with the new value") {
-                        val capturingSlot = slot<Types.DataEntry>()
+                        val capturingSlot = slot<List<KuksaValV1.EntryUpdate>>()
 
                         verify(timeout = 100) {
-                            propertyListener.onPropertyChanged(any(), any(), capture(capturingSlot))
+                            propertyListener.onPropertyChanged(capture(capturingSlot))
                         }
 
-                        val dataEntry = capturingSlot.captured
-                        val capturedDatapoint = dataEntry.value
+                        val entryUpdates = capturingSlot.captured
+                        val capturedDatapoint = entryUpdates[0].entry.value
                         val float = capturedDatapoint.float
 
                         assertEquals(newValue, float, 0.0001f)

diff --git a/kuksa-sdk/src/test/kotlin/org/eclipse/kuksa/DataBrokerTransporterTest.kt b/kuksa-sdk/src/test/kotlin/org/eclipse/kuksa/DataBrokerTransporterTest.kt
@@ -111,7 +111,7 @@ class DataBrokerTransporterTest : BehaviorSpec({
 
                         then("The PropertyListener should be notified") {
                             verify {
-                                propertyListener.onPropertyChanged(vssPath, Types.Field.FIELD_VALUE, any())
+                                propertyListener.onPropertyChanged(any())
                             }
                         }
                     }

diff --git a/kuksa-sdk/src/test/kotlin/org/eclipse/kuksa/extensions/DataBrokerTransporterExtensions.kt b/kuksa-sdk/src/test/kotlin/org/eclipse/kuksa/extensions/DataBrokerTransporterExtensions.kt
@@ -60,3 +60,22 @@ internal suspend fun DataBrokerTransporter.updateRandomUint32Value(
 
     return randomValue
 }
+
+internal suspend fun DataBrokerTransporter.toggleBoolean(vssPath: String): Boolean {
+    val fields = listOf(Types.Field.FIELD_VALUE)
+
+    var newBoolean: Boolean? = null
+    try {
+        val response = fetch(vssPath, fields)
+        val currentBool = response.entriesList[0].value.bool
+
+        newBoolean = !currentBool
+        val newDatapoint = Types.Datapoint.newBuilder().setBool(newBoolean).build()
+
+        update(vssPath, fields, newDatapoint)
+    } catch (e: Exception) {
+        fail("Updating $vssPath to $newBoolean failed: $e")
+    }
+
+    return newBoolean == true
+}