Fix hang in publish(n) when more collect

Author: akarnokd

README.md

@@ -11,7 +11,7 @@ Extensions to the Kotlin Flow library.
 
 ```groovy
 dependencies {
-    implementation "com.github.akarnokd:kotlin-flow-extensions:0.0.6"
+    implementation "com.github.akarnokd:kotlin-flow-extensions:0.0.7"
 }
 ```
 

gradle.properties

@@ -1,5 +1,5 @@
 GROUP=com.github.akarnokd
-VERSION_NAME=0.0.6
+VERSION_NAME=0.0.7
 
 POM_ARTIFACT_ID=kotlin-flow-extensions
 POM_NAME=Kotlin Flow Extensions

src/main/kotlin/hu/akarnokd/kotlin/flow/MulticastSubject.kt

@@ -144,8 +144,17 @@ class MulticastSubject<T>(private val expectedCollectors: Int) : AbstractFlow<T>
     override suspend fun collectSafely(collector: FlowCollector<T>) {
         val rc = ResumableCollector<T>()
         if (add(rc)) {
-            if (remainingCollectors.decrementAndGet() == 0) {
-                producer.resume()
+            while (true) {
+                val a = remainingCollectors.get()
+                if (a == 0) {
+                    break;
+                }
+                if (remainingCollectors.compareAndSet(a, a - 1)) {
+                    if (a == 1) {
+                        producer.resume();
+                    }
+                    break;
+                }
             }
             rc.drain(collector) { remove(it) }
         } else {

src/test/kotlin/hu/akarnokd/kotlin/flow/MulticastSubjectTest.kt

@@ -466,4 +466,28 @@ class MulticastSubjectTest {
         job1.join()
         job2.join()
     }
+
+    @Test(timeout = 5000)
+    @ExperimentalCoroutinesApi
+    fun moreThanExpectedCollectors() = runBlocking {
+        val subject = MulticastSubject<Int>(2)
+        val result = mutableListOf<Int>()
+
+        val job = launch(Dispatchers.IO) {
+            merge(subject, subject, subject)
+            .collect { result.add(it) }
+        }
+
+        // wait for the collector to arrive
+        while (subject.collectorCount() != 3) {
+            delay(1)
+        }
+
+        subject.emit(1)
+        subject.complete()
+
+        job.join()
+
+        assertEquals(listOf(1, 1, 1), result)
+    }
 }