Await comply with reactive streams Subscriber rule 2.7

reactive/kotlinx-coroutines-reactive/src/Await.kt

@@ -198,12 +198,22 @@ private suspend fun <T> Publisher<T>.awaitOne(
             /** cancelling the new subscription due to rule 2.5, though the publisher would either have to
              * subscribe more than once, which would break 2.12, or leak this [Subscriber]. */
             if (subscription != null) {
-                sub.cancel()
+                withSubscriptionLock {
+                    sub.cancel()
+                }
                 return
             }
             subscription = sub
-            cont.invokeOnCancellation { sub.cancel() }
-            sub.request(if (mode == Mode.FIRST || mode == Mode.FIRST_OR_DEFAULT) 1 else Long.MAX_VALUE)
+
+            cont.invokeOnCancellation {
+                withSubscriptionLock {
+                    sub.cancel()
+                }
+            }
+
+            withSubscriptionLock {
+                sub.request(if (mode == Mode.FIRST || mode == Mode.FIRST_OR_DEFAULT) 1 else Long.MAX_VALUE)
+            }
         }
 
         override fun onNext(t: T) {
@@ -228,12 +238,16 @@ private suspend fun <T> Publisher<T>.awaitOne(
                         return
                     }
                     seenValue = true
-                    sub.cancel()
+                    withSubscriptionLock {
+                        sub.cancel()
+                    }
                     cont.resume(t)
                 }
                 Mode.LAST, Mode.SINGLE, Mode.SINGLE_OR_DEFAULT -> {
                     if ((mode == Mode.SINGLE || mode == Mode.SINGLE_OR_DEFAULT) && seenValue) {
-                        sub.cancel()
+                        withSubscriptionLock {
+                            sub.cancel()
+                        }
                         /* the check for `cont.isActive` is needed in case `sub.cancel() above calls `onComplete` or
                          `onError` on its own. */
                         if (cont.isActive) {
@@ -289,6 +303,14 @@ private suspend fun <T> Publisher<T>.awaitOne(
             inTerminalState = true
             return true
         }
+
+        /**
+         * Enforce rule 2.7: [Subscription.request] and [Subscription.cancel] must be executed serially
+         */
+        @Synchronized
+        private fun withSubscriptionLock(block: () -> Unit) {
+            block()
+        }
     })
 }
 

reactive/kotlinx-coroutines-reactive/test/AwaitTest.kt

@@ -5,6 +5,7 @@
 package kotlinx.coroutines.reactive
 
 import kotlinx.coroutines.*
+import kotlinx.coroutines.sync.*
 import org.junit.*
 import org.reactivestreams.*
 
@@ -40,4 +41,43 @@ class AwaitTest: TestBase() {
         finish(7)
     }
 
-}
+    @Test
+    fun testAwaitCancellationPerformedSerially() = runTest {
+        val requestCompletion = Mutex(locked = true)
+        val subscriptionStarted = Mutex(locked = true)
+        expect(1)
+        val publisher = Publisher<Int> { s ->
+            s.onSubscribe(object : Subscription {
+                override fun request(n: Long) {
+                    expect(2)
+                    subscriptionStarted.unlock()
+                    runBlocking { requestCompletion.lock() }
+                    expect(4)
+                }
+
+                override fun cancel() {
+                    expect(5)
+                }
+            })
+        }
+        val job = launch(Dispatchers.Default) {
+            try {
+                publisher.awaitFirst()
+            } catch (e: CancellationException) {
+                expect(6)
+                throw e
+            }
+        }
+        subscriptionStarted.lock()
+        expect(3)
+
+        launch(Dispatchers.Default) {
+            job.cancel()
+        }
+        delay(10)
+        requestCompletion.unlock()
+        job.join()
+
+        finish(7)
+    }
+}