Remove "Custom Wildcard Subscription"-Logic for Specifications

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,15 @@ 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")
+
+            val entries = mutableListOf<String>().apply {
+                add("Updated Entries")
+                addAll(entryUpdates.map { it.entry.toString() })
+            }
+            val outputEntry = OutputEntry(messages = entries)
+            outputViewModel.addOutputEntry(outputEntry)
         }
 
         override fun onError(throwable: Throwable) {

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

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(vssPath, field, entry)
-                    }
+                subscription.listeners.forEach { observer ->
+                    observer.onPropertyChanged(value.updatesList)
                 }
 
                 subscription.lastSubscribeResponse = value

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].

kuksa-sdk/src/main/kotlin/org/eclipse/kuksa/subscription/DataBrokerSubscriber.kt

@@ -25,7 +25,6 @@ import org.eclipse.kuksa.DataBrokerTransporter
 import org.eclipse.kuksa.PropertyListener
 import org.eclipse.kuksa.VssSpecificationListener
 import org.eclipse.kuksa.extension.TAG
-import org.eclipse.kuksa.extension.createProperties
 import org.eclipse.kuksa.proto.v1.Types
 import org.eclipse.kuksa.proto.v1.Types.Field
 import org.eclipse.kuksa.vsscore.model.VssProperty
