perf: Netx redis-stream 구현채가 하나의 커넥션으로 모든 트랜잭션을 수신하도록 수정

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

@@ -17,7 +17,6 @@ abstract class AbstractTransactionDispatcher {
 
     fun dispatch(transaction: Transaction, messageId: String): Flux<Any> {
         return Mono.just(transaction.state)
-            .doOnNext { deleteElastic(transaction, messageId) }
             .filter { state -> transactionHandlerFunctions.containsKey(state) }
             .flatMapMany { state ->
                 Flux.fromIterable(
@@ -84,11 +83,6 @@ abstract class AbstractTransactionDispatcher {
         messageId: String
     ): Mono<Pair<Transaction, String>>
 
-    protected abstract fun deleteElastic(
-        transaction: Transaction,
-        messageId: String
-    )
-
     private companion object {
         private val cannotFindMatchedTransactionEventException =
             java.lang.IllegalStateException("Cannot find matched transaction event")

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

@@ -2,18 +2,21 @@ package org.rooftop.netx.engine
 
 import org.rooftop.netx.idl.Transaction
 import reactor.core.publisher.Flux
+import reactor.core.scheduler.Schedulers
 
 abstract class AbstractTransactionListener(
     private val transactionDispatcher: AbstractTransactionDispatcher,
 ) {
 
-    fun subscribeStream(transactionId: String): Flux<Pair<Transaction, String>> {
-        return receive(transactionId)
+    fun subscribeStream() {
+        receive()
             .flatMap { (transaction, messageId) ->
                 transactionDispatcher.dispatch(transaction, messageId)
                     .map { transaction to messageId }
             }
+            .subscribeOn(Schedulers.parallel())
+            .subscribe()
     }
 
-    protected abstract fun receive(transactionId: String): Flux<Pair<Transaction, String>>
+    protected abstract fun receive(): Flux<Pair<Transaction, String>>
 }

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

@@ -5,21 +5,16 @@ import org.rooftop.netx.idl.Transaction
 import org.rooftop.netx.idl.TransactionState
 import org.rooftop.netx.idl.transaction
 import reactor.core.publisher.Mono
-import reactor.core.scheduler.Schedulers
 
 abstract class AbstractTransactionManager(
     nodeId: Int,
     private val nodeGroup: String,
     private val nodeName: String,
     private val transactionIdGenerator: TransactionIdGenerator = TransactionIdGenerator(nodeId),
-    private val transactionListener: AbstractTransactionListener,
-    private val transactionRetrySupporter: AbstractTransactionRetrySupporter,
 ) : TransactionManager {
 
     final override fun start(undo: String): Mono<String> {
         return startTransaction(undo)
-            .subscribeTransaction()
-            .watchTransaction()
             .contextWrite { it.put(CONTEXT_TX_KEY, transactionIdGenerator.generate()) }
     }
 
@@ -37,10 +32,16 @@ abstract class AbstractTransactionManager(
     }
 
     final override fun join(transactionId: String, undo: String): Mono<String> {
-        return exists(transactionId)
+        return findAnyTransaction(transactionId)
+            .map {
+                if (it == TransactionState.TRANSACTION_STATE_ROLLBACK ||
+                    it == TransactionState.TRANSACTION_STATE_COMMIT
+                ) {
+                    error("Cannot join transaction cause, transaction \"$transactionId\" already ${it.name}")
+                }
+                transactionId
+            }
             .joinTransaction(undo)
-            .subscribeTransaction()
-            .watchTransaction()
             .contextWrite { it.put(CONTEXT_TX_KEY, transactionId) }
     }
 
@@ -56,18 +57,6 @@ abstract class AbstractTransactionManager(
         }
     }
 
-    private fun Mono<String>.subscribeTransaction(): Mono<String> {
-        return this.doOnSuccess {
-            transactionListener.subscribeStream(it)
-                .subscribeOn(Schedulers.parallel())
-                .subscribe()
-        }
-    }
-
-    private fun Mono<String>.watchTransaction(): Mono<String> {
-        return this.flatMap { transactionRetrySupporter.watchTransaction(it) }
-    }
-
     final override fun rollback(transactionId: String, cause: String): Mono<String> {
         return exists(transactionId)
             .publishTransaction(transaction {
@@ -93,17 +82,13 @@ abstract class AbstractTransactionManager(
 
     final override fun exists(transactionId: String): Mono<String> {
         return findAnyTransaction(transactionId)
-            .switchIfEmpty(
-                Mono.error {
-                    IllegalStateException("Cannot find exists transaction id \"$transactionId\"")
-                }
-            ).mapTransactionId()
+            .mapTransactionId()
             .contextWrite { it.put(CONTEXT_TX_KEY, transactionId) }
     }
 
-    protected abstract fun findAnyTransaction(transactionId: String): Mono<Transaction>
+    protected abstract fun findAnyTransaction(transactionId: String): Mono<TransactionState>
 
-    protected fun Mono<*>.mapTransactionId(): Mono<String> {
+    private fun Mono<*>.mapTransactionId(): Mono<String> {
         return this.flatMap {
             Mono.deferContextual { Mono.just(it["transactionId"]) }
         }

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

@@ -21,7 +21,5 @@ abstract class AbstractTransactionRetrySupporter(
             .subscribe()
     }
 
-    abstract fun watchTransaction(transactionId: String): Mono<String>
-
     protected abstract fun handleOrphanTransaction(): Flux<Pair<Transaction, String>>
 }

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

@@ -21,7 +21,6 @@ import kotlin.reflect.full.declaredMemberFunctions
 class RedisStreamTransactionDispatcher(
     private val applicationContext: ApplicationContext,
     private val reactiveRedisTemplate: ReactiveRedisTemplate<String, ByteArray>,
-    private val redisStreamTransactionRemover: RedisStreamTransactionRemover,
     private val nodeGroup: String,
 ) : AbstractTransactionDispatcher() {
 
@@ -67,7 +66,7 @@ class RedisStreamTransactionDispatcher(
 
     override fun findOwnTransaction(transaction: Transaction): Mono<Transaction> {
         return reactiveRedisTemplate.opsForStream<String, String>()
-            .read(StreamOffset.create(transaction.id, ReadOffset.from("0")))
+            .read(StreamOffset.create(STREAM_KEY, ReadOffset.from("0")))
             .map { Transaction.parseFrom(it.value["data"]?.toByteArray()) }
             .filter { it.group == nodeGroup }
             .filter { hasUndo(it) }
@@ -80,7 +79,7 @@ class RedisStreamTransactionDispatcher(
 
     override fun ack(transaction: Transaction, messageId: String): Mono<Pair<Transaction, String>> {
         return reactiveRedisTemplate.opsForStream<String, ByteArray>()
-            .acknowledge(transaction.id, nodeGroup, messageId)
+            .acknowledge(STREAM_KEY, nodeGroup, messageId)
             .map { transaction to messageId }
             .switchIfEmpty(
                 Mono.error {
@@ -89,12 +88,9 @@ class RedisStreamTransactionDispatcher(
             )
     }
 
-    override fun deleteElastic(
-        transaction: Transaction,
-        messageId: String
-    ) = redisStreamTransactionRemover.deleteElastic(transaction)
-
     private companion object {
+        private const val STREAM_KEY = "NETX_STREAM"
+
         private val notMatchedTransactionHandlerException =
             IllegalStateException("Cannot find matched Transaction handler")
     }

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

@@ -1,5 +1,6 @@
 package org.rooftop.netx.redis
 
+import io.lettuce.core.RedisBusyException
 import org.rooftop.netx.engine.AbstractTransactionDispatcher
 import org.rooftop.netx.engine.AbstractTransactionListener
 import org.rooftop.netx.idl.Transaction
@@ -10,6 +11,7 @@ import org.springframework.data.redis.connection.stream.StreamOffset
 import org.springframework.data.redis.core.ReactiveRedisTemplate
 import org.springframework.data.redis.stream.StreamReceiver
 import reactor.core.publisher.Flux
+import reactor.core.publisher.Mono
 import reactor.core.scheduler.Schedulers
 import kotlin.time.Duration.Companion.hours
 import kotlin.time.toJavaDuration
@@ -28,20 +30,30 @@ class RedisStreamTransactionListener(
 
     private val receiver = StreamReceiver.create(connectionFactory, options)
 
-    override fun receive(transactionId: String): Flux<Pair<Transaction, String>> {
-        return createGroupIfNotExists(transactionId)
+    override fun receive(): Flux<Pair<Transaction, String>> {
+        return createGroupIfNotExists()
             .flatMap {
                 receiver.receive(
                     Consumer.from(nodeGroup, nodeName),
-                    StreamOffset.create(transactionId, ReadOffset.from(">"))
+                    StreamOffset.create(STREAM_KEY, ReadOffset.from(">"))
                 ).publishOn(Schedulers.parallel())
                     .map { Transaction.parseFrom(it.value["data"]?.toByteArray()) to it.id.value }
             }
     }
 
-    private fun createGroupIfNotExists(transactionId: String): Flux<String> {
+    private fun createGroupIfNotExists(): Flux<String> {
         return reactiveRedisTemplate.opsForStream<String, ByteArray>()
-            .createGroup(transactionId, ReadOffset.from("0"), nodeGroup)
+            .createGroup(STREAM_KEY, ReadOffset.from("0"), nodeGroup)
+            .onErrorResume {
+                if (it.cause is RedisBusyException) {
+                    return@onErrorResume Mono.just("OK")
+                }
+                throw it
+            }
             .flatMapMany { Flux.just(it) }
     }
+
+    private companion object {
+        private const val STREAM_KEY = "NETX_STREAM"
+    }
 }

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

@@ -1,46 +1,65 @@
 package org.rooftop.netx.redis
 
-import org.rooftop.netx.engine.AbstractTransactionListener
+import org.redisson.api.RedissonReactiveClient
 import org.rooftop.netx.engine.AbstractTransactionManager
 import org.rooftop.netx.engine.AbstractTransactionRetrySupporter
 import org.rooftop.netx.idl.Transaction
-import org.springframework.data.domain.Range
+import org.rooftop.netx.idl.TransactionState
 import org.springframework.data.redis.connection.stream.Record
 import org.springframework.data.redis.core.ReactiveRedisTemplate
 import reactor.core.publisher.Mono
+import reactor.core.scheduler.Schedulers
+import java.util.concurrent.TimeUnit
 
 class RedisStreamTransactionManager(
     nodeId: Int,
     nodeName: String,
     nodeGroup: String,
-    transactionListener: AbstractTransactionListener,
     transactionRetrySupporter: AbstractTransactionRetrySupporter,
     private val reactiveRedisTemplate: ReactiveRedisTemplate<String, ByteArray>,
+    private val redissonReactiveClient: RedissonReactiveClient,
 ) : AbstractTransactionManager(
     nodeId = nodeId,
     nodeName = nodeName,
     nodeGroup = nodeGroup,
-    transactionListener = transactionListener,
-    transactionRetrySupporter = transactionRetrySupporter,
 ) {
 
-    override fun findAnyTransaction(transactionId: String): Mono<Transaction> {
-        return reactiveRedisTemplate.opsForStream<String, ByteArray>()
-            .range(transactionId, Range.open("-", "+"))
-            .map { Transaction.parseFrom(it.value[DATA].toString().toByteArray()) }
-            .next()
+    override fun findAnyTransaction(transactionId: String): Mono<TransactionState> {
+        return reactiveRedisTemplate
+            .opsForValue()[transactionId]
+            .switchIfEmpty(
+                Mono.error {
+                    error("Cannot find exists transaction id \"$transactionId\"")
+                }
+            )
+            .map { TransactionState.valueOf(String(it)) }
     }
 
     override fun publishTransaction(transactionId: String, transaction: Transaction): Mono<String> {
         return reactiveRedisTemplate.opsForStream<String, ByteArray>()
             .add(
                 Record.of<String?, String?, ByteArray?>(mapOf(DATA to transaction.toByteArray()))
-                    .withStreamKey(transactionId)
+                    .withStreamKey(STREAM_KEY)
             )
-            .mapTransactionId()
+            .flatMap {
+                redissonReactiveClient.getLock("$transactionId-key")
+                    .tryLock(10, TimeUnit.MINUTES)
+            }
+            .flatMap {
+                reactiveRedisTemplate.opsForValue()
+                    .set(transactionId, transaction.state.name.toByteArray())
+            }
+            .doFinally {
+                redissonReactiveClient.getLock("$transactionId-key")
+                    .forceUnlock()
+                    .subscribeOn(Schedulers.parallel())
+                    .subscribe()
+            }
+            .map { transactionId }
     }
 
     private companion object {
         private const val DATA = "data"
+        private const val STREAM_KEY = "NETX_STREAM"
     }
 }