From d197a9bd32ca69b81c9b4d02936ca54f0ae7d526 Mon Sep 17 00:00:00 2001
From: sokomishalov <sokomishalov@mail.ru>
Date: Mon, 5 Oct 2020 13:25:23 +0300
Subject: [PATCH] Fix EXEC without MULTI when using coroutines over async #1441

---
 .../api/async/RedisAsyncCommandsExtensions.kt | 13 ++++++-------
 .../RedisCoroutinesCommandsExtensions.kt      | 13 ++++++-------
 .../api/sync/RedisSyncCommandsExtensions.kt   | 13 ++++++-------
 .../TransactionExtensionsIntegrationTests.kt  | 19 +++++++++++++++++++
 4 files changed, 37 insertions(+), 21 deletions(-)

diff --git a/src/main/kotlin/io/lettuce/core/api/async/RedisAsyncCommandsExtensions.kt b/src/main/kotlin/io/lettuce/core/api/async/RedisAsyncCommandsExtensions.kt
index f74d24a0d9..88fb16b3b0 100644
--- a/src/main/kotlin/io/lettuce/core/api/async/RedisAsyncCommandsExtensions.kt
+++ b/src/main/kotlin/io/lettuce/core/api/async/RedisAsyncCommandsExtensions.kt
@@ -26,12 +26,11 @@ import kotlinx.coroutines.future.await
  * @since 6.0
  */
 @ExperimentalLettuceCoroutinesApi
-suspend inline fun <K, V> RedisAsyncCommands<K, V>.multi(action: RedisAsyncCommands<K, V>.() -> Unit): TransactionResult {
+suspend inline fun <K, V> RedisAsyncCommands<K, V>.multi(action: RedisAsyncCommands<K, V>.() -> Unit): TransactionResult = try {
     multi().await()
-    runCatching {
-        action.invoke(this)
-    }.onFailure {
-        discard()
-    }
-    return exec().await()
+    action.invoke(this)
+    exec().await()
+} catch (thr: Throwable) {
+    discard().await()
+    throw thr
 }
diff --git a/src/main/kotlin/io/lettuce/core/api/coroutines/RedisCoroutinesCommandsExtensions.kt b/src/main/kotlin/io/lettuce/core/api/coroutines/RedisCoroutinesCommandsExtensions.kt
index ccfca58c21..33579410bf 100644
--- a/src/main/kotlin/io/lettuce/core/api/coroutines/RedisCoroutinesCommandsExtensions.kt
+++ b/src/main/kotlin/io/lettuce/core/api/coroutines/RedisCoroutinesCommandsExtensions.kt
@@ -25,12 +25,11 @@ import io.lettuce.core.TransactionResult
  * @since 6.0
  */
 @ExperimentalLettuceCoroutinesApi
-suspend inline fun <K : Any, V : Any> RedisCoroutinesCommands<K, V>.multi(action: RedisCoroutinesCommands<K, V>.() -> Unit): TransactionResult? {
+suspend inline fun <K : Any, V : Any> RedisCoroutinesCommands<K, V>.multi(action: RedisCoroutinesCommands<K, V>.() -> Unit): TransactionResult? = try {
     multi()
-    runCatching {
-        action.invoke(this)
-    }.onFailure {
-        discard()
-    }
-    return exec()
+    action.invoke(this)
+    exec()
+} catch (thr: Throwable) {
+    discard()
+    throw thr
 }
diff --git a/src/main/kotlin/io/lettuce/core/api/sync/RedisSyncCommandsExtensions.kt b/src/main/kotlin/io/lettuce/core/api/sync/RedisSyncCommandsExtensions.kt
index d8e502aff9..523279c6c1 100644
--- a/src/main/kotlin/io/lettuce/core/api/sync/RedisSyncCommandsExtensions.kt
+++ b/src/main/kotlin/io/lettuce/core/api/sync/RedisSyncCommandsExtensions.kt
@@ -25,12 +25,11 @@ import io.lettuce.core.TransactionResult
  * @since 6.0
  */
 @ExperimentalLettuceCoroutinesApi
-inline fun <K, V> RedisCommands<K, V>.multi(action: RedisCommands<K, V>.() -> Unit): TransactionResult {
+inline fun <K, V> RedisCommands<K, V>.multi(action: RedisCommands<K, V>.() -> Unit): TransactionResult = try {
     multi()
-    runCatching {
-        action.invoke(this)
-    }.onFailure {
-        discard()
-    }
-    return exec()
+    action.invoke(this)
+    exec()
+} catch (thr: Throwable) {
+    discard()
+    throw thr
 }
diff --git a/src/test/kotlin/io/lettuce/core/TransactionExtensionsIntegrationTests.kt b/src/test/kotlin/io/lettuce/core/TransactionExtensionsIntegrationTests.kt
index 8c80fe5533..5b8ef91e12 100644
--- a/src/test/kotlin/io/lettuce/core/TransactionExtensionsIntegrationTests.kt
+++ b/src/test/kotlin/io/lettuce/core/TransactionExtensionsIntegrationTests.kt
@@ -19,6 +19,7 @@ import io.lettuce.core.api.StatefulRedisConnection
 import io.lettuce.core.api.async.multi
 import io.lettuce.core.api.sync.multi
 import io.lettuce.test.LettuceExtension
+import kotlinx.coroutines.future.await
 import kotlinx.coroutines.runBlocking
 import org.assertj.core.api.Assertions.assertThat
 import org.junit.jupiter.api.Test
@@ -62,4 +63,22 @@ class TransactionExtensionsIntegrationTests : TestSupport() {
         }
     }
 
+    @Test
+    @Inject
+    internal fun shouldDiscardMultiClosureOverAsync(connection: StatefulRedisConnection<String, String>) {
+
+        runBlocking {
+            val transactionResult = runCatching {
+                connection.async().multi {
+                    set("key", "value")
+                    throw RedisCommandExecutionException("oops")
+                }
+            }
+
+            assertThat(transactionResult.isFailure).isTrue()
+            assertThat(transactionResult.exceptionOrNull()).isInstanceOf(RedisCommandExecutionException::class.java)
+            assertThat(connection.async().get("key").await()).isNull()
+        }
+    }
+
 }