diff --git a/src/main/kotlin/org/rooftop/netx/engine/AbstractTransactionDispatcher.kt b/src/main/kotlin/org/rooftop/netx/engine/AbstractTransactionDispatcher.kt index b4f5efc..2b0b55a 100644 --- a/src/main/kotlin/org/rooftop/netx/engine/AbstractTransactionDispatcher.kt +++ b/src/main/kotlin/org/rooftop/netx/engine/AbstractTransactionDispatcher.kt @@ -1,26 +1,41 @@ 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>, Any>>>() + private val monoTransactionHandleFunctions = + mutableMapOf>, Any>>>() - protected abstract fun initHandlers() + private val notPublisherTransactionHandlerFunctions = + mutableMapOf, Any>>>() fun dispatch(transaction: Transaction, messageId: String): Flux { + return dispatchToMonoHandler(transaction) + .flatMap { dispatchToNotPublisherHandler(transaction) } + .doOnComplete { + ack(transaction, messageId) + .subscribeOn(Schedulers.boundedElastic()) + .subscribe() + } + } + + private fun dispatchToMonoHandler(transaction: Transaction): Flux { 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 ) } @@ -28,10 +43,20 @@ abstract class AbstractTransactionDispatcher { 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 + @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 getFunctions( + foundHandlers: List, + returnType: KClass + ): MutableMap, Any>>> { + val handlers = mutableMapOf, 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 to handler) + }.onFailure { + throw IllegalStateException("Cannot add TransactionHandler", it) + } + } + } + } + + return handlers + } + + private fun getNotPublisherFunctions( + foundHandlers: List + ): MutableMap, Any>>> { + val handlers = mutableMapOf, 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 findHandlers(type: KClass): List + + 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") } } diff --git a/src/main/kotlin/org/rooftop/netx/redis/RedisStreamTransactionDispatcher.kt b/src/main/kotlin/org/rooftop/netx/redis/RedisStreamTransactionDispatcher.kt index 7263cd0..40475c3 100644 --- a/src/main/kotlin/org/rooftop/netx/redis/RedisStreamTransactionDispatcher.kt +++ b/src/main/kotlin/org/rooftop/netx/redis/RedisStreamTransactionDispatcher.kt @@ -1,20 +1,11 @@ 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, @@ -22,46 +13,13 @@ class RedisStreamTransactionDispatcher( 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> to handler) - } - } - } - } - } - - private fun findHandlers(type: KClass): List { + override fun findHandlers(type: KClass): List { 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 { return reactiveRedisTemplate.opsForHash()[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") } } diff --git a/src/test/kotlin/org/rooftop/netx/redis/AbstractTransactionHandlerAssertions.kt b/src/test/kotlin/org/rooftop/netx/redis/AbstractTransactionHandlerAssertions.kt new file mode 100644 index 0000000..834c9ef --- /dev/null +++ b/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() + + 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 + } +} diff --git a/src/test/kotlin/org/rooftop/netx/redis/TransactionHandlerAssertions.kt b/src/test/kotlin/org/rooftop/netx/redis/MonoTransactionHandlerAssertions.kt similarity index 50% rename from src/test/kotlin/org/rooftop/netx/redis/TransactionHandlerAssertions.kt rename to src/test/kotlin/org/rooftop/netx/redis/MonoTransactionHandlerAssertions.kt index 9cfa3a9..4428453 100644 --- a/src/test/kotlin/org/rooftop/netx/redis/TransactionHandlerAssertions.kt +++ b/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() - - 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 { @@ -54,7 +31,4 @@ class TransactionHandlerAssertions { return Mono.just(Unit) } - private fun put(key: String) { - methodInvocationCounts[key] = methodInvocationCounts.getOrDefault(key, 0) + 1 - } } diff --git a/src/test/kotlin/org/rooftop/netx/redis/NoAckRedisStreamTransactionDispatcher.kt b/src/test/kotlin/org/rooftop/netx/redis/NoAckRedisStreamTransactionDispatcher.kt index 00a060b..8d32e99 100644 --- a/src/test/kotlin/org/rooftop/netx/redis/NoAckRedisStreamTransactionDispatcher.kt +++ b/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, 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> to handler) - } - } - } - } - } - - private fun findHandlers(type: KClass): List { - 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 findHandlers(type: KClass): List { + return listOf() } override fun findOwnUndo(transaction: Transaction): Mono { @@ -72,10 +28,5 @@ class NoAckRedisStreamTransactionDispatcher( override fun ack(transaction: Transaction, messageId: String): Mono> = Mono.just(transaction to messageId) - - private companion object { - private val notMatchedTransactionHandlerException = - IllegalStateException("Cannot find matched Transaction handler") - } } diff --git a/src/test/kotlin/org/rooftop/netx/redis/NoPublisherTransactionHandlerAssertions.kt b/src/test/kotlin/org/rooftop/netx/redis/NoPublisherTransactionHandlerAssertions.kt new file mode 100644 index 0000000..ec70a2b --- /dev/null +++ b/src/test/kotlin/org/rooftop/netx/redis/NoPublisherTransactionHandlerAssertions.kt @@ -0,0 +1,34 @@ +package org.rooftop.netx.redis + +import org.rooftop.netx.api.* +import org.rooftop.netx.meta.TransactionHandler + +@TransactionHandler +class NoPublisherTransactionHandlerAssertions : AbstractTransactionHandlerAssertions() { + + @TransactionRollbackHandler + fun handleRollback(event: TransactionRollbackEvent): Long { + put("ROLLBACK") + return Long.MIN_VALUE + } + + @TransactionCommitHandler + fun handleCommit(event: TransactionCommitEvent) { + put("COMMIT") + } + + @TransactionStartHandler + fun handleStart(event: TransactionStartEvent): Foo { + put("START") + return Foo("START") + } + + @TransactionJoinHandler + fun handleJoin(event: TransactionJoinEvent): Any { + put("JOIN") + return Any::class + } + + class Foo(name: String) + +} diff --git a/src/test/kotlin/org/rooftop/netx/redis/RedisStreamTransactionManagerTest.kt b/src/test/kotlin/org/rooftop/netx/redis/RedisStreamTransactionManagerTest.kt index 816e554..c1ed219 100644 --- a/src/test/kotlin/org/rooftop/netx/redis/RedisStreamTransactionManagerTest.kt +++ b/src/test/kotlin/org/rooftop/netx/redis/RedisStreamTransactionManagerTest.kt @@ -8,24 +8,27 @@ import org.rooftop.netx.meta.EnableDistributedTransaction import org.springframework.test.context.ContextConfiguration import org.springframework.test.context.TestPropertySource import reactor.test.StepVerifier -import kotlin.time.Duration.Companion.minutes +import kotlin.time.Duration.Companion.seconds @EnableDistributedTransaction @ContextConfiguration( classes = [ RedisContainer::class, - TransactionHandlerAssertions::class, + MonoTransactionHandlerAssertions::class, + NoPublisherTransactionHandlerAssertions::class, ] ) @DisplayName("RedisStreamTransactionManager 클래스의") @TestPropertySource("classpath:application.properties") internal class RedisStreamTransactionManagerTest( private val transactionManager: TransactionManager, - private val transactionHandlerAssertions: TransactionHandlerAssertions, + private val monoTransactionHandlerAssertions: MonoTransactionHandlerAssertions, + private val noPublisherTransactionHandlerAssertions: NoPublisherTransactionHandlerAssertions, ) : DescribeSpec({ beforeEach { - transactionHandlerAssertions.clear() + monoTransactionHandlerAssertions.clear() + noPublisherTransactionHandlerAssertions.clear() } describe("start 메소드는") { @@ -33,8 +36,9 @@ internal class RedisStreamTransactionManagerTest( it("트랜잭션을 시작하고 transaction-id를 반환한다.") { transactionManager.start(REPLAY).subscribe() - eventually(5.minutes) { - transactionHandlerAssertions.startCountShouldBe(1) + eventually(5.seconds) { + monoTransactionHandlerAssertions.startCountShouldBe(1) + noPublisherTransactionHandlerAssertions.startCountShouldBe(1) } } } @@ -44,8 +48,9 @@ internal class RedisStreamTransactionManagerTest( transactionManager.start(REPLAY).block() transactionManager.start(REPLAY).block() - eventually(5.minutes) { - transactionHandlerAssertions.startCountShouldBe(2) + eventually(5.seconds) { + monoTransactionHandlerAssertions.startCountShouldBe(2) + noPublisherTransactionHandlerAssertions.startCountShouldBe(2) } } } @@ -58,8 +63,9 @@ internal class RedisStreamTransactionManagerTest( it("트랜잭션에 참여한다.") { transactionManager.join(transactionId, REPLAY).subscribe() - eventually(5.minutes) { - transactionHandlerAssertions.joinCountShouldBe(1) + eventually(5.seconds) { + monoTransactionHandlerAssertions.joinCountShouldBe(1) + noPublisherTransactionHandlerAssertions.joinCountShouldBe(1) } } } @@ -104,8 +110,9 @@ internal class RedisStreamTransactionManagerTest( it("commit 메시지를 publish 한다") { transactionManager.commit(transactionId).block() - eventually(5.minutes) { - transactionHandlerAssertions.commitCountShouldBe(1) + eventually(5.seconds) { + monoTransactionHandlerAssertions.commitCountShouldBe(1) + noPublisherTransactionHandlerAssertions.commitCountShouldBe(1) } } } @@ -127,8 +134,9 @@ internal class RedisStreamTransactionManagerTest( it("rollback 메시지를 publish 한다") { transactionManager.rollback(transactionId, "rollback occured for test").block() - eventually(5.minutes) { - transactionHandlerAssertions.rollbackCountShouldBe(1) + eventually(5.seconds) { + monoTransactionHandlerAssertions.rollbackCountShouldBe(1) + noPublisherTransactionHandlerAssertions.rollbackCountShouldBe(1) } } } diff --git a/src/test/kotlin/org/rooftop/netx/redis/RedisTransactionRetrySupporterTest.kt b/src/test/kotlin/org/rooftop/netx/redis/RedisTransactionRetrySupporterTest.kt index ff7b3e7..f29e992 100644 --- a/src/test/kotlin/org/rooftop/netx/redis/RedisTransactionRetrySupporterTest.kt +++ b/src/test/kotlin/org/rooftop/netx/redis/RedisTransactionRetrySupporterTest.kt @@ -6,14 +6,15 @@ import io.kotest.core.spec.style.DescribeSpec import org.rooftop.netx.api.TransactionManager import org.springframework.test.context.ContextConfiguration import org.springframework.test.context.TestPropertySource -import kotlin.time.Duration.Companion.minutes +import kotlin.time.Duration.Companion.seconds @ContextConfiguration( classes = [ RedisContainer::class, RedisAssertions::class, NoAckRedisTransactionConfigurer::class, - TransactionHandlerAssertions::class, + MonoTransactionHandlerAssertions::class, + NoPublisherTransactionHandlerAssertions::class, ] ) @TestPropertySource("classpath:application.properties") @@ -21,11 +22,13 @@ import kotlin.time.Duration.Companion.minutes internal class RedisTransactionRetrySupporterTest( private val redisAssertions: RedisAssertions, private val transactionManager: TransactionManager, - private val transactionHandlerAssertions: TransactionHandlerAssertions, + private val monoTransactionHandlerAssertions: MonoTransactionHandlerAssertions, + private val noPublisherTransactionHandlerAssertions: NoPublisherTransactionHandlerAssertions, ) : DescribeSpec({ beforeEach { - transactionHandlerAssertions.clear() + monoTransactionHandlerAssertions.clear() + noPublisherTransactionHandlerAssertions.clear() } describe("handleOrphanTransaction 메소드는") { @@ -33,10 +36,10 @@ internal class RedisTransactionRetrySupporterTest( it("해당 트랜잭션을 찾아서 처리하고, ack 상태로 변경한다.") { val transactionId = transactionManager.start("undo").block()!! - Thread.sleep(3_000) + eventually(10.seconds) { + noPublisherTransactionHandlerAssertions.startCountShouldBe(1) + monoTransactionHandlerAssertions.startCountShouldBe(1) - eventually(10.minutes) { - transactionHandlerAssertions.startCountShouldBe(1) redisAssertions.pendingMessageCountShouldBe(transactionId, 0) } }