core/kotlinx-coroutines-io/src/test/kotlin/kotlinx/coroutines/experimental/io/ByteBufferChannelScenarioTest.kt

package kotlinx.coroutines.experimental.io

import kotlinx.coroutines.experimental.TestBase
import kotlinx.coroutines.experimental.channels.ClosedReceiveChannelException
import kotlinx.coroutines.experimental.launch
import kotlinx.coroutines.experimental.runBlocking
import kotlinx.coroutines.experimental.yield
import org.junit.*
import java.io.IOException
import kotlin.test.assertEquals
import kotlin.test.fail

class ByteBufferChannelScenarioTest : TestBase() {
    private val ch = ByteBufferChannel(true)

    @After
    fun finish() {
        ch.close(InterruptedException())
    }

    @Test
    fun testReadBeforeAvailable() {
        expect(1)

        runBlocking {
            launch(coroutineContext) {
                expect(3)

                val bb = ByteBuffer.allocate(10)
                val rc = ch.readAvailable(bb) // should suspend

                expect(5)
                assertEquals(4, rc)

                expect(6)
            }

            expect(2)
            yield()

            expect(4)
            ch.writeInt(0xff) // should resume

            yield()

            finish(7)
        }
    }

    @Test
    fun testReadBeforeAvailable2() {
        expect(1)

        runBlocking {
            launch(coroutineContext) {
                expect(3)

                val bb = ByteBuffer.allocate(4)
                ch.readFully(bb) // should suspend

                expect(5)

                bb.flip()
                assertEquals(4, bb.remaining())

                expect(6)
            }

            expect(2)
            yield()

            expect(4)
            ch.writeInt(0xff) // should resume

            yield()

            finish(7)
        }
    }

    @Test
    fun testReadAfterAvailable() {
        expect(1)

        runBlocking {
            ch.writeInt(0xff) // should resume

            launch(coroutineContext) {
                expect(3)

                val bb = ByteBuffer.allocate(10)
                val rc = ch.readAvailable(bb) // should NOT suspend

                expect(4)
                assertEquals(4, rc)

                expect(5)
            }

            expect(2)
            yield()

            finish(6)
        }
    }

    @Test
    fun testReadAfterAvailable2() {
        expect(1)

        runBlocking {
            ch.writeInt(0xff) // should resume

            launch(coroutineContext) {
                expect(3)

                val bb = ByteBuffer.allocate(4)
                ch.readFully(bb) // should NOT suspend

                expect(4)
                bb.flip()
                assertEquals(4, bb.remaining())

                expect(5)
            }

            expect(2)
            yield()

            finish(6)
        }
    }

    @Test
    fun testReadToEmpty() {
        runBlocking {
            expect(1)

            val rc = ch.readAvailable(ByteBuffer.allocate(0))

            expect(2)

            assertEquals(0, rc)

            finish(3)
        }
    }

    @Test
    fun testReadToEmptyFromFailedChannel() {
        runBlocking {
            expect(1)

            ch.close(IOException())

            try {
                ch.readAvailable(ByteBuffer.allocate(0))
                fail("Should throw exception")
            } catch (expected: IOException) {
            }

            finish(2)
        }
    }

    @Test
    fun testReadToEmptyFromClosedChannel() {
        runBlocking {
            expect(1)

            ch.close()

            val rc = ch.readAvailable(ByteBuffer.allocate(0))

            expect(2)

            assertEquals(-1, rc)

            finish(3)
        }
    }

    @Test
    fun testReadFullyToEmptyFromClosedChannel() {
        runBlocking {
            expect(1)

            ch.close()

            ch.readFully(ByteBuffer.allocate(0))

            finish(2)
        }
    }

    @Test
    fun testReadFullyFromClosedChannel() {
        runBlocking {
            expect(1)

            ch.close()
            try {
                ch.readFully(ByteBuffer.allocate(1))
                fail("Should throw exception")
            } catch (expected: ClosedReceiveChannelException) {
            }

            finish(2)
        }
    }

    @Test
    fun testReadFullyToEmptyFromFailedChannel() {
        runBlocking {
            expect(1)

            ch.close(IOException())

            try {
                ch.readFully(ByteBuffer.allocate(0))
                fail("Should throw exception")
            } catch (expected: IOException) {
            }

            finish(2)
        }
    }

