<commit_before>package com.fireflysource.net.tcp.aio

import com.fireflysource.common.coroutine.CoroutineDispatchers.singleThread
import com.fireflysource.common.coroutine.launchGlobally
import com.fireflysource.common.lifecycle.AbstractLifeCycle.stopAll
import com.fireflysource.common.sys.Result.discard
import com.fireflysource.net.tcp.launch
import com.fireflysource.net.tcp.onAcceptAsync
import com.fireflysource.net.tcp.startReadingAndAwaitHandshake
import kotlinx.coroutines.future.await
import kotlinx.coroutines.runBlocking
import kotlinx.coroutines.time.delay
import org.junit.jupiter.api.Assertions.assertEquals
import org.junit.jupiter.api.Assertions.assertTrue
import org.junit.jupiter.api.Test
import org.junit.jupiter.params.ParameterizedTest
import org.junit.jupiter.params.provider.Arguments
import org.junit.jupiter.params.provider.Arguments.arguments
import org.junit.jupiter.params.provider.MethodSource
import java.nio.ByteBuffer
import java.time.Duration
import java.util.concurrent.atomic.AtomicInteger
import java.util.stream.Stream
import kotlin.math.roundToLong
import kotlin.system.measureTimeMillis


/**
 * @author <NAME>
 */
@Suppress("BlockingMethodInNonBlockingContext")
class TestAioServerAndClient {

    companion object {
        @JvmStatic
        fun testParametersProvider(): Stream<Arguments> {
            return Stream.of(
                arguments("single", true),
                arguments("array", true),
                arguments("list", true),
                arguments("single", false),
                arguments("array", false),
                arguments("list", false)
            )
        }
    }

    @ParameterizedTest
    @MethodSource("testParametersProvider")
    fun test(bufType: String, enableSecure: Boolean) = runBlocking {
        val host = "localhost"
        val port = 4001
        val maxMsgCount = 10_000
        val connectionNum = 4
        val msgCount = AtomicInteger()
        val tcpConfig = TcpConfig(30, enableSecure)

        AioTcpServer(tcpConfig).onAcceptAsync { connection ->
            println("accept connection. ${connection.id}")
            val inputChannel = connection.startReadingAndAwaitHandshake().inputChannel
            recvLoop@ while (true) {
                val buf = inputChannel.receive()

                readBufLoop@ while (buf.hasRemaining()) {
                    val num = buf.int
                    msgCount.incrementAndGet()

                    val newBuf = ByteBuffer.allocate(4)
                    newBuf.putInt(num).flip()
                    connection.write(newBuf, discard())
                }
            }
        }.listen(host, port)

        val client = AioTcpClient(tcpConfig)
        val time = measureTimeMillis {
            val maxCount = maxMsgCount / connectionNum
            val jobs = (1..connectionNum).map {
                launchGlobally(singleThread) {
                    val connection = client.connect(host, port).await()
                    println("create connection. ${connection.id}")
                    connection.startReading()

                    if (connection.isSecureConnection) {
                        val success = connection.onHandshakeComplete().await()
                        println("client TLS handshake success. $success")
                    }

                    val readingJob = connection.launch {
                        val inputChannel = connection.inputChannel
                        recvLoop@ while (true) {
                            val buf = inputChannel.receive()

                            readBufLoop@ while (buf.hasRemaining()) {
                                val num = buf.int
                                msgCount.incrementAndGet()

                                if (num == maxCount) {
                                    connection.close()
                                    break@recvLoop
                                }
                            }
                        }
                    }

                    when (bufType) {
                        "single" -> {
                            (1..maxCount).forEach { i ->
                                val buf = ByteBuffer.allocate(4)
                                buf.putInt(i).flip()
                                connection.write(buf, discard())
                            }
                        }
                        "array" -> {
                            val bufArray = Array<ByteBuffer>(maxCount) { index ->
                                val buf = ByteBuffer.allocate(4)
                                buf.putInt(index + 1).flip()
                                buf
                            }
                            connection.write(bufArray, 0, bufArray.size, discard())
                        }
                        "list" -> {
                            val bufList = List<ByteBuffer>(maxCount) { index ->
                                val buf = ByteBuffer.allocate(4)
                                buf.putInt(index + 1).flip()
                                buf
                            }
                            connection.write(bufList, 0, bufList.size, discard())
                        }
                    }

                    readingJob.join()

                    assertTrue(connection.readBytes > 0)
                    assertTrue(connection.writtenBytes > 0)
                    assertTrue(connection.lastActiveTime > 0L)
                    assertTrue(connection.lastReadTime > 0L)
                    assertTrue(connection.lastWrittenTime > 0L)
                }
            }

            jobs.forEach { it.join() }

            assertEquals(maxMsgCount * 2, msgCount.get())
        }
        val throughput = maxMsgCount / (time / 1000.00)
        println("success. $time ms, ${throughput.roundToLong()} qps")

        val stopTime = measureTimeMillis {
            stopAll()
        }
        println("stop success. $stopTime")
    }

