Introduce an abstract RemoteTransport class

Author: theUniC

build.gradle.kts

@@ -7,6 +7,7 @@ plugins {
     alias(libs.plugins.nmcp)
     alias(libs.plugins.vanniktech)
     alias(libs.plugins.dokka)
+    alias(libs.plugins.serialization)
     `java-library`
 }
 

gradle/libs.versions.toml

@@ -6,15 +6,18 @@ arrow = "2.0.1"
 nmcp = "0.0.9"
 vanniktech = "0.30.0"
 dokka = "2.0.0"
+kotlinxSerializationJson = "1.8.0"
 
 [libraries]
 kotlinx-coroutines = { module = "org.jetbrains.kotlinx:kotlinx-coroutines-core", version.ref = "kotlinCoroutines" }
 kotest-runner-junit5 = { module = "io.kotest:kotest-runner-junit5", version.ref = "kotestVersion" }
 kotest-assertions-core = { module = "io.kotest:kotest-assertions-core", version.ref = "kotestVersion" }
 arrow-core = { module = "io.arrow-kt:arrow-core", version.ref = "arrow" }
+kotlinx-serialization-json = { module = "org.jetbrains.kotlinx:kotlinx-serialization-json", version.ref = "kotlinxSerializationJson" }
 
 [plugins]
 kotlin = { id = "org.jetbrains.kotlin.jvm", version.ref = "kotlin" }
 nmcp = { id = "com.gradleup.nmcp", version.ref = "nmcp" }
 vanniktech = { id = "com.vanniktech.maven.publish", version.ref = "vanniktech" }
 dokka = { id = "org.jetbrains.dokka", version.ref = "dokka" }
+serialization = { id = "org.jetbrains.kotlin.plugin.serialization", version.ref = "kotlin" }

kcommand-core/build.gradle.kts

