feat: 반환타입이 Publisher가 아니여도 @TransactionHandler 를 사용할 수 있도록 구현

src/main/kotlin/org/rooftop/netx/engine/AbstractTransactionDispatcher.kt

@@ -1,37 +1,62 @@
 package org.rooftop.netx.engine
 
+import jakarta.annotation.PostConstruct
 import org.rooftop.netx.api.*
 import org.rooftop.netx.idl.Transaction
 import org.rooftop.netx.idl.TransactionState
+import org.rooftop.netx.meta.TransactionHandler
 import reactor.core.publisher.Flux
 import reactor.core.publisher.Mono
 import reactor.core.scheduler.Schedulers
+import kotlin.reflect.KClass
 import kotlin.reflect.KFunction
+import kotlin.reflect.full.declaredMemberFunctions
 
 abstract class AbstractTransactionDispatcher {
 
-    protected val transactionHandlerFunctions =
-        mutableMapOf<TransactionState, MutableList<Pair<KFunction<Mono<Any>>, Any>>>()
+    private val monoTransactionHandleFunctions =
+        mutableMapOf<TransactionState, MutableList<Pair<KFunction<Mono<*>>, Any>>>()
 
-    protected abstract fun initHandlers()
+    private val notPublisherTransactionHandlerFunctions =
+        mutableMapOf<TransactionState, MutableList<Pair<KFunction<*>, Any>>>()
 
     fun dispatch(transaction: Transaction, messageId: String): Flux<Any> {
+        return dispatchToMonoHandler(transaction)
+            .flatMap { dispatchToNotPublisherHandler(transaction) }
+            .doOnComplete {
+                ack(transaction, messageId)
+                    .subscribeOn(Schedulers.boundedElastic())
+                    .subscribe()
+            }
+    }
+
+    private fun dispatchToMonoHandler(transaction: Transaction): Flux<Any> {
         return Mono.just(transaction.state)
-            .filter { state -> transactionHandlerFunctions.containsKey(state) }
+            .filter { state -> monoTransactionHandleFunctions.containsKey(state) }
             .flatMapMany { state ->
                 Flux.fromIterable(
-                    transactionHandlerFunctions[state]
+                    monoTransactionHandleFunctions[state]
                         ?: throw cannotFindMatchedHandlerFunctionException
                 )
             }
             .flatMap { (function, instance) ->
                 mapToTransactionEvent(transaction)
                     .flatMap { function.call(instance, it) }
             }
-            .doOnComplete {
-                ack(transaction, messageId)
-                    .subscribeOn(Schedulers.boundedElastic())
-                    .subscribe()
+    }
+
+    private fun dispatchToNotPublisherHandler(transaction: Transaction): Flux<*> {
+        return Mono.just(transaction.state)
+            .filter { state -> notPublisherTransactionHandlerFunctions.containsKey(state) }
+            .flatMapMany { state ->
+                Flux.fromIterable(
+                    notPublisherTransactionHandlerFunctions[state]
+                        ?: throw cannotFindMatchedHandlerFunctionException
+                )
+            }
+            .flatMap { (function, instance) ->
+                mapToTransactionEvent(transaction)
+                    .map { function.call(instance, it) }
             }
     }
 
@@ -78,6 +103,81 @@ abstract class AbstractTransactionDispatcher {
 
     protected abstract fun findOwnUndo(transaction: Transaction): Mono<String>
 
+    @PostConstruct
+    fun initHandler() {
+        val transactionHandler = findHandlers(TransactionHandler::class)
+        val monoFunctions = getFunctions(transactionHandler, Mono::class)
+        monoTransactionHandleFunctions.putAll(monoFunctions)
+        val notPublisherFunctions = getNotPublisherFunctions(transactionHandler)
+        notPublisherTransactionHandlerFunctions.putAll(notPublisherFunctions)
+    }
+
+    @Suppress("unchecked_cast")
+    private fun <T : Any> getFunctions(
+        foundHandlers: List<Any>,
+        returnType: KClass<T>
+    ): MutableMap<TransactionState, MutableList<Pair<KFunction<T>, Any>>> {
+        val handlers = mutableMapOf<TransactionState, MutableList<Pair<KFunction<T>, Any>>>()
+
+        for (handler in foundHandlers) {
+            val returnTypeMatchedHandlers = handler::class.declaredMemberFunctions
+                .filter { it.returnType.classifier == returnType }
+
+            returnTypeMatchedHandlers.forEach { function ->
+                function.annotations
+                    .forEach { annotation ->
+                        runCatching {
+                            val transactionState = matchedTransactionState(annotation)
+                            handlers.putIfAbsent(transactionState, mutableListOf())
+                            handlers[transactionState]?.add(function as KFunction<T> to handler)
+                        }.onFailure {
+                            throw IllegalStateException("Cannot add TransactionHandler", it)
+                        }
+                    }
+            }
+        }
+
+        return handlers
+    }
+
+    private fun getNotPublisherFunctions(
+        foundHandlers: List<Any>
+    ): MutableMap<TransactionState, MutableList<Pair<KFunction<*>, Any>>> {
+        val handlers = mutableMapOf<TransactionState, MutableList<Pair<KFunction<*>, Any>>>()
+
+        for (handler in foundHandlers) {
+            val returnTypeMatchedHandlers = handler::class.declaredMemberFunctions
+                .filter { it.returnType.classifier != Mono::class && it.returnType.classifier != Flux::class }
+
+            returnTypeMatchedHandlers.forEach { function ->
+                function.annotations
+                    .forEach { annotation ->
+                        runCatching {
+                            val transactionState = matchedTransactionState(annotation)
+                            handlers.putIfAbsent(transactionState, mutableListOf())
+                            handlers[transactionState]?.add(function to handler)
+                        }.onFailure {
+                            throw IllegalStateException("Cannot add TransactionHandler", it)
+                        }
+                    }
+            }
+        }
+
+        return handlers
+    }
+
+    protected abstract fun <T : Annotation> findHandlers(type: KClass<T>): List<Any>
+
+    private fun matchedTransactionState(annotation: Annotation): TransactionState {
+        return when (annotation) {
+            is TransactionStartHandler -> TransactionState.TRANSACTION_STATE_START
+            is TransactionCommitHandler -> TransactionState.TRANSACTION_STATE_COMMIT
+            is TransactionJoinHandler -> TransactionState.TRANSACTION_STATE_JOIN
+            is TransactionRollbackHandler -> TransactionState.TRANSACTION_STATE_ROLLBACK
+            else -> throw notMatchedTransactionHandlerException
+        }
+    }
+
     protected abstract fun ack(
         transaction: Transaction,
         messageId: String
@@ -89,5 +189,8 @@ abstract class AbstractTransactionDispatcher {
 
         private val cannotFindMatchedHandlerFunctionException =
             IllegalStateException("Cannot find matched handler function")
+
+        private val notMatchedTransactionHandlerException =
+            IllegalStateException("Cannot find matched Transaction handler")
     }
 }

src/main/kotlin/org/rooftop/netx/redis/RedisStreamTransactionDispatcher.kt

@@ -1,67 +1,25 @@
 package org.rooftop.netx.redis
 
-import jakarta.annotation.PostConstruct
-import org.rooftop.netx.api.TransactionCommitHandler
-import org.rooftop.netx.api.TransactionJoinHandler
-import org.rooftop.netx.api.TransactionRollbackHandler
-import org.rooftop.netx.api.TransactionStartHandler
 import org.rooftop.netx.engine.AbstractTransactionDispatcher
 import org.rooftop.netx.idl.Transaction
-import org.rooftop.netx.idl.TransactionState
-import org.rooftop.netx.meta.TransactionHandler
 import org.springframework.context.ApplicationContext
 import org.springframework.data.redis.core.ReactiveRedisTemplate
 import reactor.core.publisher.Mono
 import kotlin.reflect.KClass
-import kotlin.reflect.KFunction
-import kotlin.reflect.full.declaredMemberFunctions
 
 class RedisStreamTransactionDispatcher(
     private val applicationContext: ApplicationContext,
     private val reactiveRedisTemplate: ReactiveRedisTemplate<String, ByteArray>,
     private val nodeGroup: String,
 ) : AbstractTransactionDispatcher() {
 
-    @PostConstruct
-    @Suppress("Unchecked_cast")
-    override fun initHandlers() {
-        val transactionHandler = findHandlers(TransactionHandler::class)
-        transactionHandler.forEach { handler ->
-            handler::class.declaredMemberFunctions
-                .filter { it.returnType.classifier == Mono::class }
-                .forEach { function ->
-                    function.annotations
-                        .forEach { annotation ->
-                            runCatching {
-                                val transactionState = matchedTransactionState(annotation)
-                                transactionHandlerFunctions.putIfAbsent(
-                                    transactionState,
-                                    mutableListOf()
-                                )
-                                transactionHandlerFunctions[transactionState]?.add(function as KFunction<Mono<Any>> to handler)
-                            }
-                        }
-                }
-        }
-    }
-
-    private fun <T : Annotation> findHandlers(type: KClass<T>): List<Any> {
+    override fun <T : Annotation> findHandlers(type: KClass<T>): List<Any> {
         return applicationContext.getBeansWithAnnotation(type.java)
             .entries.asSequence()
             .map { it.value }
             .toList()
     }
 
-    private fun matchedTransactionState(annotation: Annotation): TransactionState {
-        return when (annotation) {
-            is TransactionStartHandler -> TransactionState.TRANSACTION_STATE_START
-            is TransactionCommitHandler -> TransactionState.TRANSACTION_STATE_COMMIT
-            is TransactionJoinHandler -> TransactionState.TRANSACTION_STATE_JOIN
-            is TransactionRollbackHandler -> TransactionState.TRANSACTION_STATE_ROLLBACK
-            else -> throw notMatchedTransactionHandlerException
-        }
-    }
-
     override fun findOwnUndo(transaction: Transaction): Mono<String> {
         return reactiveRedisTemplate.opsForHash<String, String>()[transaction.id, nodeGroup]
             .switchIfEmpty(
@@ -84,8 +42,5 @@ class RedisStreamTransactionDispatcher(
 
     private companion object {
         private const val STREAM_KEY = "NETX_STREAM"
-
-        private val notMatchedTransactionHandlerException =
-            IllegalStateException("Cannot find matched Transaction handler")
     }
 }

src/test/kotlin/org/rooftop/netx/redis/AbstractTransactionHandlerAssertions.kt

@@ -0,0 +1,32 @@
+package org.rooftop.netx.redis
+
+import io.kotest.matchers.shouldBe
+
+abstract class AbstractTransactionHandlerAssertions {
+
+    private val methodInvocationCounts = mutableMapOf<String, Int>()
+
+    fun clear() {
+        methodInvocationCounts.clear()
+    }
+
+    fun joinCountShouldBe(count: Int) {
+        (methodInvocationCounts["JOIN"] ?: 0) shouldBe count
+    }
+
+    fun startCountShouldBe(count: Int) {
+        (methodInvocationCounts["START"] ?: 0) shouldBe count
+    }
+
+    fun commitCountShouldBe(count: Int) {
+        (methodInvocationCounts["COMMIT"] ?: 0) shouldBe count
+    }
+
+    fun rollbackCountShouldBe(count: Int) {
+        (methodInvocationCounts["ROLLBACK"] ?: 0) shouldBe count
+    }
+
+    protected fun put(key: String) {
+        methodInvocationCounts[key] = methodInvocationCounts.getOrDefault(key, 0) + 1
+    }
+}

src/test/kotlin/org/rooftop/netx/redis/TransactionHandlerAssertions.kt → src/test/kotlin/org/rooftop/netx/redis/MonoTransactionHandlerAssertions.kt

@@ -1,34 +1,11 @@
 package org.rooftop.netx.redis
 
-import io.kotest.matchers.shouldBe
 import org.rooftop.netx.api.*
 import org.rooftop.netx.meta.TransactionHandler
 import reactor.core.publisher.Mono
 
 @TransactionHandler
-class TransactionHandlerAssertions {
-
-    private val methodInvocationCounts = mutableMapOf<String, Int>()
-
-    fun clear() {
-        methodInvocationCounts.clear()
-    }
-
-    fun joinCountShouldBe(count: Int) {
-        (methodInvocationCounts["JOIN"] ?: 0) shouldBe count
-    }
-
-    fun startCountShouldBe(count: Int) {
-        (methodInvocationCounts["START"] ?: 0) shouldBe count
-    }
-
-    fun commitCountShouldBe(count: Int) {
-        (methodInvocationCounts["COMMIT"] ?: 0) shouldBe count
-    }
-
-    fun rollbackCountShouldBe(count: Int) {
-        (methodInvocationCounts["ROLLBACK"] ?: 0) shouldBe count
-    }
+class MonoTransactionHandlerAssertions : AbstractTransactionHandlerAssertions() {
 
     @TransactionRollbackHandler
     fun handleRollback(event: TransactionRollbackEvent): Mono<Unit> {
@@ -54,7 +31,4 @@ class TransactionHandlerAssertions {
         return Mono.just(Unit)
     }
 
-    private fun put(key: String) {
-        methodInvocationCounts[key] = methodInvocationCounts.getOrDefault(key, 0) + 1
-    }
 }

src/test/kotlin/org/rooftop/netx/redis/NoAckRedisStreamTransactionDispatcher.kt

@@ -1,64 +1,20 @@
 package org.rooftop.netx.redis
 
-import org.rooftop.netx.api.TransactionCommitHandler
-import org.rooftop.netx.api.TransactionJoinHandler
-import org.rooftop.netx.api.TransactionRollbackHandler
-import org.rooftop.netx.api.TransactionStartHandler
 import org.rooftop.netx.engine.AbstractTransactionDispatcher
 import org.rooftop.netx.idl.Transaction
-import org.rooftop.netx.idl.TransactionState
-import org.rooftop.netx.meta.TransactionHandler
 import org.springframework.context.ApplicationContext
-import org.springframework.data.redis.connection.stream.ReadOffset
-import org.springframework.data.redis.connection.stream.StreamOffset
 import org.springframework.data.redis.core.ReactiveRedisTemplate
 import reactor.core.publisher.Mono
 import kotlin.reflect.KClass
-import kotlin.reflect.KFunction
-import kotlin.reflect.full.declaredMemberFunctions
 
 class NoAckRedisStreamTransactionDispatcher(
     private val applicationContext: ApplicationContext,
     private val reactiveRedisTemplate: ReactiveRedisTemplate<String, ByteArray>,
     private val nodeGroup: String,
 ) : AbstractTransactionDispatcher() {
-    @Suppress("Unchecked_cast")
-    override fun initHandlers() {
-        val transactionHandler = findHandlers(TransactionHandler::class)
-        transactionHandler.forEach { handler ->
-            handler::class.declaredMemberFunctions
-                .filter { it.returnType is Mono<*> }
-                .forEach { function ->
-                    function.annotations
-                        .forEach { annotation ->
-                            runCatching {
-                                val transactionState = matchedTransactionState(annotation)
-                                val handlerFunctions = transactionHandlerFunctions.getOrDefault(
-                                    transactionState,
-                                    mutableListOf()
-                                )
-                                handlerFunctions.add(function as KFunction<Mono<Any>> to handler)
-                            }
-                        }
-                }
-        }
-    }
-
-    private fun <T : Annotation> findHandlers(type: KClass<T>): List<Any> {
-        return applicationContext.getBeansWithAnnotation(type.java)
-            .entries.asSequence()
-            .map { it.value }
-            .toList()
-    }
 
-    private fun matchedTransactionState(annotation: Annotation): TransactionState {
-        return when (annotation) {
-            is TransactionStartHandler -> TransactionState.TRANSACTION_STATE_START
-            is TransactionCommitHandler -> TransactionState.TRANSACTION_STATE_COMMIT
-            is TransactionJoinHandler -> TransactionState.TRANSACTION_STATE_JOIN
-            is TransactionRollbackHandler -> TransactionState.TRANSACTION_STATE_ROLLBACK
-            else -> throw notMatchedTransactionHandlerException
-        }
+    override fun <T : Annotation> findHandlers(type: KClass<T>): List<Any> {
+        return listOf()
     }
 
     override fun findOwnUndo(transaction: Transaction): Mono<String> {
@@ -72,10 +28,5 @@ class NoAckRedisStreamTransactionDispatcher(
 
     override fun ack(transaction: Transaction, messageId: String): Mono<Pair<Transaction, String>> =
         Mono.just(transaction to messageId)
-
-    private companion object {
-        private val notMatchedTransactionHandlerException =
-            IllegalStateException("Cannot find matched Transaction handler")
-    }
 }