Rework coroutines to make more sense.

CHANGELOG

@@ -2,4 +2,5 @@ v0.2.0 [in development]
 
   * Added support for connecting over TLS
   * Simplified how event handlers are registered
+  * Improved use of coroutines so users don't have to worry about them
   * (Internal) Upgraded to Gradle 5.1.1

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

@@ -4,16 +4,16 @@ import com.dmdirc.ktirc.events.*
 import com.dmdirc.ktirc.io.*
 import com.dmdirc.ktirc.messages.*
 import com.dmdirc.ktirc.model.*
+import kotlinx.coroutines.*
 import kotlinx.coroutines.channels.map
-import kotlinx.coroutines.coroutineScope
-import kotlinx.coroutines.runBlocking
+import java.util.concurrent.atomic.AtomicBoolean
 import java.util.logging.Level
 import java.util.logging.LogManager
 
 
 interface IrcClient {
 
-    suspend fun send(message: String)
+    fun send(message: String)
 
     val serverState: ServerState
     val channelState: ChannelStateMap
@@ -44,15 +44,20 @@ class IrcClientImpl(private val server: Server, private val profile: Profile) :
     private val parser = MessageParser()
     private var socket: LineBufferedSocket? = null
 
-    // TODO: It would be cleaner if this didn't suspend but returned immediately
-    override suspend fun send(message: String) {
-        socket?.sendLine(message)
+    private val scope = CoroutineScope(Dispatchers.IO)
+    private val connecting = AtomicBoolean(false)
+
+    private var connectionJob: Job? = null
+
+    override fun send(message: String) {
+        scope.launch {
+            socket?.sendLine(message)
+        }
     }
 
-    suspend fun connect() {
-        // TODO: Concurrency!
-        check(socket == null)
-        coroutineScope {
+    fun connect() {
+        check(!connecting.getAndSet(true))
+        connectionJob = scope.launch {
             with(socketFactory(server.host, server.port, server.tls)) {
                 socket = this
                 connect()
@@ -62,7 +67,7 @@ class IrcClientImpl(private val server: Server, private val profile: Profile) :
                 // TODO: Send correct host
                 sendLine(userMessage(profile.userName, "localhost", server.host, profile.realName))
                 // TODO: This should be elsewhere
-                messageHandler.processMessages(this@IrcClientImpl, readLines(this@coroutineScope).map { parser.parse(it) })
+                messageHandler.processMessages(this@IrcClientImpl, readLines(scope).map { parser.parse(it) })
             }
         }
     }
@@ -71,9 +76,13 @@ class IrcClientImpl(private val server: Server, private val profile: Profile) :
         socket?.disconnect()
     }
 
+    suspend fun join() {
+        connectionJob?.join()
+    }
+
     override fun onEvent(handler: (IrcEvent) -> Unit) {
         messageHandler.handlers.add(object : EventHandler {
-            override suspend fun processEvent(client: IrcClient, event: IrcEvent) {
+            override fun processEvent(client: IrcClient, event: IrcEvent) {
                 handler(event)
             }
         })
@@ -90,13 +99,12 @@ fun main() {
     runBlocking {
         val client = IrcClientImpl(Server("testnet.inspircd.org", 6667), Profile("KtIrc", "Kotlin!", "kotlin"))
         client.onEvent { event ->
-            runBlocking {
-                when (event) {
-                    is ServerWelcome -> client.send(joinMessage("#ktirc"))
-                    is MessageReceived -> if (event.message == "!test") client.send(privmsgMessage(event.target, "Test successful!"))
-                }
+            when (event) {
+                is ServerWelcome -> client.send(joinMessage("#ktirc"))
+                is MessageReceived -> if (event.message == "!test") client.send(privmsgMessage(event.target, "Test successful!"))
             }
         }
         client.connect()
+        client.join()
     }
 }

src/main/kotlin/com/dmdirc/ktirc/events/CapabilitiesHandler.kt

@@ -12,7 +12,7 @@ class CapabilitiesHandler : EventHandler {
 
     private val log by logger()
 
-    override suspend fun processEvent(client: IrcClient, event: IrcEvent) {
+    override fun processEvent(client: IrcClient, event: IrcEvent) {
         when (event) {
             is ServerCapabilitiesReceived -> handleCapabilitiesReceived(client.serverState.capabilities, event.capabilities)
             is ServerCapabilitiesFinished -> handleCapabilitiesFinished(client)
@@ -24,7 +24,7 @@ class CapabilitiesHandler : EventHandler {
         state.advertisedCapabilities.putAll(capabilities)
     }
 
-    private suspend fun handleCapabilitiesFinished(client: IrcClient) {
+    private fun handleCapabilitiesFinished(client: IrcClient) {
         // TODO: We probably need to split the outgoing REQ lines if there are lots of caps
         // TODO: For caps with values we'll need to decide which value to use/whether to enable them/etc
         with (client.serverState.capabilities) {
@@ -41,7 +41,7 @@ class CapabilitiesHandler : EventHandler {
         }
     }
 
-    private suspend fun handleCapabilitiesAcknowledged(client: IrcClient, capabilities: Map<Capability, String>) {
+    private fun handleCapabilitiesAcknowledged(client: IrcClient, capabilities: Map<Capability, String>) {
         // TODO: Check if everything we wanted is enabled
         with (client.serverState.capabilities) {
             log.info { "Acknowledged capabilities: ${capabilities.keys.map { it.name }.toList()}" }

src/main/kotlin/com/dmdirc/ktirc/events/ChannelStateHandler.kt

@@ -9,7 +9,7 @@ class ChannelStateHandler : EventHandler {
 
     private val log by logger()
 
-    override suspend fun processEvent(client: IrcClient, event: IrcEvent) {
+    override fun processEvent(client: IrcClient, event: IrcEvent) {
         when (event) {
             is ChannelJoined -> handleJoin(client, event)
             is ChannelParted -> handlePart(client, event)
@@ -64,6 +64,4 @@ class ChannelStateHandler : EventHandler {
         client.channelState.forEach { it.users -= event.user.nickname }
     }
 
-    private fun String.nickname(prefixLength: Int) = substring(prefixLength).substringBefore('!')
-
 }

src/main/kotlin/com/dmdirc/ktirc/events/EventHandler.kt

@@ -5,7 +5,7 @@ import com.dmdirc.ktirc.IrcClient
 @FunctionalInterface
 interface EventHandler {
 
-    suspend fun processEvent(client: IrcClient, event: IrcEvent)
+    fun processEvent(client: IrcClient, event: IrcEvent)
 
 }
 

src/main/kotlin/com/dmdirc/ktirc/events/PingHandler.kt

@@ -5,7 +5,7 @@ import com.dmdirc.ktirc.messages.pongMessage
 
 class PingHandler : EventHandler {
 
-    override suspend fun processEvent(client: IrcClient, event: IrcEvent) {
+    override fun processEvent(client: IrcClient, event: IrcEvent) {
         when (event) {
             is PingReceived -> client.send(pongMessage(event.nonce))
         }

src/main/kotlin/com/dmdirc/ktirc/events/ServerStateHandler.kt

@@ -4,7 +4,7 @@ import com.dmdirc.ktirc.IrcClient
 
 class ServerStateHandler : EventHandler {
 
-    override suspend fun processEvent(client: IrcClient, event: IrcEvent) {
+    override fun processEvent(client: IrcClient, event: IrcEvent) {
         when (event) {
             is ServerWelcome -> client.serverState.localNickname = event.localNick
             is ServerFeaturesUpdated -> client.serverState.features.setAll(event.serverFeatures)

src/main/kotlin/com/dmdirc/ktirc/events/UserStateHandler.kt

@@ -5,7 +5,7 @@ import com.dmdirc.ktirc.model.UserState
 
 class UserStateHandler : EventHandler {
 
-    override suspend fun processEvent(client: IrcClient, event: IrcEvent) {
+    override fun processEvent(client: IrcClient, event: IrcEvent) {
         when (event) {
             is ChannelJoined -> handleJoin(client.userState, event)
             is ChannelParted -> handlePart(client, event)

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

@@ -9,6 +9,7 @@ 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 kotlinx.coroutines.channels.ReceiveChannel
 import kotlinx.coroutines.channels.produce
@@ -78,6 +79,7 @@ class KtorLineBufferedSocket(private val host: String, private val port: Int, pr
 
     override suspend fun sendLine(line: String) = sendLine(line.toByteArray())
 
+    @ExperimentalCoroutinesApi
     override fun readLines(coroutineScope: CoroutineScope) = coroutineScope.produce {
         val lineBuffer = ByteArray(4096)
         var index = 0

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

@@ -6,9 +6,11 @@ import com.dmdirc.ktirc.io.CaseMapping
 import com.dmdirc.ktirc.io.LineBufferedSocket
 import com.dmdirc.ktirc.model.*
 import com.nhaarman.mockitokotlin2.*
+import kotlinx.coroutines.GlobalScope
 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
@@ -44,97 +46,72 @@ internal class IrcClientImplTest {
 
     @Test
     fun `IrcClientImpl uses socket factory to create a new socket on connect`() {
-        runBlocking {
-            val client = IrcClientImpl(Server(HOST, PORT), Profile(NICK, REAL_NAME, USER_NAME))
-            client.socketFactory = mockSocketFactory
-            readLineChannel.close()
-
-            client.connect()
+        val client = IrcClientImpl(Server(HOST, PORT), Profile(NICK, REAL_NAME, USER_NAME))
+        client.socketFactory = mockSocketFactory
+        client.connect()
 
-            verify(mockSocketFactory).invoke(HOST, PORT, false)
-        }
+        verify(mockSocketFactory, timeout(500)).invoke(HOST, PORT, false)
     }
 
     @Test
     fun `IrcClientImpl uses socket factory to create a new tls on connect`() {
-        runBlocking {
-            val client = IrcClientImpl(Server(HOST, PORT, true), Profile(NICK, REAL_NAME, USER_NAME))
-            client.socketFactory = mockSocketFactory
-            readLineChannel.close()
-
-            client.connect()
+        val client = IrcClientImpl(Server(HOST, PORT, true), Profile(NICK, REAL_NAME, USER_NAME))
+        client.socketFactory = mockSocketFactory
+        client.connect()
 
-            verify(mockSocketFactory).invoke(HOST, PORT, true)
-        }
+        verify(mockSocketFactory, timeout(500)).invoke(HOST, PORT, true)
     }
 
     @Test
     fun `IrcClientImpl throws if socket already exists`() {
-        runBlocking {
-            val client = IrcClientImpl(Server(HOST, PORT), Profile(NICK, REAL_NAME, USER_NAME))
-            client.socketFactory = mockSocketFactory
-            readLineChannel.close()
+        val client = IrcClientImpl(Server(HOST, PORT), Profile(NICK, REAL_NAME, USER_NAME))
+        client.socketFactory = mockSocketFactory
+        client.connect()
 
+        assertThrows<IllegalStateException> {
             client.connect()
-
-            assertThrows<IllegalStateException> {
-                runBlocking {
-                    client.connect()
-                }
-            }
         }
     }
 
     @Test
-    fun `IrcClientImpl sends basic connection strings`() {
-        runBlocking {
-            val client = IrcClientImpl(Server(HOST, PORT), Profile(NICK, REAL_NAME, USER_NAME))
-            client.socketFactory = mockSocketFactory
-            readLineChannel.close()
-
-            client.connect()
+    fun `IrcClientImpl sends basic connection strings`() = runBlocking {
+        val client = IrcClientImpl(Server(HOST, PORT), Profile(NICK, REAL_NAME, USER_NAME))
+        client.socketFactory = mockSocketFactory
+        client.connect()
 
-            with(inOrder(mockSocket).verify(mockSocket)) {
-                sendLine("CAP LS 302")
-                sendLine("NICK :$NICK")
-                sendLine("USER $USER_NAME localhost $HOST :$REAL_NAME")
-            }
+        with(inOrder(mockSocket).verify(mockSocket, timeout(500))) {
+            sendLine("CAP LS 302")
+            sendLine("NICK :$NICK")
+            sendLine("USER $USER_NAME localhost $HOST :$REAL_NAME")
         }
     }
 
     @Test
-    fun `IrcClientImpl sends password first, when present`() {
-        runBlocking {
-            val client = IrcClientImpl(Server(HOST, PORT, password = PASSWORD), Profile(NICK, REAL_NAME, USER_NAME))
-            client.socketFactory = mockSocketFactory
-            readLineChannel.close()
-
-            client.connect()
-
-            with(inOrder(mockSocket).verify(mockSocket)) {
-                sendLine("CAP LS 302")
-                sendLine("PASS :$PASSWORD")
-                sendLine("NICK :$NICK")
-            }
+    fun `IrcClientImpl sends password first, when present`() = runBlocking {
+        val client = IrcClientImpl(Server(HOST, PORT, password = PASSWORD), Profile(NICK, REAL_NAME, USER_NAME))
+        client.socketFactory = mockSocketFactory
+        client.connect()
+
+        with(inOrder(mockSocket).verify(mockSocket, timeout(500))) {
+            sendLine("CAP LS 302")
+            sendLine("PASS :$PASSWORD")
+            sendLine("NICK :$NICK")
         }
     }
 
     @Test
     fun `IrcClientImpl sends events to provided event handler`() {
-        runBlocking {
-            val client = IrcClientImpl(Server(HOST, PORT, password = PASSWORD), Profile(NICK, REAL_NAME, USER_NAME))
-            client.socketFactory = mockSocketFactory
-            client.onEvent(mockEventHandler)
+        val client = IrcClientImpl(Server(HOST, PORT, password = PASSWORD), Profile(NICK, REAL_NAME, USER_NAME))
+        client.socketFactory = mockSocketFactory
+        client.onEvent(mockEventHandler)
 
-            launch {
-                readLineChannel.send(":the.gibson 001 acidBurn :Welcome to the IRC!".toByteArray())
-                readLineChannel.close()
-            }
+        GlobalScope.launch {
+            readLineChannel.send(":the.gibson 001 acidBurn :Welcome to the IRC!".toByteArray())
+        }
 
-            client.connect()
+        client.connect()
 
-            verify(mockEventHandler).invoke(isA<ServerWelcome>())
-        }
+        verify(mockEventHandler, timeout(500)).invoke(isA<ServerWelcome>())
     }
 
     @Test
@@ -161,4 +138,41 @@ internal class IrcClientImplTest {
         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
+            })
+        }
+    }
+
 }