feat: add flow graph

Author: silenium-dev

build.gradle.kts

@@ -1,5 +1,6 @@
 plugins {
     kotlin("jvm") version "2.0.0"
+    id("org.jetbrains.kotlinx.atomicfu") version "0.25.0"
 }
 
 group = "dev.silenium.playground"
@@ -17,6 +18,8 @@ dependencies {
     implementation("org.jetbrains.kotlinx:kotlinx-coroutines-jdk9:$coroutines")
     implementation("org.jetbrains.kotlinx:kotlinx-coroutines-reactive:$coroutines")
 
+    implementation("org.jetbrains.kotlinx:atomicfu:0.25.0")
+
     val kotest = "5.9.0"
     testImplementation("io.kotest:kotest-runner-junit5-jvm:$kotest")
     testImplementation("io.kotest:kotest-assertions-core-jvm:$kotest")

src/main/kotlin/Source.kt

@@ -0,0 +1,10 @@
+package dev.silenium.libs.flows.api
+
+import kotlinx.coroutines.Job
+import kotlinx.coroutines.flow.launchIn
+import kotlinx.coroutines.flow.onEach
+
+interface FlowGraphBuilder : FlowGraph {
+    infix fun <T, P> Source<T, P>.connectTo(sink: Sink<T, P>): Job =
+        flow.onEach(sink::submit).launchIn(this@FlowGraphBuilder)
+}

src/main/kotlin/dev/silenium/libs/flows/api/Extensions.kt

@@ -0,0 +1,27 @@
+package dev.silenium.libs.flows.api
+
+import kotlinx.coroutines.CoroutineScope
+
+interface FlowGraph : AutoCloseable, CoroutineScope {
+    val elements: List<FlowGraphElement<*>>
+    val size: Int
+
+    fun <T, P, E : Source<T, P>> source(
+        source: E,
+        name: String = "${source.javaClass.name}-${size}",
+    ): SourceFlowGraphElement<T, P, E>
+
+    fun <T, P, E : Sink<T, P>> sink(
+        sink: E,
+        name: String = "${sink.javaClass.name}-${size}",
+    ): SinkFlowGraphElement<T, P, E>
+
+    fun <IT, IP, OT, OP, E : Transformer<IT, IP, OT, OP>> transformer(
+        transform: E,
+        name: String = "${transform.javaClass.name}-${size}",
+    ): TransformerFlowGraphElement<IT, IP, OT, OP, E>
+
+    fun <E : Source<*, *>> source(name: String): FlowGraphElement<E>?
+    fun <E : Sink<*, *>> sink(name: String): FlowGraphElement<E>?
+    fun <E : Transformer<*, *, *, *>> transformer(name: String): FlowGraphElement<E>?
+}

src/main/kotlin/dev/silenium/libs/flows/api/FlowGraph.kt

@@ -0,0 +1,15 @@
+package dev.silenium.libs.flows.api
+
+import kotlin.reflect.KClass
+
+interface FlowGraphElement<T : Any> : AutoCloseable {
+    val name: String
+    val type: KClass<T>
+    val impl: T
+
+    override fun close()
+}
+
+interface SourceFlowGraphElement<T, P, S : Source<T, P>> : FlowGraphElement<S>, Source<T, P>
+interface SinkFlowGraphElement<T, P, S : Sink<T, P>> : FlowGraphElement<S>, Sink<T, P>
+interface TransformerFlowGraphElement<IT, IP, OT, OP, T : Transformer<IT, IP, OT, OP>> : FlowGraphElement<T>, Transformer<IT, IP, OT, OP>

src/main/kotlin/dev/silenium/libs/flows/api/FlowGraphElement.kt

@@ -0,0 +1,17 @@
+package dev.silenium.libs.flows.api
+
+data class FlowItem<T, P>(val pad: UInt, val metadata: P, val value: T) : ReferenceCounted<FlowItem<T, P>> {
+    @Suppress("UNCHECKED_CAST")
+    override fun clone(): Result<FlowItem<T, P>> {
+        return when (value) {
+            is ReferenceCounted<*> -> value.clone().map { FlowItem(pad, metadata, it as T) }
+            else -> Result.success(this)
+        }
+    }
+
+    override fun close() {
+        if (value is AutoCloseable) {
+            (value as AutoCloseable).close()
+        }
+    }
+}

src/main/kotlin/dev/silenium/libs/flows/api/FlowItem.kt

@@ -1,3 +1,5 @@
+package dev.silenium.libs.flows.api
+
 interface ReferenceCounted<T : AutoCloseable> : AutoCloseable {
     fun clone(): Result<T>
 }

src/main/kotlin/ReferenceCounted.kt renamed to src/main/kotlin/dev/silenium/libs/flows/api/ReferenceCounted.kt

@@ -0,0 +1,18 @@
+package dev.silenium.libs.flows.api
+
+import kotlinx.coroutines.flow.FlowCollector
+
+interface Sink<T, P> : FlowCollector<FlowItem<T, P>>, AutoCloseable {
+    val inputMetadata: Map<UInt, P?>
+
+    fun configure(pad: UInt, metadata: P): Result<Unit>
+
+    suspend fun submit(item: FlowItem<T, P>): Result<Unit>
+
+    override suspend fun emit(value: FlowItem<T, P>) {
+        check(inputMetadata.containsKey(value.pad)) { "pad not configured" }
+        check(inputMetadata[value.pad] == value.metadata) { "metadata mismatch" }
+
+        submit(value).getOrThrow()
+    }
+}

src/main/kotlin/dev/silenium/libs/flows/api/Sink.kt

@@ -0,0 +1,8 @@
+package dev.silenium.libs.flows.api
+
+import kotlinx.coroutines.flow.Flow
+
+interface Source<T, P> : AutoCloseable {
+    val outputMetadata: Map<UInt, P>
+    val flow: Flow<FlowItem<T, P>>
+}

src/main/kotlin/dev/silenium/libs/flows/api/Source.kt

@@ -0,0 +1,3 @@
+package dev.silenium.libs.flows.api
+
+interface Transformer<IT, IP, OT, OP> : Sink<IT, IP>, Source<OT, OP>