저번에 만든게 맘에 안들어서 다시 고민고민 해봄.

AsyncResult<T>를 만들 필요가 있나?

사실 간단한 비동기 결과받기 래퍼로써 만든 것인데 굳이 이걸 쓸 필요가 있을 지 싶다. 그냥 코틀린 코루틴에서 제공하는 Deferred<T>를 사용하는 것이 나을 것이다.

하지만 Deferred<T>는 결과를 위해 대기해야한다.(=suspend fun이라는 말)

기존에 만든 AsyncResult<T>는 대기하는 로직이 없기 때문에 그냥 완료되기 전에 읽는 일이 있으면 그냥 UNSET을 T로 캐스트 할 수 없다고만 나온다. 이게 굉장히 이상하다.

따라서 대기하는 것이 필요한데, 그것이 await()라는 메서드이다. 이건 unaryPlus operator로 그 스코프 내에서만 가능하도록 오버로딩을 해줘서 표기를 간단하게 할 것이다.

해당 로직만 따로 처리하면 안되나?

저번에 만든 예시는 기본적인 로직만 지닌 것이고 그걸 이용하려면 같은 로직을 복사해야한다. 이걸 쉽게 하기 위해서 Mixin을 써서 사용할 것이다.

그래서 나온 코드가 아래다.

@DslMarker
annotation class AsyncScopeDsl

@AsyncScopeDsl
interface AsyncScope<S : SynchronizedScope<S, A>, A : AsynchronousScope<S, A>> {
    suspend operator fun <R> Deferred<R>.unaryPlus(): R = this.await()
    fun synchronizedScope(): S
    fun asynchronousScope(coroutineScope: CoroutineScope): A
}

@AsyncScopeDsl
interface SynchronizedScope<S : SynchronizedScope<S, A>, A : AsynchronousScope<S, A>>
    : AsyncScope<S, A> {
    suspend fun <T> sequential(block: suspend S.() -> T): T = synchronizedScope().block()
    suspend fun <T> parallel(block: suspend A.() -> T): T = coroutineScope {
        asynchronousScope(this).block()
    }
}

@AsyncScopeDsl
interface AsynchronousScope<S : SynchronizedScope<S, A>, A : AsynchronousScope<S, A>> : AsyncScope<S, A> {
    val scope: CoroutineScope

    fun <T> sequential(block: suspend S.() -> T): Deferred<T> {
        return scope.async {
            synchronizedScope().block()
        }
    }

    fun <T> parallel(block: suspend A.() -> T): Deferred<T> {
        return scope.async {
            coroutineScope {
                asynchronousScope(this).block()
            }
        }
    }
}

@AsyncScopeDsl
abstract class AbstractAsyncScope<S : SynchronizedScope<S, A>, A : AsynchronousScope<S, A>> : SynchronizedScope<S, A>

AbstractAsyncScope가 SyncrhonizedScope를 상속받는 것은, 최상단 스코프의 경우 suspend fun 안에서 동기적으로 실행되길 원하기 때문이다.

다음은 사용례

interface WorkerMixin {
    companion object {
        private val log = LoggerFactory.getLogger(WorkerMixin::class.java)
    }

    val minTimeout: Long

    suspend fun delayRandom(maxTimeout: Long) {
        if (minTimeout > maxTimeout) {
            error("maxTimeout must be greater than minTimeout $minTimeout")
        } else if (minTimeout == maxTimeout) {
            delay(maxTimeout.milliseconds)
        } else {
            val timeout = minTimeout + Random.nextLong(maxTimeout - minTimeout)
            delay(timeout.milliseconds)
        }
    }

    suspend fun <R> withDelayRandom(maxTimeout: Long, block: suspend () -> R): R {
        val result = block()
        delayRandom(maxTimeout)
        return result
    }

    suspend fun <R> delayedOutput(taskName: String, maxTimeout: Long, block: suspend () -> R): R {
        log.info {
            "Task $taskName Started at ${
                Clock.System.now().toLocalDateTime(TimeZone.currentSystemDefault())
            }"
        }
        return withDelayRandom(maxTimeout, block).also {
            log.info {
                "Task $taskName Finished at ${
                    Clock.System.now().toLocalDateTime(TimeZone.currentSystemDefault())
                }"
            }
        }
    }
}

우선 대충 만든 WorkerMixin이다. 그냥 최솟값~최댓값 내의 시간으로 랜덤으로 대기하고 결과를 그냥 내놓는다.

interface WorkerAsyncScopeMixin : AsyncScope<SynchronizedWorkerScope, AsynchronousWorkerScope>, WorkerMixin {
    override fun asynchronousScope(coroutineScope: CoroutineScope): AsynchronousWorkerScope =
        AsynchronousWorkerScope(coroutineScope, minTimeout)

    override fun synchronizedScope(): SynchronizedWorkerScope = SynchronizedWorkerScope(minTimeout)
}

class SynchronizedWorkerScope(override val minTimeout: Long) :
    SynchronizedScope<SynchronizedWorkerScope, AsynchronousWorkerScope>, WorkerAsyncScopeMixin

class AsynchronousWorkerScope(override val scope: CoroutineScope, override val minTimeout: Long) :
    AsynchronousScope<SynchronizedWorkerScope, AsynchronousWorkerScope>, WorkerAsyncScopeMixin

class WorkerAsyncScope(override val minTimeout: Long) :
    AbstractAsyncScope<SynchronizedWorkerScope, AsynchronousWorkerScope>(), WorkerAsyncScopeMixin

@Test
fun testDsl(): Unit = runTest {
    val resultData = with(WorkerAsyncScope(1000)) {
        val task1Result = parallel {
            val task1SubResult1 = sequential {
                val task1SubResult1Data1 = delayedOutput("task1-subResult1-data1", 3000) { 1 }
                val task1SubResult1Data2 = delayedOutput("task1-subResult1-data2", 1100) { 2 }
                mapOf(
                    "task1-subResult1-data1" to task1SubResult1Data1,
                    "task1-subResult1-data2" to task1SubResult1Data2,
                )
            }
            val task1SubResult2 = sequential {
                val task1SubResult2Data = delayedOutput("task1-subResult2-data1", 1500) { 3 }
                mapOf("task1-subResult2-data" to task1SubResult2Data)
            }
            +task1SubResult1 + +task1SubResult2
        }
        val task2Result = sequential {
            val task2SubResult1 = delayedOutput("task2-subResult1", 1000) { 4 }
            val task2SubResult2 = delayedOutput("task2-subResult2", 1010) { 5 }
            mapOf(
                "task2-subResult1" to task2SubResult1,
                "task2-subResult2" to task2SubResult2,
            )
        }
        task1Result + task2Result
    }

    log.info { "result: $resultData" }
}

끗.