    @Test
    fun testTimeout() = runBlocking {
        val host = "localhost"
        val port = 4001
        val server = AioTcpServer(TcpConfig(1)).listen(host, port)
        val serverAcceptsConnJob = launchGlobally(singleThread) {
            val tcpConnChannel = server.tcpConnectionChannel
            acceptLoop@ while (true) {
                val conn = tcpConnChannel.receive()
                conn.startReading()
                delay(Duration.ofSeconds(2))
                assertTrue(conn.isClosed)
                break@acceptLoop
            }
        }

        val client = AioTcpClient(TcpConfig(1))
        val conn = client.connect(host, port).await()
        conn.startReading()
        assertEquals(port, conn.remoteAddress.port)
        delay(Duration.ofSeconds(2))
        assertTrue(conn.isClosed)
        assertTrue(conn.duration > 0)
        assertTrue(conn.closeTime > 0)

        val stopTime = measureTimeMillis {
            serverAcceptsConnJob.cancel()
            stopAll()
        }
        println("stop success. $stopTime")
    }
}<commit_msg>[chore]: clean up codes.<commit_after>package com.fireflysource.net.tcp.aio

import com.fireflysource.common.coroutine.CoroutineDispatchers.singleThread
import com.fireflysource.common.coroutine.launchGlobally
import com.fireflysource.common.lifecycle.AbstractLifeCycle.stopAll
import com.fireflysource.common.sys.Result.discard
import com.fireflysource.net.tcp.launch
import com.fireflysource.net.tcp.onAcceptAsync
import com.fireflysource.net.tcp.startReadingAndAwaitHandshake
import kotlinx.coroutines.future.await
import kotlinx.coroutines.runBlocking
import kotlinx.coroutines.time.delay
import org.junit.jupiter.api.Assertions.assertEquals
import org.junit.jupiter.api.Assertions.assertTrue
import org.junit.jupiter.api.Test
import org.junit.jupiter.params.ParameterizedTest
import org.junit.jupiter.params.provider.Arguments
import org.junit.jupiter.params.provider.Arguments.arguments
import org.junit.jupiter.params.provider.MethodSource
import java.nio.ByteBuffer
import java.time.Duration
import java.util.concurrent.atomic.AtomicInteger
import java.util.stream.Stream
import kotlin.math.roundToLong
import kotlin.system.measureTimeMillis


/**
 * @author <NAME>
 */
@Suppress("BlockingMethodInNonBlockingContext")
class TestAioServerAndClient {

    companion object {
        @JvmStatic
        fun testParametersProvider(): Stream<Arguments> {
            return Stream.of(
                arguments("single", true),
                arguments("array", true),
                arguments("list", true),
                arguments("single", false),
                arguments("array", false),
                arguments("list", false)
            )
        }
    }

