fix: Publish event after save transaction state (#70)

* refactor: 잘못된 에러 메시지를 수정한다 * fix: 트랜잭션 상태를 저장하기전에 publish하는 버그를 수정한다 * docs: badge version 0.2.6 to 0.2.7
devxb · Mar 4, 2024 · 92eefae · 92eefae
1 parent 8b57da4
commit 92eefae
Show file tree

Hide file tree

Showing 3 changed files with 11 additions and 8 deletions.
diff --git a/README.md b/README.md
@@ -6,7 +6,7 @@
 
 <br>
 
-![version 0.2.6](https://img.shields.io/badge/version-0.2.6-black?labelColor=black&style=flat-square) ![jdk 17](https://img.shields.io/badge/minimum_jdk-17-orange?labelColor=black&style=flat-square) ![load-test](https://img.shields.io/badge/load%20test%2010%2C000%2C000-success-brightgreen?labelColor=black&style=flat-square)    
+![version 0.2.7](https://img.shields.io/badge/version-0.2.7-black?labelColor=black&style=flat-square) ![jdk 17](https://img.shields.io/badge/minimum_jdk-17-orange?labelColor=black&style=flat-square) ![load-test](https://img.shields.io/badge/load%20test%2010%2C000%2C000-success-brightgreen?labelColor=black&style=flat-square)    
 ![redis--stream](https://img.shields.io/badge/-redis--stream-da2020?style=flat-square&logo=Redis&logoColor=white)
 
 Choreography 방식으로 구현된 분산 트랜잭션 라이브러리 입니다.   

diff --git a/src/main/kotlin/org/rooftop/netx/engine/AbstractTransactionManager.kt b/src/main/kotlin/org/rooftop/netx/engine/AbstractTransactionManager.kt
@@ -102,7 +102,7 @@ abstract class AbstractTransactionManager(
                 }
                 transactionId
             }
-            .warningOnError("Cannot join transaction cause, transaction \"$transactionId\" already Rollback state")
+            .warningOnError("Cannot join transaction")
             .map { codec.encode(undo) }
             .flatMap {
                 joinTransaction(transactionId, it, null)
@@ -118,7 +118,7 @@ abstract class AbstractTransactionManager(
                 }
                 transactionId
             }
-            .warningOnError("Cannot join transaction cause, transaction \"$transactionId\" already Rollback state")
+            .warningOnError("Cannot join transaction")
             .map { codec.encode(undo) to codec.encode(event) }
             .flatMap { (encodedUndo, encodedEvent) ->
                 joinTransaction(transactionId, encodedUndo, encodedEvent)

diff --git a/src/main/kotlin/org/rooftop/netx/redis/RedisStreamTransactionManager.kt b/src/main/kotlin/org/rooftop/netx/redis/RedisStreamTransactionManager.kt
@@ -33,11 +33,7 @@ class RedisStreamTransactionManager(
     }
 
     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(STREAM_KEY)
-            )
+        return Mono.fromCallable { hasUndo(transaction) }
             .flatMap {
                 if (hasUndo(transaction)) {
                     return@flatMap reactiveRedisTemplate.opsForHash<String, String>()
@@ -55,6 +51,13 @@ class RedisStreamTransactionManager(
                         )
                     )
             }
+            .flatMap {
+                reactiveRedisTemplate.opsForStream<String, ByteArray>()
+                    .add(
+                        Record.of<String?, String?, ByteArray?>(mapOf(DATA to transaction.toByteArray()))
+                            .withStreamKey(STREAM_KEY)
+                    )
+            }
             .map { transactionId }
     }