Improve couroutines and channel use in the socket

CHANGELOG

     * Added sendTagMessage() to send message tags without any content
     * The reply() utility automatically marks messages as a reply
     * Added react() utility to send a reaction client tag
+ * (Internal) improved how coroutines and channels are used in LineBufferedSocket
 
 v0.5.0
 

src/itest/kotlin/com/dmdirc/ktirc/KtIrcIntegrationTest.kt

 import com.dmdirc.irctest.IrcLibraryTests
 import com.dmdirc.ktirc.model.Profile
 import com.dmdirc.ktirc.model.Server
-import kotlinx.coroutines.delay
 import kotlinx.coroutines.runBlocking
 import org.junit.jupiter.api.TestFactory
 
 
         override fun terminate() {
             runBlocking {
-                delay(100)
                 ircClient.disconnect()
-                ircClient.join()
             }
         }
 

src/main/kotlin/com/dmdirc/ktirc/IrcClient.kt

 import com.dmdirc.ktirc.messages.*
 import com.dmdirc.ktirc.model.*
 import com.dmdirc.ktirc.util.currentTimeProvider
+import io.ktor.util.KtorExperimentalAPI
 import kotlinx.coroutines.*
-import kotlinx.coroutines.channels.Channel
 import kotlinx.coroutines.channels.map
 import java.util.concurrent.atomic.AtomicBoolean