    @Test
    fun testWriteBlock() {
        runBlocking {
            launch(coroutineContext) {
                expect(1)

                ch.write {
                    it.putLong(0x1234567812345678L)
                }

                expect(2)
            }

            yield()
            expect(3)

            assertEquals(0x1234567812345678L, ch.readLong())
            assertEquals(0, ch.availableForRead)

            finish(4)
        }
    }

    @Test
    fun testWriteBlockSuspend() {
        runBlocking {
            launch(coroutineContext) {
                expect(1)

                ch.writeFully(ByteArray(4088))

                expect(2)

                ch.write(8) {
                    it.putLong(0x1234567812345678L)
                }

                expect(4)
            }

            yield()
            expect(3)

            ch.readFully(ByteArray(9))
            yield()
            expect(5)

            ch.readFully(ByteArray(4088 - 9))

            expect(6)

            assertEquals(0x1234567812345678L, ch.readLong())
            assertEquals(0, ch.availableForRead)

            finish(7)
        }
    }

    @Test
    fun testReadBlock() = runBlocking {
        ch.writeLong(0x1234567812345678L)

        ch.read {
            assertEquals(0x1234567812345678L, it.getLong())
        }

        finish(1)
    }

    @Test
    fun testReadBlockSuspend() = runBlocking {
        ch.writeByte(0x12)

        launch(coroutineContext) {
            expect(1)
            ch.read(8) {
                assertEquals(0x1234567812345678L, it.getLong())
            }

            expect(3)
        }

        yield()
        expect(2)

        ch.writeLong(0x3456781234567800L)
        yield()

        expect(4)
        ch.readByte()
        assertEquals(0, ch.availableForRead)

        finish(5)
    }

    @Test
    fun testReadBlockSuspend2() = runBlocking {
        launch(coroutineContext) {
            expect(1)
            ch.read(8) {
                assertEquals(0x1234567812345678L, it.getLong())
            }

            expect(3)
        }

        yield()
        expect(2)

        ch.writeLong(0x1234567812345678L)
        yield()

        expect(4)
        assertEquals(0, ch.availableForRead)

        finish(5)
    }

    @Test
    fun testWriteByteSuspend() = runBlocking {
        launch(coroutineContext) {
            expect(1)
            ch.writeByte(1)
            ch.writeFully(ByteArray(ch.availableForWrite))
            expect(2)
            ch.writeByte(1)
            expect(5)
            ch.close()
        }

        yield()
        expect(3)
        yield()
        expect(4)
        yield()

        ch.readByte()
        yield()

        ch.readRemaining()
        finish(6)
    }

    @Test
    fun testWriteShortSuspend() = runBlocking {
        launch(coroutineContext) {
            expect(1)
            ch.writeByte(1)
            ch.writeFully(ByteArray(ch.availableForWrite))
            expect(2)
            ch.writeShort(1)
            expect(5)
            ch.close()
        }

        yield()
        expect(3)
        yield()
        expect(4)
        yield()

        ch.readShort()
        yield()

        ch.readRemaining()
        finish(6)
    }

    @Test
    fun testWriteIntSuspend() = runBlocking {
        launch(coroutineContext) {
            expect(1)
            ch.writeByte(1)
            ch.writeFully(ByteArray(ch.availableForWrite))
            expect(2)
            ch.writeInt(1)
            expect(5)
            ch.close()
        }

        yield()
        expect(3)
        yield()
        expect(4)
        yield()

        ch.readInt()
        yield()

        ch.readRemaining()
        finish(6)
    }

    @Test
    fun testWriteLongSuspend() = runBlocking {
        launch(coroutineContext) {
            expect(1)
            ch.writeByte(1)
            ch.writeFully(ByteArray(ch.availableForWrite))
            expect(2)
            ch.writeLong(1)
            expect(5)
            ch.close()
        }

        yield()
        expect(3)
        yield()
        expect(4)
        yield()

        ch.readLong()
        yield()

        ch.readRemaining()
        finish(6)
    }

}