@@ -6,9 +6,14 @@ description =
     to the Message Bus implementations.
     """.trimIndent().replace("\n", " ")
 
+plugins {
+    alias(libs.plugins.serialization)
+}
+
 dependencies {
     implementation(libs.kotlinx.coroutines)
     implementation(libs.arrow.core)
+    implementation(libs.kotlinx.serialization.json)
     testImplementation(libs.kotest.runner.junit5)
     testImplementation(libs.kotest.assertions.core)
 }

kcommand-core/src/main/kotlin/io/github/theunic/kcommand/core/Envelope.kt

@@ -0,0 +1,9 @@
+package io.github.theunic.kcommand.core
+
+import kotlinx.serialization.Serializable
+
+@Serializable
+data class Envelope(
+    val type: String,
+    val payload: String,
+)

kcommand-core/src/main/kotlin/io/github/theunic/kcommand/core/MessageRegistry.kt

@@ -0,0 +1,23 @@
+package io.github.theunic.kcommand.core
+
+import kotlinx.serialization.KSerializer
+import kotlin.reflect.KClass
+
+class MessageRegistry<M : Any> {
+    private val classToEntry = mutableMapOf<KClass<out M>, Pair<String, KSerializer<out M>>>()
+    private val nameToSerializer = mutableMapOf<String, KSerializer<out M>>()
+
+    fun <T : M> register(kclass: KClass<T>, serializer: KSerializer<T>) {
+        val typeName = kclass.qualifiedName.toString()
+        classToEntry[kclass] = typeName to serializer
+        nameToSerializer[typeName] = serializer
+    }
+
+    fun typeNameAndSerializer(kclass: KClass<out M>): Pair<String, KSerializer<out M>>? {
+        return classToEntry[kclass]
+    }
+
+    fun serializerFor(typeName: String): KSerializer<out M>? {
+        return nameToSerializer[typeName]
+    }
+}

kcommand-core/src/main/kotlin/io/github/theunic/kcommand/core/RemoteTransport.kt

@@ -0,0 +1,61 @@
+package io.github.theunic.kcommand.core
+
+import arrow.core.Either
+import kotlinx.serialization.json.Json
+import kotlinx.coroutines.CompletableDeferred
+import kotlinx.coroutines.flow.Flow
+import kotlinx.serialization.KSerializer
+import kotlinx.coroutines.flow.map
+import kotlinx.coroutines.flow.filterNotNull
+
+abstract class RemoteTransport<M: Any, R: Any>(
+    private val registry: MessageRegistry<M>,
+    private val json: Json = Json {
+        ignoreUnknownKeys = false
+        encodeDefaults = true
+        prettyPrint = false
+    }
+): Transport<M, R> {
+    protected abstract fun doSend(serializedMessage: String)
+    protected abstract fun doReceiveFlow(): Flow<String>
+
+    override suspend fun send(message: M): Either<Unit, CompletableDeferred<R>> {
+        val kclass = message::class
+        val (typeName, serializer) = registry.typeNameAndSerializer(kclass)
+            ?: return Either.Left(Unit)
+
+        @Suppress("UNCHECKED_CAST")
+        val payloadJson = json.encodeToString(serializer as KSerializer<M>, message)
+        val envelope = Envelope(typeName, payloadJson)
+        val envelopeJson = json.encodeToString(Envelope.serializer(), envelope)
+
+        doSend(envelopeJson)
+
+        // Fire-and-forget
+        return Either.Left(Unit)
+    }
+
+    override fun receive(): Flow<Pair<M, Either<Unit, CompletableDeferred<R>>>> {
+        return doReceiveFlow()
+            .map { raw ->
+                try {
+                    val envelope = json.decodeFromString(Envelope.serializer(), raw)
+                    val serializer = registry.serializerFor(envelope.type)
+                    if (serializer != null) {
+                        val obj = json.decodeFromString(serializer, envelope.payload)
+                        val casted = obj as? M
+                        if (casted != null) {
+                            casted to Either.Left(Unit)
+                        } else {
+                            null
+                        }
+                    } else {
+                        null
+                    }
+                } catch (e: Exception) {
+                    null
+                }
+            }
+            .filterNotNull()
+    }
+}

kcommand-core/src/test/kotlin/io/github/theunic/kcommand/core/RemoteTransportTest.kt

@@ -0,0 +1,116 @@
+package io.github.theunic.kcommand.core
+
+import io.kotest.core.spec.style.BehaviorSpec
+import io.kotest.matchers.shouldBe
+import kotlinx.coroutines.delay
+import kotlinx.coroutines.flow.first
+import kotlinx.coroutines.launch
+import kotlinx.coroutines.test.runTest
+import kotlinx.serialization.Serializable
+import kotlinx.coroutines.flow.Flow
+import kotlinx.coroutines.flow.MutableSharedFlow
+import kotlinx.serialization.json.Json
+
+class DummyRemoteTransport<M : Any, R : Any>(
+    registry: MessageRegistry<M>,
+    json: Json = Json { ignoreUnknownKeys = true }
+) : RemoteTransport<M, R>(registry, json) {
+
+    val sentMessages = mutableListOf<String>()
+    private val incomingFlow = MutableSharedFlow<String>()
+
+    override fun doSend(serializedMessage: String) {
+        sentMessages.add(serializedMessage)
+    }
+
+    override fun doReceiveFlow(): Flow<String> {
+        return incomingFlow
+    }
+
+    suspend fun emitIncomingMessage(raw: String) {
+        incomingFlow.emit(raw)
+    }
+}
+
+@Serializable
+sealed class BaseCommand
+
+@Serializable
+data class MyCommand(val id: Int, val payload: String) : BaseCommand()
+
+@Serializable
+data class OtherCommand(val name: String) : BaseCommand()
+
+class RemoteTransportTest : BehaviorSpec({
+     val registry = MessageRegistry<BaseCommand>().apply {
+         register(MyCommand::class, MyCommand.serializer())
+         register(OtherCommand::class, OtherCommand.serializer())
+     }
+
+     val transport = DummyRemoteTransport<BaseCommand, Any>(registry)
+
+     given("A RemoteTransport with a dummy implementation") {
+
+         `when`("we send a MyCommand object") {
+             then("it should serialize the object into an Envelope and store it") {
+                 runTest {
+                     val cmd = MyCommand(42, "HelloTransport")
+                     val result = transport.send(cmd)
+
+                     result.isLeft() shouldBe true
+
+                     transport.sentMessages.size shouldBe 1
+                     val raw = transport.sentMessages.first()
+
+                     raw.contains(MyCommand::class.qualifiedName.toString()) shouldBe true
+                     raw.contains("HelloTransport") shouldBe true
+                 }
+             }
+         }
+
+         `when`("we receive a raw envelope for OtherCommand") {
+             then("it should reconstruct the object and emit it in the flow") {
+                 runTest {
+                     val job = launch {
+                         transport.receive().collect {
+                             println("DEBUG: flow emitted => $it")
+                         }
+                     }
+
+                     val envelopeJson = """
+                        {
+                            "type": "${OtherCommand::class.qualifiedName}",
+                            "payload": "{\"name\":\"myName\"}"
+                        }
+                     """.trimIndent()
+
+                     transport.emitIncomingMessage(envelopeJson)
+
+                     delay(500)
+                     job.cancel()
+                 }
+             }
+         }
+
+         `when`("we receive an envelope with a type not registered") {
+             then("it should emit nothing (and we don't see a Pair)") {
+                 runTest {
+                     val job = launch {
+                         transport.receive().first()
+                     }
+
+                     val invalidEnvelopeJson = """
+                        {
+                            "type": "com.example.UnknownCommand",
+                            "payload": "{}"
+                        }
+                     """.trimIndent()
+
+                     transport.emitIncomingMessage(invalidEnvelopeJson)
+
+                     job.cancel()
+                 }
+             }
+         }
+     }
+ })

settings.gradle.kts

@@ -4,3 +4,4 @@ plugins {
 
 rootProject.name = "kcommand"
 include(":kcommand-core")
+include("kcommand-kafka-transport")