-import java.util.logging.Level
-import java.util.logging.LogManager
 
 /**
  * Primary interface for interacting with KtIrc.
 // TODO: How should alternative nicknames work?
 // TODO: Should IRC Client take a pool of servers and rotate through, or make the caller do that?
 // TODO: Should there be a default profile?
-class IrcClientImpl(private val server: Server, override val profile: Profile) : IrcClient {
+@KtorExperimentalAPI
+@ExperimentalCoroutinesApi
+class IrcClientImpl(private val server: Server, override val profile: Profile) : IrcClient, CoroutineScope {
 
-    internal var socketFactory: (String, Int, Boolean) -> LineBufferedSocket = ::KtorLineBufferedSocket
+    override val coroutineContext = GlobalScope.newCoroutineContext(Dispatchers.IO)
+
+    internal var socketFactory: (CoroutineScope, String, Int, Boolean) -> LineBufferedSocket = ::KtorLineBufferedSocket
 
     override val serverState = ServerState(profile.initialNick, server.host)
     override val channelState = ChannelStateMap { caseMapping }
     private val parser = MessageParser()
     private var socket: LineBufferedSocket? = null
 
-    private val scope = CoroutineScope(Dispatchers.IO)
     private val connecting = AtomicBoolean(false)
 
-    private var connectionJob: Job? = null
-    internal var writeChannel: Channel<ByteArray>? = null
-
     override fun send(message: String) {
-        writeChannel?.offer(message.toByteArray())
+        socket?.sendChannel?.offer(message.toByteArray())
     }
 
     override fun connect() {
         check(!connecting.getAndSet(true))
-        connectionJob = scope.launch {
-            with(socketFactory(server.host, server.port, server.tls)) {
-                // TODO: Proper error handling - what if connect() fails?
-                socket = this
 
-                emitEvent(ServerConnecting(currentTimeProvider()))
+        with(socketFactory(this, server.host, server.port, server.tls)) {
+            // TODO: Proper error handling - what if connect() fails?
+            socket = this
 
-                connect()
-
-                with(Channel<ByteArray>(Channel.UNLIMITED)) {
-                    writeChannel = this
-                    scope.launch {
-                        writeChannel?.let {
-                            writeLines(it)
-                        }
-                    }
-                }
+            emitEvent(ServerConnecting(currentTimeProvider()))
 
+            launch {
+                connect()
                 emitEvent(ServerConnected(currentTimeProvider()))
                 sendCapabilityList()
                 sendPasswordIfPresent()
                 sendNickChange(profile.initialNick)
                 // TODO: Send correct host
                 sendUser(profile.userName, profile.realName)
-                messageHandler.processMessages(this@IrcClientImpl, readLines(scope).map { parser.parse(it) })
+                messageHandler.processMessages(this@IrcClientImpl, receiveChannel.map { parser.parse(it) })
                 emitEvent(ServerDisconnected(currentTimeProvider()))
             }
         }
         socket?.disconnect()
     }
 
-    /**
-     * Joins the coroutine running the message loop, and blocks until it is completed.
-     */
-    suspend fun join() {
-        connectionJob?.join()
-    }
-
     override fun onEvent(handler: (IrcEvent) -> Unit) {
         messageHandler.handlers.add(object : EventHandler {
             override fun processEvent(client: IrcClient, event: IrcEvent): List<IrcEvent> {

src/main/kotlin/com/dmdirc/ktirc/io/LineBufferedSocket.kt

 import io.ktor.network.sockets.openReadChannel
 import io.ktor.network.sockets.openWriteChannel
 import io.ktor.network.tls.tls
-import kotlinx.coroutines.CoroutineScope
-import kotlinx.coroutines.Dispatchers
-import kotlinx.coroutines.ExperimentalCoroutinesApi
-import kotlinx.coroutines.GlobalScope
+import io.ktor.util.KtorExperimentalAPI
+import kotlinx.coroutines.*
+import kotlinx.coroutines.channels.Channel
 import kotlinx.coroutines.channels.ReceiveChannel
+import kotlinx.coroutines.channels.SendChannel
 import kotlinx.coroutines.channels.produce
 import kotlinx.coroutines.io.ByteReadChannel
 import kotlinx.coroutines.io.ByteWriteChannel
 
 internal interface LineBufferedSocket {
 
-    suspend fun connect()
+    fun connect()
     fun disconnect()
 
-    fun readLines(coroutineScope: CoroutineScope): ReceiveChannel<ByteArray>
-    suspend fun writeLines(channel: ReceiveChannel<ByteArray>)
+    val sendChannel: SendChannel<ByteArray>
+    val receiveChannel: ReceiveChannel<ByteArray>
 
 }
 
  * Asynchronous socket that buffers incoming data and emits individual lines.
  */
 // TODO: Expose advanced TLS options
-internal class KtorLineBufferedSocket(private val host: String, private val port: Int, private val tls: Boolean = false): LineBufferedSocket {
+@KtorExperimentalAPI
+@ExperimentalCoroutinesApi
+internal class KtorLineBufferedSocket(coroutineScope: CoroutineScope, private val host: String, private val port: Int, private val tls: Boolean = false) : CoroutineScope, LineBufferedSocket {
 
     companion object {
         const val CARRIAGE_RETURN = '\r'.toByte()
         const val LINE_FEED = '\n'.toByte()
     }
 
+    override val coroutineContext = coroutineScope.newCoroutineContext(Dispatchers.IO)
+    override val sendChannel: Channel<ByteArray> = Channel(Channel.UNLIMITED)
+
     var tlsTrustManager: X509TrustManager? = null
 
     private val log by logger()
     private lateinit var readChannel: ByteReadChannel
     private lateinit var writeChannel: ByteWriteChannel
 
-    @Suppress("EXPERIMENTAL_API_USAGE")
-    override suspend fun connect() {
-        log.info { "Connecting..." }
-        socket = aSocket(ActorSelectorManager(Dispatchers.IO)).tcp().connect(InetSocketAddress(host, port))
-        if (tls) {
-            // TODO: Figure out how exactly scopes work...
-            socket = socket.tls(GlobalScope.coroutineContext, randomAlgorithm = SecureRandom.getInstanceStrong().algorithm, trustManager = tlsTrustManager)
+    override fun connect() {
+        runBlocking {
+            log.info { "Connecting..." }
+            socket = aSocket(ActorSelectorManager(Dispatchers.IO)).tcp().connect(InetSocketAddress(host, port))
+            if (tls) {
+                socket = socket.tls(
+                        coroutineContext = this@KtorLineBufferedSocket.coroutineContext,
+                        randomAlgorithm = SecureRandom.getInstanceStrong().algorithm,
+                        trustManager = tlsTrustManager)
+            }
+            readChannel = socket.openReadChannel()
+            writeChannel = socket.openWriteChannel()
         }
-        readChannel = socket.openReadChannel()
-        writeChannel = socket.openWriteChannel()
+        launch { writeLines() }
     }
 
     override fun disconnect() {
         log.info { "Disconnecting..." }
         socket.close()
+        coroutineContext.cancel()
     }
 
-    @ExperimentalCoroutinesApi
-    override fun readLines(coroutineScope: CoroutineScope) = coroutineScope.produce {
-        val lineBuffer = ByteArray(4096)
-        var index = 0
-        while (!readChannel.isClosedForRead) {
-            var start = index
-            val count = readChannel.readAvailable(lineBuffer, index, lineBuffer.size - index)
-            for (i in index until index + count) {
-                if (lineBuffer[i] == CARRIAGE_RETURN || lineBuffer[i] == LINE_FEED) {
-                    if (start < i) {
-                        val line = lineBuffer.sliceArray(start until i)
-                        log.fine { "<<< ${String(line)}" }
-                        send(line)
+    override val receiveChannel
+        get() = produce {
+            val lineBuffer = ByteArray(4096)
+            var index = 0
+            while (!readChannel.isClosedForRead) {
+                var start = index
+                val count = readChannel.readAvailable(lineBuffer, index, lineBuffer.size - index)
+                for (i in index until index + count) {
+                    if (lineBuffer[i] == CARRIAGE_RETURN || lineBuffer[i] == LINE_FEED) {
+                        if (start < i) {
+                            val line = lineBuffer.sliceArray(start until i)
+                            log.fine { "<<< ${String(line)}" }
+                            send(line)
+                        }
+                        start = i + 1
                     }
-                    start = i + 1
                 }
+                lineBuffer.copyInto(lineBuffer, 0, start)
+                index = count + index - start
             }
-            lineBuffer.copyInto(lineBuffer, 0, start)
-            index = count + index - start
         }
-    }
 
-    override suspend fun writeLines(channel: ReceiveChannel<ByteArray>) {
-        for (line in channel) {
+    private suspend fun writeLines() {
+        for (line in sendChannel) {
             with(writeChannel) {
                 log.fine { ">>> ${String(line)}" }
                 writeAvailable(line, 0, line.size)

src/test/kotlin/com/dmdirc/ktirc/IrcClientTest.kt

 import com.dmdirc.ktirc.events.*
 import com.dmdirc.ktirc.io.CaseMapping
 import com.dmdirc.ktirc.io.LineBufferedSocket
-import com.dmdirc.ktirc.model.Profile
-import com.dmdirc.ktirc.model.Server
-import com.dmdirc.ktirc.model.ServerFeature
-import com.dmdirc.ktirc.model.User
+import com.dmdirc.ktirc.model.*
 import com.dmdirc.ktirc.util.currentTimeProvider
 import com.nhaarman.mockitokotlin2.*
+import io.ktor.util.KtorExperimentalAPI
 import kotlinx.coroutines.*
 import kotlinx.coroutines.channels.Channel
+import kotlinx.coroutines.channels.filter
+import kotlinx.coroutines.channels.map
 import org.junit.jupiter.api.Assertions.*
 import org.junit.jupiter.api.BeforeEach
 import org.junit.jupiter.api.Test
 import org.junit.jupiter.api.assertThrows
 
+@KtorExperimentalAPI
+@ExperimentalCoroutinesApi
 internal class IrcClientImplTest {
 
     companion object {
         private const val PASSWORD = "HackThePlanet"
     }
 
-    private val readLineChannel = Channel<ByteArray>(10)
+    private val readLineChannel = Channel<ByteArray>(Channel.UNLIMITED)
+    private val sendLineChannel = Channel<ByteArray>(Channel.UNLIMITED)
 
     private val mockSocket = mock<LineBufferedSocket> {
-        on { readLines(any()) } doReturn readLineChannel
+        on { receiveChannel } doReturn readLineChannel
+        on { sendChannel } doReturn sendLineChannel
     }
 
-    private val mockSocketFactory = mock<(String, Int, Boolean) -> LineBufferedSocket> {
-        on { invoke(eq(HOST), eq(PORT), any()) } doReturn mockSocket
+    private val mockSocketFactory = mock<(CoroutineScope, String, Int, Boolean) -> LineBufferedSocket> {
+        on { invoke(any(), eq(HOST), eq(PORT), any()) } doReturn mockSocket
     }
 
     private val mockEventHandler = mock<(IrcEvent) -> Unit>()
         client.socketFactory = mockSocketFactory
         client.connect()
 
-        verify(mockSocketFactory, timeout(500)).invoke(HOST, PORT, false)
+        verify(mockSocketFactory, timeout(500)).invoke(client, HOST, PORT, false)
     }
 
     @Test
         client.socketFactory = mockSocketFactory
         client.connect()
 
-        verify(mockSocketFactory, timeout(500)).invoke(HOST, PORT, true)
+        verify(mockSocketFactory, timeout(500)).invoke(client, HOST, PORT, true)
     }
 
     @Test
 
         client.blockUntilConnected()
 
-        assertEquals("CAP LS 302", String(client.writeChannel!!.receive()))
-        assertEquals("NICK :$NICK", String(client.writeChannel!!.receive()))
-        assertEquals("USER $USER_NAME 0 * :$REAL_NAME", String(client.writeChannel!!.receive()))
+        assertEquals("CAP LS 302", String(sendLineChannel.receive()))
+        assertEquals("NICK :$NICK", String(sendLineChannel.receive()))
+        assertEquals("USER $USER_NAME 0 * :$REAL_NAME", String(sendLineChannel.receive()))
     }
 
     @Test
 
         client.blockUntilConnected()
 
-        assertEquals("CAP LS 302", String(client.writeChannel!!.receive()))
-        assertEquals("PASS :$PASSWORD", String(client.writeChannel!!.receive()))
+        assertEquals("CAP LS 302", String(sendLineChannel.receive()))
+        assertEquals("PASS :$PASSWORD", String(sendLineChannel.receive()))
     }
 
     @Test
         assertFalse(client.isLocalUser(User("{acidBurn}", "libby", "root.localhost")))
     }
 
-    @Test
-    fun `IrcClientImpl join blocks when socket is open`() {
-        val client = IrcClientImpl(Server(HOST, PORT, password = PASSWORD), Profile(NICK, REAL_NAME, USER_NAME))
-        client.socketFactory = mockSocketFactory
-
-        GlobalScope.launch {
-            readLineChannel.send(":the.gibson 001 acidBurn :Welcome to the IRC!".toByteArray())
-        }
-
-        client.connect()
-        runBlocking {
-            assertNull(withTimeoutOrNull(100L) {
-                client.join()
-                true
-            })
-        }
-    }
-
-    @Test
-    fun `IrcClientImpl join returns when socket is closed`() {
-        val client = IrcClientImpl(Server(HOST, PORT, password = PASSWORD), Profile(NICK, REAL_NAME, USER_NAME))
-        client.socketFactory = mockSocketFactory
-
-        GlobalScope.launch {
-            readLineChannel.send(":the.gibson 001 acidBurn :Welcome to the IRC!".toByteArray())
-            readLineChannel.close()
-        }
-
-        client.connect()
-        runBlocking {
-            assertEquals(true, withTimeoutOrNull(500L) {
-                client.join()
-                true
-            })
-        }
-    }
-
     @Test
     fun `IrcClientImpl sends text to socket`() = runBlocking {
         val client = IrcClientImpl(Server(HOST, PORT), Profile(NICK, REAL_NAME, USER_NAME))
 
         assertEquals(true, withTimeoutOrNull(500) {
             var found = false
-            for (line in client.writeChannel!!) {
+            for (line in sendLineChannel) {
                 if (String(line) == "testing 123") {
                     found = true
                     break
 
         assertEquals(100, withTimeoutOrNull(500) {
             var next = 0
-            for (line in client.writeChannel!!) {
-                val stringy = String(line)
-                if (stringy.startsWith("TEST ")) {
-                    assertEquals("TEST $next", stringy)
-                    if (++next == 100) {
-                        break
-                    }
+            for (line in sendLineChannel.map { String(it) }.filter { it.startsWith("TEST ") }) {
+                assertEquals("TEST $next", line)
+                if (++next == 100) {
+                    break
                 }
             }
             next
 
     private suspend fun IrcClientImpl.blockUntilConnected() {
         // Yuck. Maybe connect should be asynchronous?
-        while (writeChannel == null) {
+        while (serverState.status <= ServerStatus.Connecting) {
             delay(50)
         }
     }

src/test/kotlin/com/dmdirc/ktirc/io/KtorLineBufferedSocketTest.kt

 package com.dmdirc.ktirc.io
 
 import io.ktor.network.tls.certificates.generateCertificate
-import kotlinx.coroutines.GlobalScope
-import kotlinx.coroutines.async
-import kotlinx.coroutines.channels.Channel
-import kotlinx.coroutines.launch
-import kotlinx.coroutines.runBlocking
+import io.ktor.util.KtorExperimentalAPI
+import kotlinx.coroutines.*
 import org.junit.jupiter.api.Assertions.assertEquals
 import org.junit.jupiter.api.Assertions.assertNotNull
 import org.junit.jupiter.api.Test
 import javax.net.ssl.SSLContext
 import javax.net.ssl.X509TrustManager
 
+@KtorExperimentalAPI
+@ExperimentalCoroutinesApi
 @Execution(ExecutionMode.SAME_THREAD)
 internal class KtorLineBufferedSocketTest {
 
-    private val writeChannel = Channel<ByteArray>(Channel.UNLIMITED)
-
     @Test
     fun `KtorLineBufferedSocket can connect to a server`() = runBlocking {
         ServerSocket(12321).use { serverSocket ->
-            val socket = KtorLineBufferedSocket("localhost", 12321)
+            val socket = KtorLineBufferedSocket(GlobalScope, "localhost", 12321)
             val clientSocketAsync = GlobalScope.async { serverSocket.accept() }
 
             socket.connect()
     @Test
     fun `KtorLineBufferedSocket can send a byte array to a server`() = runBlocking {
         ServerSocket(12321).use { serverSocket ->
-            val socket = KtorLineBufferedSocket("localhost", 12321)
+            val socket = KtorLineBufferedSocket(GlobalScope, "localhost", 12321)
             val clientBytesAsync = GlobalScope.async {
                 ByteArray(13).apply {
                     serverSocket.accept().getInputStream().read(this)
             }
 
             socket.connect()
-            GlobalScope.launch { socket.writeLines(writeChannel) }
-            writeChannel.send("Hello World".toByteArray())
+            socket.sendChannel.send("Hello World".toByteArray())
 
             val bytes = clientBytesAsync.await()
             assertNotNull(bytes)
     @Test
     fun `KtorLineBufferedSocket can send a string to a server over TLS`() = runBlocking {
         tlsServerSocket(12321).use { serverSocket ->
-            val socket = KtorLineBufferedSocket("localhost", 12321, true)
+            val socket = KtorLineBufferedSocket(GlobalScope, "localhost", 12321, true)
             socket.tlsTrustManager = getTrustingManager()
             val clientBytesAsync = GlobalScope.async {
                 ByteArray(13).apply {
             }
 
             socket.connect()
-            GlobalScope.launch { socket.writeLines(writeChannel) }
-            writeChannel.send("Hello World".toByteArray())
+            socket.sendChannel.send("Hello World".toByteArray())
 
             val bytes = clientBytesAsync.await()
             assertNotNull(bytes)
     @Test
     fun `KtorLineBufferedSocket can receive a line of CRLF delimited text`() = runBlocking {
         ServerSocket(12321).use { serverSocket ->
-            val socket = KtorLineBufferedSocket("localhost", 12321)
+            val socket = KtorLineBufferedSocket(GlobalScope, "localhost", 12321)
             GlobalScope.launch {
                 serverSocket.accept().getOutputStream().write("Hi there\r\n".toByteArray())
             }
 
             socket.connect()
-            assertEquals("Hi there", String(socket.readLines(GlobalScope).receive()))
+            assertEquals("Hi there", String(socket.receiveChannel.receive()))
         }
     }
 
     @Test
     fun `KtorLineBufferedSocket can receive a line of LF delimited text`() = runBlocking {
         ServerSocket(12321).use { serverSocket ->
-            val socket = KtorLineBufferedSocket("localhost", 12321)
+            val socket = KtorLineBufferedSocket(GlobalScope, "localhost", 12321)
             GlobalScope.launch {
                 serverSocket.accept().getOutputStream().write("Hi there\n".toByteArray())
             }
 
             socket.connect()
-            assertEquals("Hi there", String(socket.readLines(GlobalScope).receive()))
+            assertEquals("Hi there", String(socket.receiveChannel.receive()))
         }
     }
 
     @Test
     fun `KtorLineBufferedSocket can receive multiple lines of text in one packet`() = runBlocking {
         ServerSocket(12321).use { serverSocket ->
-            val socket = KtorLineBufferedSocket("localhost", 12321)
+            val socket = KtorLineBufferedSocket(GlobalScope, "localhost", 12321)
             GlobalScope.launch {
                 serverSocket.accept().getOutputStream().write("Hi there\nThis is a test\r".toByteArray())
             }
 
             socket.connect()
-            val lineProducer = socket.readLines(GlobalScope)
+            val lineProducer = socket.receiveChannel
             assertEquals("Hi there", String(lineProducer.receive()))
             assertEquals("This is a test", String(lineProducer.receive()))
         }
     @Test
     fun `KtorLineBufferedSocket can receive one line of text over multiple packets`() = runBlocking {
         ServerSocket(12321).use { serverSocket ->
-            val socket = KtorLineBufferedSocket("localhost", 12321)
+            val socket = KtorLineBufferedSocket(GlobalScope, "localhost", 12321)
             GlobalScope.launch {
                 with(serverSocket.accept().getOutputStream()) {
                     write("Hi".toByteArray())
             }
 
             socket.connect()
-            val lineProducer = socket.readLines(GlobalScope)
+            val lineProducer = socket.receiveChannel
             assertEquals("Hi there", String(lineProducer.receive()))
         }
     }
     @Test
     fun `KtorLineBufferedSocket returns from readLines when socket is closed`() = runBlocking {
         ServerSocket(12321).use { serverSocket ->
-            val socket = KtorLineBufferedSocket("localhost", 12321)
+            val socket = KtorLineBufferedSocket(GlobalScope, "localhost", 12321)
             GlobalScope.launch {
                 with(serverSocket.accept()) {
                     getOutputStream().write("Hi there\r\n".toByteArray())
             }
 
             socket.connect()
-            val lineProducer = socket.readLines(GlobalScope)
+            val lineProducer = socket.receiveChannel
             assertEquals("Hi there", String(lineProducer.receive()))
         }
     }
     @Test
     fun `KtorLineBufferedSocket disconnects from server`() = runBlocking {
         ServerSocket(12321).use { serverSocket ->
-            val socket = KtorLineBufferedSocket("localhost", 12321)
+            val socket = KtorLineBufferedSocket(GlobalScope, "localhost", 12321)
             val clientSocketAsync = GlobalScope.async { serverSocket.accept() }
 
             socket.connect()