feat: add JobTransformerBase that uses a coroutine for processing

Author: silenium-dev

src/main/kotlin/dev/silenium/libs/flows/base/JobTransformerBase.kt

@@ -0,0 +1,46 @@
+package dev.silenium.libs.flows.base
+
+import dev.silenium.libs.flows.api.Transformer
+import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.Job
+import kotlinx.coroutines.launch
+import kotlinx.coroutines.runBlocking
+import java.util.*
+
+abstract class JobTransformerBase<IT, IP, OT, OP>(
+    protected val coroutineScope: CoroutineScope,
+    protected val pads: Set<UInt>? = setOf(0u)
+) : Transformer<IT, IP, OT, OP>, SourceBase<OT, OP>() {
+
+    @OptIn(ExperimentalUnsignedTypes::class)
+    constructor(coroutineScope: CoroutineScope, vararg inputPads: UInt) : this(coroutineScope, inputPads.toSet())
+
+    protected val inputMetadata_ = mutableMapOf<UInt, IP>()
+    override val inputMetadata: Map<UInt, IP?> = Collections.unmodifiableMap(inputMetadata_)
+    protected var job: Job? = null
+
+    override fun configure(pad: UInt, metadata: IP): Result<Unit> {
+        inputMetadata_[pad] = metadata
+        outputMetadata_[pad] = outputMetadata(metadata)
+        if (pads?.equals(inputMetadata_.keys) == true) {
+            job = coroutineScope.launch { run() }
+        }
+        return Result.success(Unit)
+    }
+
+    override fun close() {
+        runBlocking { job?.join() }
+        super.close()
+    }
+
+    abstract fun outputMetadata(inputMetadata: IP): OP
+
+    /**
+     * main processing function
+     *
+     * *Note: cancellation must be handled by the implementation
+     * ([close] only joins the job, it does not cancel it)*
+     * @see CoroutineScope
+     */
+    protected abstract suspend fun CoroutineScope.run()
+}

src/test/kotlin/dev/silenium/libs/flows/base/SourceBaseTest.kt

@@ -13,6 +13,7 @@ import kotlinx.coroutines.flow.toList
 import java.nio.ByteBuffer
 import kotlin.io.encoding.Base64
 import kotlin.io.encoding.ExperimentalEncodingApi
+import kotlin.time.Duration.Companion.milliseconds
 
 
 @OptIn(ExperimentalEncodingApi::class)
@@ -36,6 +37,7 @@ class SourceBaseTest : FunSpec({
             decoder.flow.toList()
         }
         started.await()
+        delay(100.milliseconds)
         val bufs = inputs.map { input ->
             val base64 = Base64.encode(input.encodeToByteArray())
             val byteBuffer = ByteBuffer.allocateDirect(base64.length)
@@ -46,8 +48,9 @@ class SourceBaseTest : FunSpec({
             buf.close()
             buf
         }
-        job.cancelAndJoin()
         decoder.close()
+        bufferSource.close()
+        job.join()
         listAsync.await().shouldHaveSize(inputs.size).forEach { item ->
             println(item)
             item.pad shouldBe 0u

src/test/kotlin/dev/silenium/libs/flows/test/Base64Decoder.kt

@@ -1,24 +1,25 @@
 package dev.silenium.libs.flows.test
 
 import dev.silenium.libs.flows.api.FlowItem
-import dev.silenium.libs.flows.api.Transformer
-import dev.silenium.libs.flows.base.SourceBase
-import kotlinx.coroutines.*
+import dev.silenium.libs.flows.base.JobTransformerBase
+import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.Dispatchers
 import kotlinx.coroutines.channels.Channel
-import java.util.*
+import kotlinx.coroutines.delay
 import kotlin.io.encoding.Base64
 import kotlin.io.encoding.ExperimentalEncodingApi
 import kotlin.random.Random
 import kotlin.random.nextInt
 import kotlin.time.Duration.Companion.milliseconds
 
-class Base64Decoder : SourceBase<ByteArray, DataType>(), Transformer<Base64Buffer, DataType, ByteArray, DataType> {
-    override val inputMetadata: MutableMap<UInt, DataType> =
-        Collections.synchronizedMap(mutableMapOf<UInt, DataType>())
+@OptIn(ExperimentalUnsignedTypes::class)
+class Base64Decoder :
+    JobTransformerBase<Base64Buffer, DataType, ByteArray, DataType>(CoroutineScope(Dispatchers.IO), 0u) {
     private val queue = Channel<FlowItem<Base64Buffer, DataType>>(capacity = 4)
 
     @OptIn(ExperimentalEncodingApi::class)
-    private val processor = CoroutineScope(Dispatchers.Default).async {
+    override suspend fun CoroutineScope.run() {
+        println("Base64Decoder.run")
         for (item in queue) {
             delay(Random.nextInt(1..250).milliseconds)
             val array = ByteArray(item.value.buffer.remaining())
@@ -28,11 +29,9 @@ class Base64Decoder : SourceBase<ByteArray, DataType>(), Transformer<Base64Buffe
         }
     }
 
-    override fun configure(pad: UInt, metadata: DataType): Result<Unit> {
-        check(metadata == DataType.BASE64) { "metadata must be BASE64" }
-        this.inputMetadata[pad] = metadata
-        this.outputMetadata_[pad] = DataType.PLAIN
-        return Result.success(Unit)
+    override fun outputMetadata(inputMetadata: DataType): DataType {
+        check(inputMetadata == DataType.BASE64) { "metadata must be BASE64" }
+        return DataType.PLAIN
     }
 
     override suspend fun receive(item: FlowItem<Base64Buffer, DataType>): Result<Unit> = runCatching {
@@ -41,7 +40,6 @@ class Base64Decoder : SourceBase<ByteArray, DataType>(), Transformer<Base64Buffe
 
     override fun close() {
         queue.close()
-        runBlocking { processor.join() }
         super.close()
     }
 }