Fix channel BufferOverflow behavior

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

@@ -140,8 +140,8 @@ internal open class ArrayChannel<E>(
         // buffer is full
         return when (onBufferOverflow) {
             BufferOverflow.SUSPEND -> OFFER_FAILED
-            BufferOverflow.DROP_LATEST -> OFFER_SUCCESS
-            BufferOverflow.DROP_OLDEST -> null // proceed, will drop oldest in enqueueElement
+            BufferOverflow.DROP_LATEST,
+            BufferOverflow.DROP_OLDEST -> null // proceed, will drop oldest/latest in enqueueElement
         }
     }
 
@@ -152,10 +152,16 @@ internal open class ArrayChannel<E>(
             buffer[(head + currentSize) % buffer.size] = element // actually queue element
         } else {
             // buffer is full
-            assert { onBufferOverflow == BufferOverflow.DROP_OLDEST } // the only way we can get here
-            buffer[head % buffer.size] = null // drop oldest element
-            buffer[(head + currentSize) % buffer.size] = element // actually queue element
-            head = (head + 1) % buffer.size
+            if (onBufferOverflow == BufferOverflow.DROP_OLDEST) {
+                @Suppress("UNCHECKED_CAST")
+                onUndeliveredElement?.callUndeliveredElementCatchingException(buffer[head % buffer.size] as E)
+                buffer[head % buffer.size] = null // drop oldest element
+                buffer[(head + currentSize) % buffer.size] = element // actually queue element
+                head = (head + 1) % buffer.size
+            } else {
+                assert { onBufferOverflow == BufferOverflow.DROP_LATEST } // the only way we can get here
+                onUndeliveredElement?.callUndeliveredElementCatchingException(element)
+            }
         }
     }
 

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

@@ -145,6 +145,42 @@ class ArrayChannelTest : TestBase() {
         channel.receive()
     }
 
+    @Test
+    fun testBufferOverflowOldest() = runTest {
+        val capacity = 3
+        val dropped = mutableListOf<Int>()
+        val expectedDropped = listOf(1, 2, 3)
+        val expectedValues = listOf(4, 5, 6)
+
+        val channel = Channel<Int>(capacity, BufferOverflow.DROP_OLDEST) { dropped.add(it) }
+        (1..6).forEach { value ->
+            channel.send(value)
+        }
+
+        assertContentEquals(expectedDropped, dropped)
+        expectedValues.forEach { expected ->
+            assertEquals(expected, channel.receive())
+        }
+    }
+
+    @Test
+    fun testBufferOverflowLatest() = runTest {
+        val capacity = 3
+        val dropped = mutableListOf<Int>()
+        val expectedDropped = listOf(4, 5, 6)
+        val expectedValues = listOf(1, 2, 3)
+
+        val channel = Channel<Int>(capacity, BufferOverflow.DROP_LATEST) { dropped.add(it) }
+        (1..6).forEach { value ->
+            channel.send(value)
+        }
+
+        assertContentEquals(expectedDropped, dropped)
+        expectedValues.forEach { expected ->
+            assertEquals(expected, channel.receive())
+        }
+    }
+
     @Test
     fun testBufferSize() = runTest {
         val capacity = 42