Skip to content

Commit e92abf0

Browse files
author
Robert Winkler
committed
Initial implementation of the WebSocketProtocolClient
1 parent ae532aa commit e92abf0

File tree

7 files changed

+186
-50
lines changed

7 files changed

+186
-50
lines changed

‎kotlin-wot-binding-websocket/build.gradle.kts‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -8,12 +8,14 @@ dependencies {
88
implementation("io.ktor:ktor-server-netty")
99
implementation("io.ktor:ktor-server-websockets")
1010
implementation("io.ktor:ktor-client-websocket:1.1.4")
11+
implementation("io.ktor:ktor-server-content-negotiation")
1112
implementation("io.ktor:ktor-client-cio")
1213
implementation("io.ktor:ktor-client-auth")
1314
implementation("io.ktor:ktor-client-logging")
1415
implementation("io.ktor:ktor-server-call-logging")
1516
implementation("io.ktor:ktor-serialization-jackson")
1617
testImplementation("io.ktor:ktor-server-test-host")
18+
testImplementation(project(":kotlin-wot-binding-http"))
1719
testImplementation("ch.qos.logback:logback-classic:1.5.12")
1820
testImplementation("com.marcinziolo:kotlin-wiremock:2.1.1")
1921
testImplementation("io.ktor:ktor-server-test-host-jvm:3.0.0")

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

Lines changed: 72 additions & 26 deletions
Original file line numberDiff line numberDiff line change
@@ -1,72 +1,118 @@
11
package ai.ancf.lmos.wot.binding.websocket
22

33

4+
import ai.ancf.lmos.wot.JsonMapper
45
import ai.ancf.lmos.wot.content.Content
6+
import ai.ancf.lmos.wot.content.ContentManager
57
import ai.ancf.lmos.wot.thing.form.Form
68
import ai.anfc.lmos.wot.binding.ProtocolClient
79
import ai.anfc.lmos.wot.binding.ProtocolClientException
10+
import ai.anfc.lmos.wot.binding.Resource
811
import io.ktor.client.*
912
import io.ktor.client.engine.cio.*
1013
import io.ktor.client.plugins.websocket.*
11-
import io.ktor.http.*
1214
import io.ktor.serialization.jackson.*
1315
import io.ktor.websocket.*
1416
import kotlinx.coroutines.CompletableDeferred
17+
import kotlinx.coroutines.sync.Mutex
18+
import kotlinx.coroutines.sync.withLock
1519
import org.slf4j.LoggerFactory
20+
import java.util.concurrent.ConcurrentHashMap
21+
import kotlin.collections.set
1622

1723
class WebSocketProtocolClient(
1824
private val httpClientConfig: HttpClientConfig? = null,
1925
private val client: HttpClient = HttpClient(CIO) {
20-
install(WebSockets){
21-
contentConverter = JacksonWebsocketContentConverter()
26+
install(WebSockets) {
27+
contentConverter = JacksonWebsocketContentConverter(JsonMapper.instance)
2228
}
2329
}
2430
) : ProtocolClient {
2531
companion object {
2632
private val log = LoggerFactory.getLogger(WebSocketProtocolClient::class.java)
2733
}
2834

29-
private var session: DefaultClientWebSocketSession? = null
35+
// Cache for WebSocket sessions, keyed by href
36+
private val sessionCache = ConcurrentHashMap<String, DefaultClientWebSocketSession>()
37+
private val cacheMutex = Mutex()
3038

3139
override suspend fun start() {
3240
log.info("Starting WebSocketProtocolClient")
33-
client.webSocket(
34-
method = HttpMethod.Get,
35-
host = httpClientConfig?.address ?: "localhost",
36-
port = httpClientConfig?.port ?: 80,
37-
path = "/ws"
38-
) {
39-
session = this
40-
}
41+
// No global connection to start. Connections are established per href.
4142
}
4243

4344
override suspend fun stop() {
4445
log.info("Stopping WebSocketProtocolClient")
45-
session?.close()
46-
session = null
46+
// Close all cached sessions
47+
sessionCache.values.forEach { session ->
48+
try {
49+
session.close()
50+
} catch (e: Exception) {
51+
log.warn("Error closing WebSocket session: ${e.message}", e)
52+
}
53+
}
54+
sessionCache.clear()
55+
}
56+
57+
override suspend fun readResource(resource: Resource): Content {
58+
return sendMessage(resource.form, ReadPropertyMessage(resource.thingId, property = resource.name))
4759
}
4860

49-
override suspend fun readResource(form: Form): Content {
50-
return resolveRequestToContent(form)
61+
override suspend fun writeResource(resource: Resource, content: Content) {
62+
sendMessage(resource.form, WritePropertyMessage(resource.thingId, property = resource.name,
63+
data = JsonMapper.instance.readTree(content.body)
64+
))
5165
}
5266

53-
private suspend fun resolveRequestToContent(form: Form, content: Content? = null): Content {
67+
override suspend fun invokeResource(resource: Resource, content: Content?): Content {
68+
return sendMessage(resource.form, InvokeActionMessage(resource.thingId, action = resource.name,
69+
input = JsonMapper.instance.readTree(content?.body)
70+
))
71+
}
72+
73+
private suspend fun sendMessage(form: Form, message: WoTMessage): Content {
74+
val session = getOrCreateSession(form.href)
75+
5476
val response = CompletableDeferred<Content>()
5577

5678
try {
57-
session?.let {
58-
it.sendSerialized(ReadPropertyMessage("test", property = "test"))
59-
60-
val readingMessage = it.receiveDeserialized<PropertyReadingMessage>()
79+
session.sendSerialized(message)
6180

62-
val responseContent = Content(
63-
body = readingMessage.data.binaryValue()
64-
)
65-
response.complete(responseContent)
81+
when (val woTMessage = session.receiveDeserialized<WoTMessage>()) {
82+
is ErrorMessage -> throw ProtocolClientException("Error received: ${woTMessage.title} - ${woTMessage.detail}")
83+
is PropertyReadingMessage -> {
84+
val responseContent = ContentManager.valueToContent(woTMessage.data)
85+
response.complete(responseContent)
86+
}
87+
is ActionStatusMessage -> {
88+
val responseContent = ContentManager.valueToContent(woTMessage.output)
89+
response.complete(responseContent)
90+
}
91+
else -> throw ProtocolClientException("Unexpected message type received: ${woTMessage::class.simpleName}")
6692
}
6793
} catch (e: Exception) {
68-
response.completeExceptionally(ProtocolClientException("Error during http request: ${e.message}", e))
94+
response.completeExceptionally(ProtocolClientException("Error during WebSocket request: ${e.message}", e))
6995
}
96+
7097
return response.await()
7198
}
99+
100+
private suspend fun getOrCreateSession(href: String): DefaultClientWebSocketSession {
101+
cacheMutex.withLock {
102+
// Perform both the check and the update within the same lock
103+
sessionCache[href]?.let { return it }
104+
// If no session exists, create a new one
105+
val newSession = createSession(href)
106+
sessionCache[href] = newSession
107+
return newSession
108+
}
109+
}
110+
111+
private suspend fun createSession(href: String): DefaultClientWebSocketSession {
112+
try {
113+
return client.webSocketSession (href)
114+
} catch (e: Exception) {
115+
throw ProtocolClientException("Failed to create WebSocket session for $href", e)
116+
}
117+
}
72118
}

‎kotlin-wot-binding-websocket/src/main/kotlin/websocket/WebSocketProtocolServer.kt‎

Lines changed: 11 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -12,13 +12,16 @@ import ai.ancf.lmos.wot.thing.schema.ContentListener
1212
import ai.ancf.lmos.wot.thing.schema.WoTExposedThing
1313
import ai.anfc.lmos.wot.binding.ProtocolServer
1414
import ai.anfc.lmos.wot.binding.ProtocolServerException
15+
import com.fasterxml.jackson.databind.DeserializationFeature
16+
import com.fasterxml.jackson.databind.SerializationFeature
1517
import com.fasterxml.jackson.module.kotlin.readValue
1618
import io.ktor.http.*
1719
import io.ktor.serialization.jackson.*
1820
import io.ktor.server.application.*
1921
import io.ktor.server.engine.*
2022
import io.ktor.server.netty.*
2123
import io.ktor.server.plugins.calllogging.*
24+
import io.ktor.server.plugins.contentnegotiation.*
2225
import io.ktor.server.routing.*
2326
import io.ktor.server.websocket.*
2427
import io.ktor.util.reflect.*
@@ -32,11 +35,11 @@ class WebSocketProtocolServer(
3235
private val bindPort: Int = 8080,
3336
private val createServer: (host: String, port: Int, servient: Servient) -> EmbeddedServer<*, *> = ::defaultWebSocketServer
3437
) : ProtocolServer {
35-
internal val things: MutableMap<String, ExposedThing> = mutableMapOf()
38+
private val things: MutableMap<String, ExposedThing> = mutableMapOf()
3639

3740
var started = false
3841
private var server: EmbeddedServer<*, *>? = null
39-
private var actualAddresses: List<String> = listOf("http://$bindHost:$bindPort")
42+
private var actualAddresses: List<String> = listOf("ws://$bindHost:$bindPort")
4043

4144
companion object {
4245
private val log = LoggerFactory.getLogger(WebSocketProtocolServer::class.java)
@@ -151,6 +154,12 @@ fun defaultWebSocketServer(host: String, port: Int, servient: Servient): Embedde
151154

152155
fun Application.setupRoutingWithWebSockets(servient: Servient) {
153156
install(CallLogging)
157+
install(ContentNegotiation) {
158+
jackson {
159+
enable(SerializationFeature.INDENT_OUTPUT)
160+
configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false)
161+
}
162+
}
154163
install(WebSockets) {
155164
contentConverter = JacksonWebsocketContentConverter(JsonMapper.instance)
156165
}

‎kotlin-wot-binding-websocket/src/test/kotlin/websocket/WebSocketProtocolClientTest.kt‎

Lines changed: 57 additions & 17 deletions
Original file line numberDiff line numberDiff line change
@@ -2,16 +2,14 @@ package websocket
22

33
import ai.ancf.lmos.wot.Servient
44
import ai.ancf.lmos.wot.Wot
5+
import ai.ancf.lmos.wot.binding.http.HttpProtocolClientFactory
56
import ai.ancf.lmos.wot.binding.websocket.WebSocketProtocolClientFactory
67
import ai.ancf.lmos.wot.binding.websocket.WebSocketProtocolServer
78
import ai.ancf.lmos.wot.thing.exposedThing
89
import ai.ancf.lmos.wot.thing.schema.*
910
import io.mockk.clearAllMocks
1011
import kotlinx.coroutines.test.runTest
11-
import kotlin.test.AfterTest
12-
import kotlin.test.BeforeTest
13-
import kotlin.test.Test
14-
import kotlin.test.assertEquals
12+
import kotlin.test.*
1513

1614
private const val PROPERTY_NAME = "property1"
1715

@@ -29,16 +27,20 @@ private const val EVENT_NAME = "event1"
2927

3028
class WebSocketProtocolClientTest {
3129

32-
private lateinit var servient: Servient
33-
private lateinit var wot: Wot
34-
private var server = WebSocketProtocolServer()
30+
private lateinit var thing: WoTConsumedThing
31+
32+
private lateinit var servient : Servient
33+
34+
private var property1 : Int = 0
35+
private var property2 : String = ""
3536

3637
@BeforeTest
3738
fun setUp() = runTest {
3839

3940
servient = Servient(
40-
servers = listOf(server),
41-
clientFactories = listOf(WebSocketProtocolClientFactory()))
41+
servers = listOf(WebSocketProtocolServer()),
42+
clientFactories = listOf(HttpProtocolClientFactory(), WebSocketProtocolClientFactory())
43+
)
4244

4345
val exposedThing = exposedThing(servient, id="test") {
4446
intProperty(PROPERTY_NAME) {
@@ -75,45 +77,83 @@ class WebSocketProtocolClientTest {
7577
data = StringSchema()
7678
}
7779
}.setPropertyReadHandler(PROPERTY_NAME) {
78-
10.toInteractionInputValue()
80+
property1.toInteractionInputValue()
7981
}.setPropertyReadHandler(PROPERTY_NAME_2) {
8082
5.toInteractionInputValue()
8183
}.setActionHandler(ACTION_NAME) { input, _->
8284
val inputString = input.value() as DataSchemaValue.StringValue
8385
"${inputString.value} 10".toInteractionInputValue()
8486
}.setPropertyWriteHandler(PROPERTY_NAME) { input, _->
8587
val inputInt = input.value() as DataSchemaValue.IntegerValue
86-
inputInt.value.toInteractionInputValue()
88+
property1 = inputInt.value
89+
property1.toInteractionInputValue()
8790
}.setActionHandler(ACTION_NAME_2) { input, _->
88-
"10".toInteractionInputValue()
91+
"test test".toInteractionInputValue()
8992
}.setActionHandler(ACTION_NAME_3) { input, _->
93+
val inputString = input.value() as DataSchemaValue.StringValue
94+
property2 = inputString.value
9095
InteractionInput.Value(DataSchemaValue.NullValue)
9196
}.setActionHandler(ACTION_NAME_4) { _, _->
9297
InteractionInput.Value(DataSchemaValue.NullValue)
9398
}.setEventSubscribeHandler(EVENT_NAME) { _ ->
9499
}
95100

101+
property1 = 10
102+
96103
servient.addThing(exposedThing)
97104
servient.start()
98105
servient.expose("test")
99106

100-
wot = Wot.create(servient)
107+
val wot = Wot.create(servient)
108+
109+
val thingDescription = wot.requestThingDescription("http://localhost:8080/test")
110+
thing = wot.consume(thingDescription)
101111

102112
}
103113

104114
@AfterTest
105115
fun tearDown() = runTest {
106116
clearAllMocks()
117+
servient.shutdown()
107118
}
108119

109120
@Test
110121
fun `should get property`() = runTest{
111122

112-
val exposedThing = server.things["test"]!!
123+
val readProperty1 = thing.readProperty(PROPERTY_NAME).value()
124+
assertEquals(10, (readProperty1 as DataSchemaValue.IntegerValue).value)
125+
126+
val readProperty2 = thing.readProperty(PROPERTY_NAME_2).value()
127+
assertEquals(5, (readProperty2 as DataSchemaValue.IntegerValue).value)
113128

114-
val thing = wot.consume(exposedThing.getThingDescription())
115-
assertEquals("test", thing.getThingDescription().id)
116-
val readProperty = thing.readProperty(PROPERTY_NAME).value()
129+
}
130+
131+
@Test
132+
fun `should write property`() = runTest{
133+
thing.writeProperty(PROPERTY_NAME, 20.toInteractionInputValue())
134+
135+
assertEquals(20, property1)
136+
}
117137

138+
139+
@Test
140+
fun `should invoke action`() = runTest{
141+
val response = thing.invokeAction(ACTION_NAME, "test".toInteractionInputValue()).value()
142+
143+
assertEquals("test 10", (response as DataSchemaValue.StringValue).value)
144+
}
145+
146+
@Test
147+
fun `should invoke action without input`() = runTest{
148+
val response = thing.invokeAction(ACTION_NAME_2).value()
149+
150+
assertEquals("test test", (response as DataSchemaValue.StringValue).value)
151+
}
152+
153+
@Test
154+
fun `should invoke action without output`() = runTest{
155+
val response = thing.invokeAction(ACTION_NAME_3, "test".toInteractionInputValue()).value()
156+
assertEquals("test", property2)
157+
assertIs<DataSchemaValue.NullValue>(response)
118158
}
119159
}

0 commit comments

Comments
 (0)