From 92eefaefb1290599b52c1a1a25dc0683547af503 Mon Sep 17 00:00:00 2001
From: xb205 <62425964+devxb@users.noreply.github.com>
Date: Mon, 4 Mar 2024 15:25:29 +0900
Subject: [PATCH] fix: Publish event after save transaction state (#70)
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

* refactor: 잘못된 에러 메시지를 수정한다

* fix: 트랜잭션 상태를 저장하기전에 publish하는 버그를 수정한다

* docs: badge version 0.2.6 to 0.2.7
---
 README.md                                           |  2 +-
 .../netx/engine/AbstractTransactionManager.kt       |  4 ++--
 .../netx/redis/RedisStreamTransactionManager.kt     | 13 ++++++++-----
 3 files changed, 11 insertions(+), 8 deletions(-)

diff --git a/README.md b/README.md
index afd7b58..fd93ac5 100644
--- 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
index 172d012..b200847 100644
--- 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
index 128f533..377515b 100644
--- 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 }
     }