Rework how writing lines works, again.

To preserve the order of lines we offer them into a channel
without suspending. If we launch a new coroutine there's no
guarantee they'll execute in order, even if the method they
call is fair.

CHANGELOG

@@ -2,6 +2,7 @@ vNEXT (in development)
 
  * Added CtcpReceived and ActionReceived events
  * Added sendCtcp and sendAction message builders
+ * Fix issue with messages being sent out of order, which sometimes caused problems connecting to passworded servers
 
 v0.3.1
 

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

@@ -6,6 +6,7 @@ import com.dmdirc.ktirc.messages.*
 import com.dmdirc.ktirc.model.*
 import com.dmdirc.ktirc.util.currentTimeProvider
 import kotlinx.coroutines.*
+import kotlinx.coroutines.channels.Channel
 import kotlinx.coroutines.channels.map
 import java.util.concurrent.atomic.AtomicBoolean
 import java.util.logging.Level
@@ -98,12 +99,10 @@ class IrcClientImpl(private val server: Server, private val profile: Profile) :
     private val connecting = AtomicBoolean(false)
 
     private var connectionJob: Job? = null
+    internal var writeChannel: Channel<ByteArray>? = null
 
     override fun send(message: String) {
-        // TODO: What happens if sending fails?
-        scope.launch {
-            socket?.sendLine(message)
-        }
+        writeChannel?.offer(message.toByteArray())
     }
 
     override fun connect() {
@@ -112,7 +111,14 @@ class IrcClientImpl(private val server: Server, private val profile: Profile) :
             with(socketFactory(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 { writeLines(this@with) }
+                }
+
                 emitEvent(ServerConnected(currentTimeProvider()))
                 sendCapabilityList()
                 sendPasswordIfPresent()

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

@@ -15,7 +15,6 @@ import kotlinx.coroutines.channels.ReceiveChannel
 import kotlinx.coroutines.channels.produce
 import kotlinx.coroutines.io.ByteReadChannel
 import kotlinx.coroutines.io.ByteWriteChannel
-import kotlinx.coroutines.sync.Mutex
 import java.net.InetSocketAddress
 import java.security.SecureRandom
 import javax.net.ssl.X509TrustManager
@@ -25,10 +24,8 @@ internal interface LineBufferedSocket {
     suspend fun connect()
     fun disconnect()
 
-    suspend fun sendLine(line: ByteArray, offset: Int = 0, length: Int = line.size)
-    suspend fun sendLine(line: String)
-
     fun readLines(coroutineScope: CoroutineScope): ReceiveChannel<ByteArray>
+    suspend fun writeLines(channel: ReceiveChannel<ByteArray>)
 
 }
 
@@ -46,7 +43,6 @@ internal class KtorLineBufferedSocket(private val host: String, private val port
     var tlsTrustManager: X509TrustManager? = null
 
     private val log by logger()
-    private val writeLock = Mutex()
 
     private lateinit var socket: Socket
     private lateinit var readChannel: ByteReadChannel
@@ -69,23 +65,6 @@ internal class KtorLineBufferedSocket(private val host: String, private val port
         socket.close()
     }
 
-    override suspend fun sendLine(line: ByteArray, offset: Int, length: Int) {
-        writeLock.lock()
-        try {
-            with(writeChannel) {
-                log.fine { ">>> ${String(line, offset, length)}" }
-                writeAvailable(line, offset, length)
-                writeByte(CARRIAGE_RETURN)
-                writeByte(LINE_FEED)
-                flush()
-            }
-        } finally {
-            writeLock.unlock()
-        }
-    }
-
-    override suspend fun sendLine(line: String) = sendLine(line.toByteArray())
-
     @ExperimentalCoroutinesApi
     override fun readLines(coroutineScope: CoroutineScope) = coroutineScope.produce {
         val lineBuffer = ByteArray(4096)
@@ -107,4 +86,16 @@ internal class KtorLineBufferedSocket(private val host: String, private val port
             index = count + index - start
         }
     }
+
+    override suspend fun writeLines(channel: ReceiveChannel<ByteArray>) {
+        for (line in channel) {
+            with(writeChannel) {
+                log.fine { ">>> ${String(line)}" }
+                writeAvailable(line, 0, line.size)
+                writeByte(CARRIAGE_RETURN)
+                writeByte(LINE_FEED)
+                flush()
+            }
+        }
+    }
 }

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

@@ -11,11 +11,8 @@ import com.dmdirc.ktirc.model.ServerFeature
 import com.dmdirc.ktirc.model.User
 import com.dmdirc.ktirc.util.currentTimeProvider
 import com.nhaarman.mockitokotlin2.*
-import kotlinx.coroutines.GlobalScope
+import kotlinx.coroutines.*
 import kotlinx.coroutines.channels.Channel
-import kotlinx.coroutines.launch
-import kotlinx.coroutines.runBlocking
-import kotlinx.coroutines.withTimeoutOrNull
 import org.junit.jupiter.api.Assertions.*
 import org.junit.jupiter.api.BeforeEach
 import org.junit.jupiter.api.Test
@@ -97,11 +94,11 @@ internal class IrcClientImplTest {
         client.socketFactory = mockSocketFactory
         client.connect()
 
-        with(inOrder(mockSocket).verify(mockSocket, timeout(500))) {
-            sendLine("CAP LS 302")
-            sendLine("NICK :$NICK")
-            sendLine("USER $USER_NAME localhost $HOST :$REAL_NAME")
-        }
+        client.blockUntilConnected()
+
+        assertEquals("CAP LS 302", String(client.writeChannel!!.receive()))
+        assertEquals("NICK :$NICK", String(client.writeChannel!!.receive()))
+        assertEquals("USER $USER_NAME localhost $HOST :$REAL_NAME", String(client.writeChannel!!.receive()))
     }
 
     @Test
@@ -110,11 +107,10 @@ internal class IrcClientImplTest {
         client.socketFactory = mockSocketFactory
         client.connect()
 
-        with(inOrder(mockSocket).verify(mockSocket, timeout(500))) {
-            sendLine("CAP LS 302")
-            sendLine("PASS :$PASSWORD")
-            sendLine("NICK :$NICK")
-        }
+        client.blockUntilConnected()
+
+        assertEquals("CAP LS 302", String(client.writeChannel!!.receive()))
+        assertEquals("PASS :$PASSWORD", String(client.writeChannel!!.receive()))
     }
 
     @Test
@@ -199,12 +195,20 @@ internal class IrcClientImplTest {
         client.socketFactory = mockSocketFactory
         client.connect()
 
-        // Wait for it to connect
-        verify(mockSocket, timeout(500)).sendLine("CAP LS 302")
+        client.blockUntilConnected()
 
         client.send("testing 123")
 
-        verify(mockSocket, timeout(500)).sendLine("testing 123")
+        assertEquals(true, withTimeoutOrNull(500) {
+            var found = false
+            for (line in client.writeChannel!!) {
+                if (String(line) == "testing 123") {
+                    found = true
+                    break
+                }
+            }
+            found
+        })
     }
 
     @Test
@@ -213,13 +217,44 @@ internal class IrcClientImplTest {
         client.socketFactory = mockSocketFactory
         client.connect()
 
-        // Wait for it to connect
-        verify(mockSocket, timeout(500)).sendLine("CAP LS 302")
+        client.blockUntilConnected()
 
         client.disconnect()
 
         verify(mockSocket, timeout(500)).disconnect()
     }
 
+    @Test
+    fun `IrcClientImpl sends messages in order`() = runBlocking {
+        val client = IrcClientImpl(Server(HOST, PORT), Profile(NICK, REAL_NAME, USER_NAME))
+        client.socketFactory = mockSocketFactory
+        client.connect()
+
+        client.blockUntilConnected()
+
+        (0..100).forEach { client.send("TEST $it") }
+
+        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
+                    }
+                }
+            }
+            next
+        })
+    }
+
+    private suspend fun IrcClientImpl.blockUntilConnected() {
+        // Yuck. Maybe connect should be asynchronous?
+        while (writeChannel == null) {
+            delay(50)
+        }
+    }
+
 
 }

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