@@ -91,13 +90,10 @@ internal class DataBrokerSubscriber(private val dataBrokerTransporter: DataBroke
         field: Field = Field.FIELD_VALUE,
         listener: VssSpecificationListener<T>,
     ) {
-        val leafProperties = specification.createProperties(field)
-        val vssPaths = leafProperties.map { it.vssPath }
+        val vssPath = specification.vssPath
 
-        val specificationPropertyListener = SpecificationPropertyListener(specification, vssPaths, listener)
-        vssPaths.forEach { vssPath ->
-            subscribe(vssPath, field, specificationPropertyListener)
-        }
+        val specificationPropertyListener = SpecificationPropertyListener(specification, listener)
+        subscribe(vssPath, field, specificationPropertyListener)
     }
 
     /**
@@ -111,13 +107,10 @@ internal class DataBrokerSubscriber(private val dataBrokerTransporter: DataBroke
         field: Field = Field.FIELD_VALUE,
         listener: VssSpecificationListener<T>,
     ) {
-        val leafProperties = specification.createProperties(field)
-        val vssPaths = leafProperties.map { it.vssPath }
+        val vssPath = specification.vssPath
 
-        val specificationPropertyListener = SpecificationPropertyListener(specification, vssPaths, listener)
-        vssPaths.forEach { vssPath ->
-            unsubscribe(vssPath, field, specificationPropertyListener)
-        }
+        val specificationPropertyListener = SpecificationPropertyListener(specification, listener)
+        unsubscribe(vssPath, field, specificationPropertyListener)
     }
 
     private companion object {

kuksa-sdk/src/main/kotlin/org/eclipse/kuksa/subscription/SpecificationPropertyListener.kt

@@ -19,49 +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 {
-    // TODO: Remove as soon as the server supports subscribing to vssPaths which are not VssProperties
-    // 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) {

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(vssPath, field, entry)
-                }
+                observer.onPropertyChanged(lastSubscribeResponse.updatesList)
             }
         },
     )

kuksa-sdk/src/test/kotlin/org/eclipse/kuksa/DataBrokerConnectionTest.kt

@@ -21,6 +21,7 @@ package org.eclipse.kuksa
 
 import io.grpc.ConnectivityState
 import io.grpc.ManagedChannel
+import io.kotest.assertions.nondeterministic.eventually
 import io.kotest.core.spec.style.BehaviorSpec
 import io.kotest.matchers.shouldBe
 import io.kotest.matchers.string.shouldContain
@@ -31,7 +32,9 @@ import io.mockk.slot
 import io.mockk.verify
 import kotlinx.coroutines.runBlocking
 import org.eclipse.kuksa.databroker.DataBrokerConnectorProvider
+import org.eclipse.kuksa.mocking.FriendlyVssSpecificationListener
 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
@@ -40,6 +43,7 @@ import org.junit.jupiter.api.Assertions.assertEquals
 import org.junit.jupiter.api.Assertions.assertFalse
 import org.junit.jupiter.api.Assertions.assertTrue
 import kotlin.random.Random
+import kotlin.time.Duration.Companion.seconds
 
 class DataBrokerConnectionTest : BehaviorSpec({
     tags(Integration)
@@ -48,17 +52,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 +80,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)
@@ -154,15 +164,13 @@ class DataBrokerConnectionTest : BehaviorSpec({
             }
 
             `when`("Subscribing to the specification") {
-                val specificationListener =
-                    mockk<VssSpecificationListener<VssDriver>>(relaxed = true)
+                val specificationListener = FriendlyVssSpecificationListener<VssDriver>()
                 dataBrokerConnection.subscribe(specification, listener = specificationListener)
 
                 then("The #onSpecificationChanged method is triggered") {
-                    verify(
-                        timeout = 100L,
-                        exactly = 1,
-                    ) { specificationListener.onSpecificationChanged(any()) }
+                    eventually(1.seconds) {
+                        specificationListener.updatedSpecifications.size shouldBe 1
+                    }
                 }
 
                 and("The initial value is different from the default for a child") {
@@ -172,13 +180,11 @@ class DataBrokerConnectionTest : BehaviorSpec({
                     dataBrokerConnection.update(property, datapoint)
 
                     then("Every child property has been updated with the correct value") {
-                        val capturingList = mutableListOf<VssDriver>()
-
-                        verify(timeout = 100, exactly = 2) {
-                            specificationListener.onSpecificationChanged(capture(capturingList))
+                        eventually(1.seconds) {
+                            specificationListener.updatedSpecifications.size shouldBe 2
                         }
 
-                        val updatedDriver = capturingList.last()
+                        val updatedDriver = specificationListener.updatedSpecifications.last()
                         val heartRate = updatedDriver.heartRate
 
                         heartRate.value shouldBe newHeartRateValue
@@ -192,13 +198,11 @@ class DataBrokerConnectionTest : BehaviorSpec({
                     dataBrokerConnection.update(property, datapoint)
 
                     then("The subscribed Specification should be updated") {
-                        val capturingSlots = mutableListOf<VssDriver>()
-
-                        verify(timeout = 100, exactly = 3) {
-                            specificationListener.onSpecificationChanged(capture(capturingSlots))
+                        eventually(1.seconds) {
+                            specificationListener.updatedSpecifications.size shouldBe 3
                         }
 
-                        val updatedDriver = capturingSlots.last()
+                        val updatedDriver = specificationListener.updatedSpecifications.last()
                         val heartRate = updatedDriver.heartRate
 
                         heartRate.value shouldBe newHeartRateValue

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())
                             }
                         }
                     }

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
+}

kuksa-sdk/src/test/kotlin/org/eclipse/kuksa/mocking/FriendlyPropertyListener.kt

@@ -0,0 +1,40 @@
+/*
+ * Copyright (c) 2024 Contributors to the Eclipse Foundation
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ * SPDX-License-Identifier: Apache-2.0
+ *
+ */
+
+package org.eclipse.kuksa.mocking
+
+import org.eclipse.kuksa.PropertyListener
+import org.eclipse.kuksa.proto.v1.KuksaValV1
+
+class FriendlyPropertyListener : PropertyListener {
+    val updates = mutableListOf<List<KuksaValV1.EntryUpdate>>()
+    val errors = mutableListOf<Throwable>()
+    override fun onPropertyChanged(entryUpdates: List<KuksaValV1.EntryUpdate>) {
+        updates.add(entryUpdates)
+    }
+
+    override fun onError(throwable: Throwable) {
+        errors.add(throwable)
+    }
+
+    fun reset() {
+        updates.clear()
+        errors.clear()
+    }
+}