1. Interface structure is refactored

2. subChain handler is added

README.md

@@ -16,7 +16,7 @@ dependencies {
 ```
 #### **`gradle.properties`**
 ```properties
-kotlinCorVersion=0.4.0+
+kotlinCorVersion=0.5.0+
 ```
 
 ## Usage

build.gradle.kts

@@ -5,7 +5,7 @@ plugins {
 }
 
 group = "com.crowdproj.kotlin.cor"
-version = "0.4.0"
+version = "0.5.0"
 
 repositories {
     mavenCentral()

gradle.properties

@@ -1,4 +1,4 @@
-kotlinVersion=1.7.0
+kotlinVersion=1.7.10
 
 coroutinesVersion=1.6.3
 #  -native-mt

src/commonMain/kotlin/base/BaseCorChainDsl.kt

@@ -2,17 +2,17 @@ package com.crowdproj.kotlin.cor.base
 
 import com.crowdproj.kotlin.cor.*
 
-abstract class BaseCorChainDsl<T>(
+abstract class BaseCorChainDsl<T,K>(
     override var title: String = "",
     override var description: String = "",
-    protected val workers: MutableList<ICorExecDsl<T>> = mutableListOf(),
+    protected val workers: MutableList<ICorExecDsl<K>> = mutableListOf(),
     protected var blockOn: suspend T.() -> Boolean = { true },
     protected var blockExcept: suspend T.(e: Throwable) -> Unit = { e: Throwable -> throw e },
-) : ICorChainDsl<T> {
+) : ICorExecDsl<T>, ICorOnDsl<T>, ICorExceptDsl<T>, ICorAddExecDsl<K> {
 
     abstract override fun build(): ICorExec<T>
 
-    override fun add(worker: ICorExecDsl<T>) {
+    override fun add(worker: ICorExecDsl<K>) {
         workers.add(worker)
     }
 

src/commonMain/kotlin/base/BaseCorWorkerDsl.kt

@@ -8,7 +8,7 @@ abstract class BaseCorWorkerDsl<T>(
     protected var blockOn: suspend T.() -> Boolean = { true },
     protected var blockHandle: suspend T.() -> Unit = {},
     protected var blockExcept: suspend T.(e: Throwable) -> Unit = { e: Throwable -> throw e },
-) : ICorWorkerDsl<T> {
+) : ICorExecDsl<T>, ICorOnDsl<T>, ICorExceptDsl<T>, ICorHandleDsl<T> {
 
     abstract override fun build(): ICorExec<T>
 

src/commonMain/kotlin/cor.kt

@@ -9,16 +9,19 @@ interface ICorExecDsl<T> {
     fun build(): ICorExec<T>
 }
 
-interface ICorHandlerDsl<T> {
+interface ICorOnDsl<T> {
     fun on(function: suspend T.() -> Boolean)
+}
+
+interface ICorExceptDsl<T> {
     fun except(function: suspend T.(e: Throwable) -> Unit)
 }
 
-interface ICorChainDsl<T> : ICorExecDsl<T>, ICorHandlerDsl<T> {
+interface ICorAddExecDsl<T> {
     fun add(worker: ICorExecDsl<T>)
 }
 
-interface ICorWorkerDsl<T> : ICorExecDsl<T>, ICorHandlerDsl<T> {
+interface ICorHandleDsl<T> {
     fun handle(function: suspend T.() -> Unit)
 }
 

src/commonMain/kotlin/handlers/Chain.kt

@@ -1,13 +1,13 @@
 package com.crowdproj.kotlin.cor.handlers
 
 import com.crowdproj.kotlin.cor.CorDslMarker
-import com.crowdproj.kotlin.cor.ICorChainDsl
+import com.crowdproj.kotlin.cor.ICorAddExecDsl
 import com.crowdproj.kotlin.cor.ICorExec
 import com.crowdproj.kotlin.cor.base.BaseCorChain
 import com.crowdproj.kotlin.cor.base.BaseCorChainDsl
 
 @CorDslMarker
-fun <T> ICorChainDsl<T>.chain(function: CorChainDsl<T>.() -> Unit) {
+fun <T> ICorAddExecDsl<T>.chain(function: CorChainDsl<T>.() -> Unit) {
     add(CorChainDsl<T>().apply(function))
 }
 
@@ -34,7 +34,7 @@ class CorChain<T>(
  * The chains are executed sequentially.
  */
 @CorDslMarker
-class CorChainDsl<T>() : BaseCorChainDsl<T>() {
+class CorChainDsl<T>() : BaseCorChainDsl<T,T>() {
     override fun build(): ICorExec<T> = CorChain(
         title = title,
         description = description,

src/commonMain/kotlin/handlers/Loop.kt

@@ -1,13 +1,13 @@
 package com.crowdproj.kotlin.cor.handlers
 
 import com.crowdproj.kotlin.cor.CorDslMarker
-import com.crowdproj.kotlin.cor.ICorChainDsl
+import com.crowdproj.kotlin.cor.ICorAddExecDsl
 import com.crowdproj.kotlin.cor.ICorExec
 import com.crowdproj.kotlin.cor.base.BaseCorChain
 import com.crowdproj.kotlin.cor.base.BaseCorChainDsl
 
 @CorDslMarker
-fun <T> ICorChainDsl<T>.loopWhile(
+fun <T> ICorAddExecDsl<T>.loopWhile(
     function: CorLoopDsl<T>.() -> Unit
 ) {
     add(
@@ -16,7 +16,7 @@ fun <T> ICorChainDsl<T>.loopWhile(
 }
 
 @CorDslMarker
-fun <T> ICorChainDsl<T>.loopUntil(
+fun <T> ICorAddExecDsl<T>.loopUntil(
     function: CorLoopDsl<T>.() -> Unit
 ) {
     add(
@@ -72,7 +72,7 @@ class CorLoop<T>(
 class CorLoopDsl<T>(
     private val checkBefore: Boolean,
     var blockCheck: suspend T.() -> Boolean = { true },
-) : BaseCorChainDsl<T>() {
+) : BaseCorChainDsl<T,T>() {
     override fun build(): ICorExec<T> = CorLoop(
         checkBefore,
         title = title,

src/commonMain/kotlin/handlers/Parallel.kt

@@ -1,15 +1,15 @@
 package com.crowdproj.kotlin.cor.handlers
 
 import com.crowdproj.kotlin.cor.CorDslMarker
-import com.crowdproj.kotlin.cor.ICorChainDsl
+import com.crowdproj.kotlin.cor.ICorAddExecDsl
 import com.crowdproj.kotlin.cor.ICorExec
 import com.crowdproj.kotlin.cor.base.BaseCorChain
 import com.crowdproj.kotlin.cor.base.BaseCorChainDsl
 import kotlinx.coroutines.coroutineScope
 import kotlinx.coroutines.launch
 
 @CorDslMarker
-fun <T> ICorChainDsl<T>.parallel(function: CorParallelDsl<T>.() -> Unit) {
+fun <T> ICorAddExecDsl<T>.parallel(function: CorParallelDsl<T>.() -> Unit) {
     add(CorParallelDsl<T>().apply(function))
 }
 
@@ -40,7 +40,7 @@ class CorParallel<T>(
  * Chains are started simultaneously and executed in parallel.
  */
 @CorDslMarker
-class CorParallelDsl<T>(): BaseCorChainDsl<T>() {
+class CorParallelDsl<T>(): BaseCorChainDsl<T,T>() {
     override fun build(): ICorExec<T> = CorParallel(
         title = title,
         description = description,

src/commonMain/kotlin/handlers/SubChain.kt

@@ -0,0 +1,75 @@
+package com.crowdproj.kotlin.cor.handlers
+
+import com.crowdproj.kotlin.cor.CorDslMarker
+import com.crowdproj.kotlin.cor.ICorAddExecDsl
+import com.crowdproj.kotlin.cor.ICorExec
+import com.crowdproj.kotlin.cor.base.BaseCorChain
+import com.crowdproj.kotlin.cor.base.BaseCorChainDsl
+import kotlinx.coroutines.coroutineScope
+import kotlinx.coroutines.flow.*
+import kotlinx.coroutines.launch
+
+@CorDslMarker
+fun <T, K> ICorAddExecDsl<T>.subChain(function: CorSubChainDsl<T,K>.() -> Unit) {
+    add(CorSubChainDsl<T,K>().apply(function))
+}
+
+class CorSubChain<T,K>(
+    private val execs: List<ICorExec<K>>,
+    title: String,
+    description: String = "",
+    blockOn: suspend T.() -> Boolean = { true },
+    private val blockSplit: suspend T.() -> Flow<K>,
+    private val blockJoin: suspend T.(K) -> Unit,
+    blockExcept: suspend T.(Throwable) -> Unit = {},
+    private val buffer: Int = 1,
+) : BaseCorChain<T>(
+    title = title,
+    description = description,
+    blockOn = blockOn,
+    blockExcept = blockExcept
+) {
+
+    override suspend fun handle(context: T): Unit = coroutineScope {
+        context
+            .blockSplit()
+            .onEach { subCtx -> execs.map{ launch { it.exec(subCtx) } }.forEach { it.join() } }
+            .buffer(buffer)
+            .collect { context.blockJoin(it) }
+    }
+}
+
+/**
+ * DLS is the execution context of multiple chains.
+ * It can be expanded by other chains.
+ */
+@CorDslMarker
+class CorSubChainDsl<T,K>(
+): BaseCorChainDsl<T,K>() {
+    private var blockSplit: suspend T.() -> Flow<K> = { emptyFlow() }
+    private var blockJoin: suspend T.(K) -> Unit = {}
+    private var bufferSize: Int = 0
+
+    fun buffer(size: Int) {
+        bufferSize = size
+    }
+
+    fun split(funSplit: suspend T.() -> Flow<K>) {
+        blockSplit = funSplit
+    }
+
+    fun join(funJoin: suspend T.(K) -> Unit) {
+        blockJoin = funJoin
+    }
+
+    override fun build(): ICorExec<T> = CorSubChain(
+        title = title,
+        description = description,
+        execs = workers.map { it.build() }.toList(),
+        blockOn = blockOn,
+        blockExcept = blockExcept,
+        blockSplit = blockSplit,
+        blockJoin = blockJoin,
+        buffer = bufferSize
+    )
+}

src/commonMain/kotlin/handlers/Worker.kt

@@ -1,13 +1,10 @@
 package com.crowdproj.kotlin.cor.handlers
 
-import com.crowdproj.kotlin.cor.CorDslMarker
-import com.crowdproj.kotlin.cor.ICorChainDsl
-import com.crowdproj.kotlin.cor.ICorExec
-import com.crowdproj.kotlin.cor.ICorWorker
+import com.crowdproj.kotlin.cor.*
 import com.crowdproj.kotlin.cor.base.BaseCorWorkerDsl
 
 @CorDslMarker
-fun <T> ICorChainDsl<T>.worker(
+fun <T> ICorAddExecDsl<T>.worker(
     function: CorWorkerDsl<T>.() -> Unit
 ) {
     add(
@@ -16,7 +13,7 @@ fun <T> ICorChainDsl<T>.worker(
 }
 
 @CorDslMarker
-fun <T> ICorChainDsl<T>.worker(
+fun <T> ICorAddExecDsl<T>.worker(
     title: String,
     description: String = "",
     function: suspend T.() -> Unit

src/commonTest/kotlin/CorBaseTest.kt

@@ -3,6 +3,8 @@ package com.crowdproj.kotlin.cor
 import com.crowdproj.kotlin.cor.handlers.chain
 import com.crowdproj.kotlin.cor.handlers.parallel
 import com.crowdproj.kotlin.cor.handlers.worker
+import com.crowdproj.kotlin.cor.helper.CorStatuses
+import com.crowdproj.kotlin.cor.helper.TestContext
 import kotlinx.coroutines.ExperimentalCoroutinesApi
 import kotlinx.coroutines.test.runTest
 import kotlin.test.Test
@@ -56,20 +58,7 @@ class CorBaseTest {
     }
 }
 
-private fun ICorChainDsl<TestContext>.printResult() = worker(title = "Print example") {
+private fun ICorAddExecDsl<TestContext>.printResult() = worker(title = "Print example") {
     println("some = $some")
 }
 
-data class TestContext(
-    var status: CorStatuses = CorStatuses.NONE,
-    var some: Int = Int.MIN_VALUE,
-    var text: String = "",
-) {
-
-}
-
-enum class CorStatuses {
-    NONE,
-    RUNNING,
-    FAILING,
-}

src/commonTest/kotlin/CorExceptionTest.kt

@@ -2,6 +2,7 @@ package com.crowdproj.kotlin.cor
 
 import com.crowdproj.kotlin.cor.handlers.chain
 import com.crowdproj.kotlin.cor.handlers.worker
+import com.crowdproj.kotlin.cor.helper.TestContext
 import kotlinx.coroutines.ExperimentalCoroutinesApi
 import kotlinx.coroutines.test.runTest
 import kotlin.js.JsName

src/commonTest/kotlin/CorSubChainTest.kt

@@ -0,0 +1,46 @@
+package com.crowdproj.kotlin.cor
+
+import com.crowdproj.kotlin.cor.handlers.subChain
+import com.crowdproj.kotlin.cor.handlers.worker
+import com.crowdproj.kotlin.cor.helper.CorStatuses
+import com.crowdproj.kotlin.cor.helper.TestContext
+import com.crowdproj.kotlin.cor.helper.TestSubContext
+import kotlinx.coroutines.ExperimentalCoroutinesApi
+import kotlinx.coroutines.flow.asFlow
+import kotlinx.coroutines.flow.map
+import kotlinx.coroutines.test.runTest
+import kotlin.test.Test
+import kotlin.test.assertEquals
+
+@OptIn(ExperimentalCoroutinesApi::class)
+class CorSubChainTest {
+    @Test
+    fun createCor() = runTest {
+        val ctx = TestContext()
+        chain.exec(ctx)
+        assertEquals(CorStatuses.RUNNING, ctx.status)
+        assertEquals(65, ctx.some)
+    }
+
+    companion object {
+        val chain = rootChain<TestContext> {
+            worker("status") { status = CorStatuses.RUNNING }
+            worker("init") {
+                some = 0
+            }
+            subChain<TestContext, TestSubContext> {
+                buffer(20)
+                on { status == CorStatuses.RUNNING }
+                split { (1..10).asFlow().map { TestSubContext(temp = it) } }
+                worker {
+                    handle {
+                        temp ++
+                    }
+                }
+                join { sub: TestSubContext ->
+                    some += sub.temp
+                }
+            }
+        }.build()
+    }
+}

src/commonTest/kotlin/CorTest.kt

@@ -1,5 +1,7 @@
 package com.crowdproj.kotlin.cor
 
+import com.crowdproj.kotlin.cor.helper.CorStatuses
+import com.crowdproj.kotlin.cor.helper.TestContext
 import kotlinx.coroutines.ExperimentalCoroutinesApi
 import kotlinx.coroutines.test.runTest
 import kotlin.test.Test

src/commonTest/kotlin/LoopCorBaseTest.kt

@@ -3,6 +3,8 @@ package com.crowdproj.kotlin.cor
 import com.crowdproj.kotlin.cor.handlers.loopUntil
 import com.crowdproj.kotlin.cor.handlers.loopWhile
 import com.crowdproj.kotlin.cor.handlers.worker
+import com.crowdproj.kotlin.cor.helper.CorStatuses
+import com.crowdproj.kotlin.cor.helper.TestContext
 import kotlinx.coroutines.ExperimentalCoroutinesApi
 import kotlinx.coroutines.test.runTest
 import kotlin.test.Test

src/commonTest/kotlin/helper/CorStatuses.kt

@@ -0,0 +1,7 @@
+package com.crowdproj.kotlin.cor.helper
+
+enum class CorStatuses {
+    NONE,
+    RUNNING,
+    FAILING,
+}

src/commonTest/kotlin/helper/TestContext.kt

@@ -0,0 +1,9 @@
+package com.crowdproj.kotlin.cor.helper
+
+data class TestContext(
+    var status: CorStatuses = CorStatuses.NONE,
+    var some: Int = Int.MIN_VALUE,
+    var text: String = "",
+) {
+
+}

src/commonTest/kotlin/helper/TestSubContext.kt

@@ -0,0 +1,6 @@
+package com.crowdproj.kotlin.cor.helper
+
+data class TestSubContext(
+    var temp: Int = Int.MIN_VALUE,
+    var str: String = "",
+)