Adjust behavior of conflated channel (WIP)

Fixes #1235
Fixes #332

Author: elizarov

kotlinx-coroutines-core/common/src/channels/AbstractChannel.kt

@@ -331,16 +331,8 @@ internal abstract class AbstractSendChannel<E> : SendChannel<E> {
             previous as Receive<E> // type assertion
             previous.resumeReceiveClosed(closed)
         }
-
-        onClosedIdempotent(closed)
     }
 
-    /**
-     * Invoked when channel is closed as the last action of [close] invocation.
-     * This method should be idempotent and can be called multiple times.
-     */
-    protected open fun onClosedIdempotent(closed: LockFreeLinkedListNode) {}
-
     /**
      * Retrieves first receiving waiter from the queue or returns closed token.
      * @suppress **This is unstable API and it is subject to change.**

kotlinx-coroutines-core/common/src/channels/ConflatedChannel.kt

@@ -5,7 +5,6 @@
 package kotlinx.coroutines.channels
 
 import kotlinx.coroutines.selects.*
-import kotlinx.coroutines.internal.*
 
 /**
  * Channel that buffers at most one element and conflates all subsequent `send` and `offer` invocations,
@@ -24,31 +23,20 @@ internal open class ConflatedChannel<E> : AbstractChannel<E>() {
     protected final override val isBufferAlwaysFull: Boolean get() = false
     protected final override val isBufferFull: Boolean get() = false
 
-    override fun onClosedIdempotent(closed: LockFreeLinkedListNode) {
-        conflatePreviousSendBuffered(closed)
-    }
-
     // result is always `OFFER_SUCCESS | Closed`
     protected override fun offerInternal(element: E): Any {
         while (true) {
             val result = super.offerInternal(element)
             when {
                 result === OFFER_SUCCESS -> return OFFER_SUCCESS
                 result === OFFER_FAILED -> { // try to buffer
-                    val sendResult = sendConflated(element)
-                    when (sendResult) {
+                    when (val sendResult = sendConflated(element)) {
                         null -> return OFFER_SUCCESS
-                        is Closed<*> -> {
-                            conflatePreviousSendBuffered(sendResult)
-                            return sendResult
-                        }
+                        is Closed<*> -> return sendResult
                     }
                     // otherwise there was receiver in queue, retry super.offerInternal
                 }
-                result is Closed<*> -> {
-                    conflatePreviousSendBuffered(result)
-                    return result
-                }
+                result is Closed<*> -> return result
                 else -> error("Invalid offerInternal result $result")
             }
         }
@@ -64,10 +52,7 @@ internal open class ConflatedChannel<E> : AbstractChannel<E>() {
                 result === ALREADY_SELECTED -> return ALREADY_SELECTED
                 result === OFFER_SUCCESS -> return OFFER_SUCCESS
                 result === OFFER_FAILED -> {} // retry
-                result is Closed<*> -> {
-                    conflatePreviousSendBuffered(result)
-                    return result
-                }
+                result is Closed<*> -> return result
                 else -> error("Invalid result $result")
             }
         }

kotlinx-coroutines-core/common/test/channels/ConflatedChannelTest.kt

@@ -31,7 +31,13 @@ class ConflatedChannelTest : TestBase() {
     fun testConflatedClose() = runTest {
         val q = Channel<Int>(Channel.CONFLATED)
         q.send(1)
-        q.close() // shall conflate sent item and become closed
+        q.close() // shall become closed but do not conflate last sent item yet
+        assertTrue(q.isClosedForSend)
+        assertFalse(q.isClosedForReceive)
+        assertEquals(1, q.receive())
+        // not it is closed for receive, too
+        assertTrue(q.isClosedForSend)
+        assertTrue(q.isClosedForReceive)
         assertNull(q.receiveOrNull())
     }
 

kotlinx-coroutines-core/common/test/flow/channels/ChannelFlowTest.kt

@@ -34,8 +34,10 @@ class ChannelFlowTest : TestBase() {
         val flow = channelFlow(bufferSize = Channel.CONFLATED) {
             assertTrue(offer(1))
             assertTrue(offer(2))
+            assertTrue(offer(3))
+            assertTrue(offer(4))
         }
-        assertEquals(listOf(1), flow.toList())
+        assertEquals(listOf(1, 4), flow.toList()) // two elements in the middle got conflated
     }
 
     @Test