@@ -3,16 +3,17 @@ 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 org.junit.jupiter.api.Assertions.*
+import org.junit.jupiter.api.Assertions.assertEquals
+import org.junit.jupiter.api.Assertions.assertNotNull
 import org.junit.jupiter.api.Test
 import org.junit.jupiter.api.parallel.Execution
 import org.junit.jupiter.api.parallel.ExecutionMode
 import java.io.File
 import java.net.ServerSocket
 import java.security.KeyStore
-import java.security.cert.CertificateException
 import java.security.cert.X509Certificate
 import javax.net.ssl.KeyManagerFactory
 import javax.net.ssl.SSLContext
@@ -21,6 +22,8 @@ import javax.net.ssl.X509TrustManager
 @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 ->
@@ -34,23 +37,7 @@ internal class KtorLineBufferedSocketTest {
     }
 
     @Test
-    fun `KtorLineBufferedSocket throws trying to connect to a server with a bad TLS cert`() = runBlocking {
-        tlsServerSocket(12321).use { serverSocket ->
-            try {
-                val socket = KtorLineBufferedSocket("localhost", 12321)
-                val clientSocketAsync = GlobalScope.async { serverSocket.accept() }
-
-                socket.connect()
-                assertNotNull(clientSocketAsync.await())
-                fail<Unit>()
-            } catch (ex : CertificateException) {
-                // Expected
-            }
-        }
-    }
-
-    @Test
-    fun `KtorLineBufferedSocket can send a whole byte array to a server`() = runBlocking {
+    fun `KtorLineBufferedSocket can send a byte array to a server`() = runBlocking {
         ServerSocket(12321).use { serverSocket ->
             val socket = KtorLineBufferedSocket("localhost", 12321)
             val clientBytesAsync = GlobalScope.async {
@@ -60,26 +47,8 @@ internal class KtorLineBufferedSocketTest {
             }
 
             socket.connect()
-            socket.sendLine("Hello World".toByteArray())
-
-            val bytes = clientBytesAsync.await()
-            assertNotNull(bytes)
-            assertEquals("Hello World\r\n", String(bytes))
-        }
-    }
-
-    @Test
-    fun `KtorLineBufferedSocket can send a string to a server`() = runBlocking {
-        ServerSocket(12321).use { serverSocket ->
-            val socket = KtorLineBufferedSocket("localhost", 12321)
-            val clientBytesAsync = GlobalScope.async {
-                ByteArray(13).apply {
-                    serverSocket.accept().getInputStream().read(this)
-                }
-            }
-
-            socket.connect()
-            socket.sendLine("Hello World")
+            GlobalScope.launch { socket.writeLines(writeChannel) }
+            writeChannel.send("Hello World".toByteArray())
 
             val bytes = clientBytesAsync.await()
             assertNotNull(bytes)
@@ -99,7 +68,8 @@ internal class KtorLineBufferedSocketTest {
             }
 
             socket.connect()
-            socket.sendLine("Hello World")
+            GlobalScope.launch { socket.writeLines(writeChannel) }
+            writeChannel.send("Hello World".toByteArray())
 
             val bytes = clientBytesAsync.await()
             assertNotNull(bytes)
@@ -107,25 +77,6 @@ internal class KtorLineBufferedSocketTest {
         }
     }
 
-    @Test
-    fun `KtorLineBufferedSocket can send a partial byte array to a server`() = runBlocking {
-        ServerSocket(12321).use { serverSocket ->
-            val socket = KtorLineBufferedSocket("localhost", 12321)
-            val clientBytesAsync = GlobalScope.async {
-                ByteArray(7).apply {
-                    serverSocket.accept().getInputStream().read(this)
-                }
-            }
-
-            socket.connect()
-            socket.sendLine("Hello World".toByteArray(), 6, 5)
-
-            val bytes = clientBytesAsync.await()
-            assertNotNull(bytes)
-            assertEquals("World\r\n", String(bytes))
-        }
-    }
-
     @Test
     fun `KtorLineBufferedSocket can receive a line of CRLF delimited text`() = runBlocking {
         ServerSocket(12321).use { serverSocket ->