From 844052ab71c7ec3cf08d3e51cf4a4a426ac0ff80 Mon Sep 17 00:00:00 2001
From: akarnokd <akarnokd@gmail.com>
Date: Fri, 29 Oct 2021 08:40:07 +0200
Subject: [PATCH] Fix GroupBy not rethrowing errors from the upstream

---
 README.md                                     |  2 +-
 gradle.properties                             |  2 +-
 .../akarnokd/kotlin/flow/impl/FlowGroupBy.kt  |  1 +
 .../kotlin/flow/impl/FlowGroupByTest.kt       | 28 +++++++++++++++++++
 4 files changed, 31 insertions(+), 2 deletions(-)

diff --git a/README.md b/README.md
index 951d8c4..d4dc05b 100644
--- a/README.md
+++ b/README.md
@@ -11,7 +11,7 @@ Extensions to the Kotlin Flow library.
 
 ```groovy
 dependencies {
-    implementation "com.github.akarnokd:kotlin-flow-extensions:0.0.11"
+    implementation "com.github.akarnokd:kotlin-flow-extensions:0.0.12"
 }
 ```
 
diff --git a/gradle.properties b/gradle.properties
index c3adab7..ce364eb 100644
--- a/gradle.properties
+++ b/gradle.properties
@@ -1,5 +1,5 @@
 GROUP=com.github.akarnokd
-VERSION_NAME=0.0.11
+VERSION_NAME=0.0.12
 
 POM_ARTIFACT_ID=kotlin-flow-extensions
 POM_NAME=Kotlin Flow Extensions
diff --git a/src/main/kotlin/hu/akarnokd/kotlin/flow/impl/FlowGroupBy.kt b/src/main/kotlin/hu/akarnokd/kotlin/flow/impl/FlowGroupBy.kt
index e757547..919ddb3 100644
--- a/src/main/kotlin/hu/akarnokd/kotlin/flow/impl/FlowGroupBy.kt
+++ b/src/main/kotlin/hu/akarnokd/kotlin/flow/impl/FlowGroupBy.kt
@@ -81,6 +81,7 @@ internal class FlowGroupBy<T, K, V>(
             for (group in map.values) {
                 group.error(ex)
             }
+            throw ex
         }
     }
 
diff --git a/src/test/kotlin/hu/akarnokd/kotlin/flow/impl/FlowGroupByTest.kt b/src/test/kotlin/hu/akarnokd/kotlin/flow/impl/FlowGroupByTest.kt
index 63b7328..55426e5 100644
--- a/src/test/kotlin/hu/akarnokd/kotlin/flow/impl/FlowGroupByTest.kt
+++ b/src/test/kotlin/hu/akarnokd/kotlin/flow/impl/FlowGroupByTest.kt
@@ -22,6 +22,7 @@ import kotlinx.coroutines.FlowPreview
 import kotlinx.coroutines.delay
 import kotlinx.coroutines.flow.*
 import kotlinx.coroutines.runBlocking
+import org.junit.Assert.assertThrows
 import org.junit.Ignore
 import org.junit.Test
 
@@ -82,4 +83,31 @@ class FlowGroupByTest {
                 .assertResultSet(1, 2)
     }
 
+    @Test
+    fun mainErrorsNoItems() {
+        assertThrows(IllegalStateException::class.java) {
+            runBlocking {
+                (1..10)
+                        .asFlow()
+                        .map { if(it < 5) throw IllegalStateException("oops") else it }
+                        .groupBy { it % 2 == 0 }
+                        .flatMapMerge { it }
+                        .collect()
+            }
+        }
+    }
+
+    @Test
+    fun mainErrorsSomeItems() {
+        assertThrows(IllegalStateException::class.java) {
+            runBlocking {
+                (1..10)
+                        .asFlow()
+                        .map { if(it > 5) throw IllegalStateException("oops") else it }
+                        .groupBy { it % 2 == 0 }
+                        .flatMapMerge { it }
+                        .collect()
+            }
+        }
+    }
 }
\ No newline at end of file