    @ParameterizedTest
    @MethodSource("testParametersProvider")
    fun test(bufType: String, enableSecure: Boolean) = runBlocking {
        val host = "localhost"
        val port = 4001
        val maxMsgCount = 1_000
        val connectionNum = 4
        val msgCount = AtomicInteger()
        val tcpConfig = TcpConfig(30, enableSecure)

        AioTcpServer(tcpConfig).onAcceptAsync { connection ->
            println("accept connection. ${connection.id}")
            val inputChannel = connection.startReadingAndAwaitHandshake().inputChannel
            recvLoop@ while (true) {
                val buf = inputChannel.receive()

                readBufLoop@ while (buf.hasRemaining()) {
                    val num = buf.int
                    msgCount.incrementAndGet()

                    val newBuf = ByteBuffer.allocate(4)
                    newBuf.putInt(num).flip()
                    connection.write(newBuf, discard())
                }
            }
        }.listen(host, port)

        val client = AioTcpClient(tcpConfig)
        val time = measureTimeMillis {
            val maxCount = maxMsgCount / connectionNum
            val jobs = (1..connectionNum).map {
                launchGlobally(singleThread) {
                    val connection = client.connect(host, port).await()
                    println("create connection. ${connection.id}")
                    connection.startReading()

                    if (connection.isSecureConnection) {
                        val success = connection.onHandshakeComplete().await()
                        println("client TLS handshake success. $success")
                    }

                    val readingJob = connection.launch {
                        val inputChannel = connection.inputChannel
                        recvLoop@ while (true) {
                            val buf = inputChannel.receive()

                            readBufLoop@ while (buf.hasRemaining()) {
                                val num = buf.int
                                msgCount.incrementAndGet()

                                if (num == maxCount) {
                                    connection.close()
                                    break@recvLoop
                                }
                            }
                        }
                    }

                    when (bufType) {
                        "single" -> {
                            (1..maxCount).forEach { i ->
                                val buf = ByteBuffer.allocate(4)
                                buf.putInt(i).flip()
                                connection.write(buf, discard())
                            }
                        }
                        "array" -> {
                            val bufArray = Array<ByteBuffer>(maxCount) { index ->
                                val buf = ByteBuffer.allocate(4)
                                buf.putInt(index + 1).flip()
                                buf
                            }
                            connection.write(bufArray, 0, bufArray.size, discard())
                        }
                        "list" -> {
                            val bufList = List<ByteBuffer>(maxCount) { index ->
                                val buf = ByteBuffer.allocate(4)
                                buf.putInt(index + 1).flip()
                                buf
                            }
                            connection.write(bufList, 0, bufList.size, discard())
                        }
                    }

                    readingJob.join()

                    assertTrue(connection.readBytes > 0)
                    assertTrue(connection.writtenBytes > 0)
                    assertTrue(connection.lastActiveTime > 0L)
                    assertTrue(connection.lastReadTime > 0L)
                    assertTrue(connection.lastWrittenTime > 0L)
                }
            }

            jobs.forEach { it.join() }

            assertEquals(maxMsgCount * 2, msgCount.get())
        }
        val throughput = maxMsgCount / (time / 1000.00)
        println("success. $time ms, ${throughput.roundToLong()} qps")

        val stopTime = measureTimeMillis {
            stopAll()
        }
        println("stop success. $stopTime")
    }

    @Test
    fun testTimeout() = runBlocking {
        val host = "localhost"
        val port = 4001

        AioTcpServer(TcpConfig(1)).onAcceptAsync { conn ->
            conn.startReading()
            delay(Duration.ofSeconds(2))
            assertTrue(conn.isClosed)
        }.listen(host, port)

        val client = AioTcpClient(TcpConfig(1))
        val conn = client.connect(host, port).await()
        conn.startReading()
        assertEquals(port, conn.remoteAddress.port)
        delay(Duration.ofSeconds(2))
        assertTrue(conn.isClosed)
        assertTrue(conn.duration > 0)
        assertTrue(conn.closeTime > 0)

        val stopTime = measureTimeMillis {
            stopAll()
        }
        println("stop success. $stopTime")
    }
}
{'fraction_non_alphanumeric': 0.08817344645058371, 'fraction_numerical': 0.00630846204046117, 'mean_word_length': 1.4475598935226264, 'pattern_counts': {'":': 0, '<': 11, '<?xml version=': 0, '>': 40, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 16, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '262820', 'n_tokens_mistral': 3535, 'n_tokens_neox': 3291, 'n_words': 615}