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

/*
 * Copyright 2016-2018 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license.
 */

package kotlinx.coroutines.experimental.channels

import kotlinx.coroutines.experimental.*
import kotlin.test.*

class BasicOperationsTest : TestBase() {

    @Test
    fun testSimpleSendReceive() = runTest {
        // Parametrized common test :(
        TestChannelKind.values().forEach { kind -> testSendReceive(kind, 20) }
    }

    @Test
    fun testOfferAfterClose() = runTest {
        TestChannelKind.values().forEach { kind -> testOffer(kind) }
    }

    @Test
    fun testReceiveOrNullAfterClose() = runTest {
        TestChannelKind.values().forEach { kind -> testReceiveOrNull(kind) }
    }

    @Test
    fun testReceiveOrNullAfterCloseWithException() = runTest {
        TestChannelKind.values().forEach { kind -> testReceiveOrNullException(kind) }
    }

    @Test
    fun testInvokeOnClose() = TestChannelKind.values().forEach { kind ->
        reset()
        val channel = kind.create()
        channel.invokeOnClose {
            if (it is AssertionError) {
                expect(3)
            }
        }
        expect(1)
        channel.offer(42)
        expect(2)
        channel.close(AssertionError())
        finish(4)
    }

    @Test
    fun testInvokeOnClosed() = TestChannelKind.values().forEach { kind ->
        reset()
        expect(1)
        val channel = kind.create()
        channel.close()
        channel.invokeOnClose { expect(2) }
        assertFailsWith<IllegalStateException> { channel.invokeOnClose { expect(3) } }
        finish(3)
    }

    @Test
    fun testMultipleInvokeOnClose() = TestChannelKind.values().forEach { kind ->
        reset()
        val channel = kind.create()
        channel.invokeOnClose { expect(3) }
        expect(1)
        assertFailsWith<IllegalStateException> { channel.invokeOnClose { expect(4) } }
        expect(2)
        channel.close()
        finish(4)
    }

    private suspend fun testReceiveOrNull(kind: TestChannelKind) = coroutineScope {
        val channel = kind.create()
        val d = async {
            channel.receive()
        }

        yield()
        channel.close()
        assertTrue(channel.isClosedForReceive)

        assertNull(channel.receiveOrNull())
        assertNull(channel.poll())

        d.join()
        assertTrue(d.getCancellationException().cause is ClosedReceiveChannelException)
    }

    private suspend fun testReceiveOrNullException(kind: TestChannelKind) = coroutineScope {
        val channel = kind.create()
        val d = async {
            channel.receive()
        }

        yield()
        channel.close(IndexOutOfBoundsException())
        assertTrue(channel.isClosedForReceive)

        assertFailsWith<IndexOutOfBoundsException> { channel.poll() }
        try {
            channel.receiveOrNull()
            fail()
        } catch (e: IndexOutOfBoundsException) {
            // Expected
        }

        d.join()
        assertTrue(d.getCancellationException().cause is IndexOutOfBoundsException)
    }


    private suspend fun testOffer(kind: TestChannelKind) = coroutineScope {
        val channel = kind.create()
        val d = async { channel.send(42) }
        yield()
        channel.close()

        assertTrue(channel.isClosedForSend)
        try {
            channel.offer(2)
            fail()
        } catch (e: ClosedSendChannelException) {
            if (!kind.isConflated) {
                assertEquals(42, channel.receive())
            }
        }

        d.await()
    }

    private suspend fun testSendReceive(kind: TestChannelKind, iterations: Int) = coroutineScope {
        val channel = kind.create()
        launch {
            repeat(iterations) { channel.send(it) }
            channel.close()
        }
        var expected = 0
        for (x in channel) {
            if (!kind.isConflated) {
                assertEquals(expected++, x)
            } else {
                assertTrue(x >= expected)
                expected = x + 1
            }
        }
        if (!kind.isConflated) {
            assertEquals(iterations, expected)
        }
    }
}