Add stopable interface. Implement top-down stop mechanism for DefaultMessageBus. DefaultMessageBus now accepts more than one transport that can be configured per message

Author: theUniC

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

@@ -1,6 +1,7 @@
 package io.github.theunic.kcommand.core
 
 import arrow.core.getOrElse
+import io.github.theunic.kcommand.core.transport.AggregatorTransport
 import kotlinx.coroutines.CompletableDeferred
 import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.Dispatchers
@@ -9,8 +10,8 @@ import kotlinx.coroutines.flow.onEach
 
 class DefaultMessageBus<M : Any, R : Any>(
     middlewares: List<Middleware<M, R>> = listOf(),
-    private val transport: Transport<M, R> = LocalTransport(),
-) : AbstractMessageBus<M, R>(middlewares) {
+    private val transport: AggregatorTransport<M, R> = AggregatorTransport(),
+) : AbstractMessageBus<M, R>(middlewares), Stopable {
     init {
         transport
             .receive()
@@ -22,4 +23,8 @@ class DefaultMessageBus<M : Any, R : Any>(
         transport
             .send(message)
             .getOrElse { CompletableDeferred() }
+
+    override fun stop() {
+        transport.stop()
+    }
 }

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

@@ -0,0 +1,5 @@
+package io.github.theunic.kcommand.core
+
+interface Stopable {
+    fun stop()
+}

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

@@ -6,6 +6,5 @@ import kotlinx.coroutines.flow.Flow
 
 interface Transport<M : Any, R : Any> {
     suspend fun send(message: M): Either<Unit, CompletableDeferred<R>>
-
     fun receive(): Flow<Pair<M, Either<Unit, CompletableDeferred<R>>>>
 }

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

@@ -0,0 +1,43 @@
+package io.github.theunic.kcommand.core.transport
+
+import arrow.core.Either
+import io.github.theunic.kcommand.core.Stopable
+import io.github.theunic.kcommand.core.Transport
+import kotlinx.coroutines.CompletableDeferred
+import kotlinx.coroutines.flow.Flow
+import kotlinx.coroutines.flow.merge
+
+class AggregatorTransport<M : Any, R : Any>(
+    private val transports: Map<String, Transport<M, R>> = mapOf(),
+    private val transportResolver: (M) -> String? = { _ -> null } ,
+) : Transport<M, R>, Stopable {
+
+    private val localTransport: Transport<M, R> = LocalTransport()
+
+    override suspend fun send(message: M): Either<Unit, CompletableDeferred<R>> {
+        val transportName = transportResolver(message)
+        val fallbackTransport = if (transportName == null) {
+            localTransport
+        } else {
+            transports[transportName] ?: localTransport
+        }
+
+        return fallbackTransport.send(message)
+    }
+
+    override fun receive(): Flow<Pair<M, Either<Unit, CompletableDeferred<R>>>> {
+        val flows = buildList {
+            add(localTransport.receive())
+            addAll(transports.values.map { it.receive() })
+        }
+        return flows.merge()
+    }
+
+    override fun stop() {
+        transports.values.forEach { tr ->
+            if (tr is Stopable) {
+                tr.stop()
+            }
+        }
+    }
+}

…/theunic/kcommand/core/LocalTransport.kt …command/core/transport/LocalTransport.ktkcommand-core/src/main/kotlin/io/github/theunic/kcommand/core/LocalTransport.kt renamed to kcommand-core/src/main/kotlin/io/github/theunic/kcommand/core/transport/LocalTransport.kt kcommand-core/src/main/kotlin/io/github/theunic/kcommand/core/LocalTransport.kt renamed to kcommand-core/src/main/kotlin/io/github/theunic/kcommand/core/transport/LocalTransport.kt

@@ -1,6 +1,7 @@
-package io.github.theunic.kcommand.core
+package io.github.theunic.kcommand.core.transport
 
 import arrow.core.Either
+import io.github.theunic.kcommand.core.Transport
 import kotlinx.coroutines.CompletableDeferred
 import kotlinx.coroutines.flow.Flow
 import kotlinx.coroutines.flow.MutableSharedFlow

…theunic/kcommand/core/RemoteTransport.kt …ommand/core/transport/RemoteTransport.ktkcommand-core/src/main/kotlin/io/github/theunic/kcommand/core/RemoteTransport.kt renamed to kcommand-core/src/main/kotlin/io/github/theunic/kcommand/core/transport/RemoteTransport.kt kcommand-core/src/main/kotlin/io/github/theunic/kcommand/core/RemoteTransport.kt renamed to kcommand-core/src/main/kotlin/io/github/theunic/kcommand/core/transport/RemoteTransport.kt

@@ -1,6 +1,9 @@
-package io.github.theunic.kcommand.core
+package io.github.theunic.kcommand.core.transport
 
 import arrow.core.Either
+import io.github.theunic.kcommand.core.Envelope
+import io.github.theunic.kcommand.core.MessageRegistry
+import io.github.theunic.kcommand.core.Transport
 import kotlinx.serialization.json.Json
 import kotlinx.coroutines.CompletableDeferred
 import kotlinx.coroutines.flow.Flow

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

@@ -1,5 +1,6 @@
 package io.github.theunic.kcommand.core
 
+import io.github.theunic.kcommand.core.transport.RemoteTransport
 import io.kotest.core.spec.style.BehaviorSpec
 import io.kotest.matchers.shouldBe
 import kotlinx.coroutines.delay

kcommand-kafka-transport/src/main/kotlin/io/github/theunic/kcommand/transport/kafka/KafkaStreamsTransport.kt

@@ -2,7 +2,8 @@ package io.github.theunic.kcommand.transport.kafka
 
 import io.github.oshai.kotlinlogging.KotlinLogging
 import io.github.theunic.kcommand.core.MessageRegistry
-import io.github.theunic.kcommand.core.RemoteTransport
+import io.github.theunic.kcommand.core.Stopable
+import io.github.theunic.kcommand.core.transport.RemoteTransport
 import kotlinx.coroutines.channels.awaitClose
 import kotlinx.coroutines.flow.Flow
 import kotlinx.coroutines.flow.callbackFlow
@@ -34,7 +35,7 @@ suspend fun KafkaProducer<String, String>.sendAwait(record: ProducerRecord<Strin
 class KafkaStreamsRemoteTransport<M : Any, R : Any, TOPIC: Enum<TOPIC>>(
     private val config: KafkaStreamsTransportConfig<M, TOPIC>,
     registry: MessageRegistry<M>,
-) : RemoteTransport<M, R>(registry) {
+) : RemoteTransport<M, R>(registry), Stopable {
 
     private val producer: KafkaProducer<String, String> by lazy {
         val props = Properties().apply {
@@ -77,7 +78,7 @@ class KafkaStreamsRemoteTransport<M : Any, R : Any, TOPIC: Enum<TOPIC>>(
         }
     }
 
-    fun stop() {
+    override fun stop() {
         streams.close(Duration.ofSeconds(1))
         producer.close()
     }