Breaking: Get rid of atomic cancellation and provide a replacement (#1937)

This is a problematic for Android when Main dispatcher is cancelled on destroyed activity.
Atomic nature of channels is designed to prevent loss of elements,
which is really not an issue for a typical application, but creates problem when used with channels.

* Internal suspendAtomicCancellableCoroutine -> suspendCancellableCoroutine
* Internal suspendAtomicCancellableCoroutineReusable -> suspendCancellableCoroutineReusable
* Remove atomic cancellation from docs
* Ensures that flowOn does not resume downstream after cancellation.
* MODE_ATOMIC_DEFAULT renamed into MODE_ATOMIC
* Introduced MODE_CANCELLABLE_REUSABLE to track suspendCancellableCoroutineReusable
* Better documentation for MODE_XXX constants.
* Added stress test for proper handling of MODE_CANCELLABLE_REUSABLE
  and fixed test for #1123 bug with job.join (working in MODE_CANCELLABLE) that was not
  properly failing in the absence of the proper code in CancellableContinuationImpl.getResult
* Added test for Flow.combine that should be fixed
* Support extended invokeOnCancellation contract
* Introduced internal tryResumeAtomic
* Channel onUnderliveredElement is introduced as a replacement.

Fixes #1265
Fixes #1813
Fixes #1915
Fixes #1936

Co-authored-by: Louis CAD <louis.cognault@gmail.com>
Co-authored-by: Vsevolod Tolstopyatov <qwwdfsad@gmail.com>

1 files changed, 10 insertions, 10 deletions

diff --git a/kotlinx-coroutines-core/common/test/channels/BasicOperationsTest.kt b/kotlinx-coroutines-core/common/test/channels/BasicOperationsTest.kt
index a6ddd811..91d941b3 100644
--- a/kotlinx-coroutines-core/common/test/channels/BasicOperationsTest.kt
+++ b/kotlinx-coroutines-core/common/test/channels/BasicOperationsTest.kt
@@ -42,7 +42,7 @@ class BasicOperationsTest : TestBase() {
     @Test
     fun testInvokeOnClose() = TestChannelKind.values().forEach { kind ->
         reset()
-        val channel = kind.create()
+        val channel = kind.create<Int>()
         channel.invokeOnClose {
             if (it is AssertionError) {
                 expect(3)
@@ -59,7 +59,7 @@ class BasicOperationsTest : TestBase() {
     fun testInvokeOnClosed() = TestChannelKind.values().forEach { kind ->
         reset()
         expect(1)
-        val channel = kind.create()
+        val channel = kind.create<Int>()
         channel.close()
         channel.invokeOnClose { expect(2) }
         assertFailsWith<IllegalStateException> { channel.invokeOnClose { expect(3) } }
@@ -69,7 +69,7 @@ class BasicOperationsTest : TestBase() {
     @Test
     fun testMultipleInvokeOnClose() = TestChannelKind.values().forEach { kind ->
         reset()
-        val channel = kind.create()
+        val channel = kind.create<Int>()
         channel.invokeOnClose { expect(3) }
         expect(1)
         assertFailsWith<IllegalStateException> { channel.invokeOnClose { expect(4) } }
@@ -81,7 +81,7 @@ class BasicOperationsTest : TestBase() {
     @Test
     fun testIterator() = runTest {
         TestChannelKind.values().forEach { kind ->
-            val channel = kind.create()
+            val channel = kind.create<Int>()
             val iterator = channel.iterator()
             assertFailsWith<IllegalStateException> { iterator.next() }
             channel.close()
@@ -91,7 +91,7 @@ class BasicOperationsTest : TestBase() {
     }
 
     private suspend fun testReceiveOrNull(kind: TestChannelKind) = coroutineScope {
-        val channel = kind.create()
+        val channel = kind.create<Int>()
         val d = async(NonCancellable) {
             channel.receive()
         }
@@ -108,7 +108,7 @@ class BasicOperationsTest : TestBase() {
     }
 
     private suspend fun testReceiveOrNullException(kind: TestChannelKind) = coroutineScope {
-        val channel = kind.create()
+        val channel = kind.create<Int>()
         val d = async(NonCancellable) {
             channel.receive()
         }
@@ -132,7 +132,7 @@ class BasicOperationsTest : TestBase() {
     @Suppress("ReplaceAssertBooleanWithAssertEquality")
     private suspend fun testReceiveOrClosed(kind: TestChannelKind) = coroutineScope {
         reset()
-        val channel = kind.create()
+        val channel = kind.create<Int>()
         launch {
             expect(2)
             channel.send(1)
@@ -159,7 +159,7 @@ class BasicOperationsTest : TestBase() {
     }
 
     private suspend fun testOffer(kind: TestChannelKind) = coroutineScope {
-        val channel = kind.create()
+        val channel = kind.create<Int>()
         val d = async { channel.send(42) }
         yield()
         channel.close()
@@ -184,7 +184,7 @@ class BasicOperationsTest : TestBase() {
     private suspend fun testSendAfterClose(kind: TestChannelKind) {
         assertFailsWith<ClosedSendChannelException> {
             coroutineScope {
-                val channel = kind.create()
+                val channel = kind.create<Int>()
                 channel.close()
 
                 launch {
@@ -195,7 +195,7 @@ class BasicOperationsTest : TestBase() {
     }
 
     private suspend fun testSendReceive(kind: TestChannelKind, iterations: Int) = coroutineScope {
-        val channel = kind.create()
+        val channel = kind.create<Int>()
         launch {
             repeat(iterations) { channel.send(it) }
             channel.close()