Fixes

Author: Robert Winkler

kotlin-wot-binding-websocket/src/main/kotlin/websocket/WebSocketProtocolClient.kt

@@ -91,6 +91,7 @@ class WebSocketProtocolClient(
             requestAndReply(resource.form, message)
         } finally {
             resourceChannels[resource.name]?.close()
+            resourceChannels.remove(resource.name)
         }
     }
 
@@ -105,7 +106,7 @@ class WebSocketProtocolClient(
         resourceChannels[resource.name] = channel
 
         return channel.consumeAsFlow().onCompletion {
-            resourceChannels.remove(resource.name)
+            unlinkResource(resource, resourceType)
         }
     }
 

kotlin-wot-integration-tests/build.gradle.kts

@@ -8,9 +8,9 @@ plugins {
     id("io.spring.dependency-management") version "1.1.7"
 }
 
-tasks.named<Test>("test") {
-    enabled = false
-}
+//tasks.named<Test>("test") {
+//    enabled = false
+//}
 
 dependencies {
     // Replace the following with the starter dependencies of specific modules you wish to use
@@ -47,6 +47,7 @@ dependencies {
     testImplementation("org.springframework.boot:spring-boot-starter-test")
     testImplementation("com.hivemq:hivemq-mqtt-client:1.3.3")
     implementation("org.testcontainers:testcontainers:1.20.3")
+    testImplementation("app.cash.turbine:turbine:1.2.0")
 
 }
 

kotlin-wot-integration-tests/src/main/kotlin/integration/ThingToFunctionsMapper.kt

@@ -37,14 +37,13 @@ object ThingToFunctionsMapper {
 
     suspend fun requestThingDescription(wot: Wot, functions: Functions, group: String, url: String, securityScheme: SecurityScheme = NoSecurityScheme()): List<LLMFunction> {
         val thingDescription = wot.requestThingDescription(url, securityScheme)
-        val consumedThings = consumeThings(wot, setOf(thingDescription))
-        val retrieveAllFunction = createRetrieveAllFunction(functions, group, setOf(thingDescription))
-        mapAllThingFunctions(functions, group, consumedThings)
-        val allFunctions = retrieveAllFunction+ functionCache.values.flatten()
+        val consumedThing = wot.consume(thingDescription)
+        mapThingDescriptionToFunctions2(functions, group, thingDescription)
+        val allFunctions = functionCache.values.flatten()
         return allFunctions
     }
 
-     fun createRetrieveAllFunction(functions: Functions, group: String, thingDescriptions: Set<WoTThingDescription>): List<LLMFunction> {
+    fun createRetrieveAllFunction(functions: Functions, group: String, thingDescriptions: Set<WoTThingDescription>): List<LLMFunction> {
         return functions("retrieveAllThings", "Returns the metadata information of all available devices.", group) {
             summarizeThingDescriptions(thingDescriptions)
         }
@@ -105,13 +104,13 @@ object ThingToFunctionsMapper {
             val params = defaultParams + actionParams
             functionCache.getOrPut(actionName) {
                 functions(actionName, action.description ?: "No Description available", group, params) { (thingId, input) ->
-                    invokeAction(thingDescription.id, actionName, input, action.input)
+                    invokeAction(thingId, actionName, input, action.input)
                 }
             }
         }
     }
 
-    private suspend fun invokeAction(thingId: String, actionName: String, input: String?, inputSchema: DataSchema<*>?): String {
+    private suspend fun invokeAction(thingId: String?, actionName: String, input: String?, inputSchema: DataSchema<*>?): String {
         return try {
             val jsonInput : JsonNode = inputSchema?.let { mapSchemaToJsonNode(it, input!!) } ?: TextNode(input)
             thingDescriptionsMap[thingId]?.invokeAction(actionName, jsonInput)?.asText()

kotlin-wot-integration-tests/src/main/resources/application.yaml

@@ -27,7 +27,7 @@ arc:
     clients:
       - id: GPT-4o
         model-name: GPT35T-1106
-        api-key: 0c798e9b59c64f07ab35d0b1aee65d19
+        api-key: dummy
         client: langchain4j-azure
         url: https://gpt4-uk.openai.azure.com
     tools:

kotlin-wot-integration-tests/src/test/kotlin/integration/QuickTest.kt

@@ -4,8 +4,12 @@ package ai.ancf.lmos.wot.integration
 import ai.ancf.lmos.sdk.agents.WotConversationalAgent
 import ai.ancf.lmos.sdk.agents.lastMessage
 import ai.ancf.lmos.sdk.agents.toAgentRequest
+import ai.ancf.lmos.sdk.model.AgentEvent
+import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.flow.launchIn
+import kotlinx.coroutines.flow.onEach
 import kotlinx.coroutines.runBlocking
-import org.eclipse.lmos.arc.agents.AgentEvent
 import java.util.concurrent.CountDownLatch
 import kotlin.test.Test
 
@@ -41,11 +45,13 @@ class QuickTest {
             println("Event: $it")
             latch.countDown()
         }
-         */
-        agent.consumeEvent("agentEvent", AgentEvent::class).collect {
+        */
+
+        agent.consumeEvent("agentEvent", AgentEvent::class).onEach {
             println("Event: $it")
             latch.countDown()
-        }
+        }.launchIn(CoroutineScope(Dispatchers.IO))
+
 
         //val command = "What is the state of my lamp?"
         val command = "Scrape the page https://eclipse.dev/lmos/\""

kotlin-wot/src/main/kotlin/thing/ConsumedThing.kt

@@ -29,7 +29,7 @@ import org.slf4j.LoggerFactory
 import java.io.File
 import java.io.IOException
 import java.net.MalformedURLException
-import java.net.URL
+import java.net.URI
 import java.util.concurrent.ConcurrentHashMap
 import java.util.concurrent.ConcurrentMap
 
@@ -337,16 +337,27 @@ data class ConsumedThing(
 
         if (subscribedEvents.containsKey(eventName)) {
             return flow {
-                throw IllegalStateException("ConsumedThing '$title' already has a function subscribed to $eventName. You can only subscribe once.")
+                throw ConsumedThingException("ConsumedThing '$title' already has a function subscribed to $eventName. You can only subscribe once.")
             }
         }
 
         log.debug("ConsumedThing '$title' subscribing to ${form.href}")
 
         val formWithoutURITemplates = handleUriVariables(this, eventAffordance, form, options)
 
-        return client.subscribeResource(Resource(id, eventName, formWithoutURITemplates), ResourceType.EVENT)
+        val subscription = InternalEventSubscription(this, eventName, client, form)
+        subscribedEvents[eventName] = subscription
+
+         val flow = client.subscribeResource(Resource(id, eventName, formWithoutURITemplates), ResourceType.EVENT)
             .map { handleInteractionOutput(it, form, eventAffordance.data) }
+            .onCompletion {
+                error -> if (error != null) {
+                    log.warn("Error while processing observe property for ${eventAffordance.title}: ${error.message}", error)
+                }
+                subscription.stop(options)
+            }
+
+        return flow
     }
 
     @WithSpan(kind = SpanKind.CLIENT)
@@ -363,7 +374,7 @@ data class ConsumedThing(
         val (client, form) = getClientFor(eventAffordance.forms, Operation.SUBSCRIBE_EVENT, options)
 
         if (subscribedEvents.containsKey(eventName)) {
-            throw IllegalStateException("ConsumedThing '$title' already has a function subscribed to $eventName. You can only subscribe once.")
+            throw ConsumedThingException("ConsumedThing '$title' already has a function subscribed to $eventName. You can only subscribe once.")
         }
 
         log.debug("ConsumedThing '$title' subscribing to ${form.href}")
@@ -760,8 +771,8 @@ fun findFormIndexWithScoring(
 
         // Compare the origins of the URLs
         try {
-            val formUrl = URL(form.href)
-            val refFormUrl = URL(refForm.href)
+            val formUrl = URI(form.href).toURL()
+            val refFormUrl = URI(refForm.href).toURL()
             if (formUrl.protocol == refFormUrl.protocol && formUrl.host == refFormUrl.host) {
                 score += 1
             }

kotlin-wot/src/test/kotlin/thing/ConsumedThingTest.kt

@@ -5,6 +5,7 @@ import ai.ancf.lmos.wot.Servient
 import ai.ancf.lmos.wot.content.Content
 import ai.ancf.lmos.wot.content.ContentManager
 import ai.ancf.lmos.wot.content.JsonCodec
+import ai.ancf.lmos.wot.content.toJsonContent
 import ai.ancf.lmos.wot.security.BasicSecurityScheme
 import ai.ancf.lmos.wot.thing.action.ThingAction
 import ai.ancf.lmos.wot.thing.event.ThingEvent
@@ -15,14 +16,14 @@ import ai.anfc.lmos.wot.binding.ProtocolClient
 import ai.anfc.lmos.wot.binding.ProtocolClientFactory
 import ai.anfc.lmos.wot.binding.Resource
 import ai.anfc.lmos.wot.binding.ResourceType
+import app.cash.turbine.test
 import com.fasterxml.jackson.databind.node.TextNode
+import com.fasterxml.jackson.module.kotlin.treeToValue
 import io.mockk.*
 import kotlinx.coroutines.flow.emptyFlow
+import kotlinx.coroutines.flow.flow
 import kotlinx.coroutines.runBlocking
-import kotlin.test.BeforeTest
-import kotlin.test.Test
-import kotlin.test.assertEquals
-import kotlin.test.assertNotNull
+import kotlin.test.*
 
 class ConsumedThingTest {
 
@@ -158,6 +159,59 @@ class ConsumedThingTest {
         assertEquals(true, subscription.active)
     }
 
+    @Test
+    fun `subscribe twice to the same event should fail`(): Unit = runBlocking {
+        val listener = mockk<InteractionListener>(relaxed = true)
+        coEvery { protocolClient.subscribeResource(any<Resource>(), any<ResourceType>()) } returns emptyFlow()
+
+        consumedThing.subscribeEvent("testEvent", listener)
+        assertFailsWith<ConsumedThingException> { consumedThing.subscribeEvent("testEvent", listener) }
+    }
+
+    @Test
+    fun `test consumeEvent`(): Unit = runBlocking {
+        coEvery { protocolClient.subscribeResource(any<Resource>(), any<ResourceType>()) } returns flow {
+            emit("testValue".toJsonContent())
+        }
+
+        consumedThing.consumeEvent("testEvent").test {
+            val item = awaitItem().value()
+            val value : String = JsonMapper.instance.treeToValue(item)
+            assertEquals("testValue", value)
+            awaitComplete()
+        }
+
+        coVerify { protocolClient.unlinkResource(any<Resource>(), any<ResourceType>()) }
+    }
+
+    @Test
+    fun `test consumeEvent with error`(): Unit = runBlocking {
+        coEvery { protocolClient.subscribeResource(any<Resource>(), any<ResourceType>()) } returns flow {
+            throw Exception("test error")
+        }
+
+        consumedThing.consumeEvent("testEvent").test {
+            val error = awaitError() // Waits for the expected exception
+            assertTrue(error is Exception)
+        }
+
+        coVerify { protocolClient.unlinkResource(any<Resource>(), any<ResourceType>()) }
+    }
+
+    @Test
+    fun `consume twice to the same event should fail`(): Unit = runBlocking {
+        coEvery { protocolClient.subscribeResource(any<Resource>(), any<ResourceType>()) } returns emptyFlow()
+
+        consumedThing.consumeEvent("testEvent")
+        consumedThing.consumeEvent("testEvent").test {
+            val error = awaitError() // Waits for the expected exception
+            assertTrue(error is ConsumedThingException)
+        }
+
+        coVerify(exactly = 0) { protocolClient.unlinkResource(any<Resource>(), any<ResourceType>()) }
+    }
+
+
     @Test
     fun testEquals() {
         val thingDescription = ThingDescription(

lmos-kotlin-sdk-base/src/main/kotlin/sdk/agents/ConversationalAgent.kt

@@ -30,6 +30,17 @@ fun String.toAgentRequest(): AgentRequest {
     )
 }
 
+fun String.toAgentResult(): AgentResult {
+    return AgentResult(
+        messages = listOf(
+            Message(
+                role = "user",
+                content = this
+            )
+        )
+    )
+}
+
 fun AgentResult.lastMessage(): String {
     return this.messages.last().content
 }

lmos-kotlin-sdk-client/src/main/kotlin/sdk/WotConversationalAgent.kt

@@ -63,6 +63,7 @@ class WotConversationalAgent private constructor(private val thing : ConsumedThi
     override suspend fun <T : Any> consumeEvent(eventName: String, clazz: KClass<T>): Flow<T> {
         return thing.consumeEvent(eventName).map { event ->
             try {
+                log.info("Event:" + event.value())
                 JsonMapper.instance.treeToValue(event.value(), clazz.java)
             } catch (e: Exception) {
                 log.error("Failed to parse event", e)