Main-safe Kotlin coroutine implementation of TCP client based on Okio

SocketTcpClient.kt

import android.util.Log
import kotlinx.coroutines.CoroutineScope
import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.cancel
import kotlinx.coroutines.withContext
import net.monetizemyapp.toolbox.extentions.toHexString
import okio.*
import java.net.Socket
import javax.net.ssl.SSLSocketFactory

interface TcpClient {

    val isConnected: Boolean
    val isShutdown: Boolean

    @Throws(IOException::class)
    suspend fun connect()

    @Throws(IOException::class)
    fun shutdown()

    @Throws(IOException::class)
    suspend fun sendString(string: String)

    @Throws(IOException::class)
    suspend fun sendBytes(bytes: ByteArray)

    @Throws(IOException::class)
    suspend fun readString(): String?

    @Throws(IOException::class)
    suspend fun readBytes(count: Long): ByteArray

    @Throws(IOException::class)
    suspend fun readByte(): Byte

}


@Suppress("BlockingMethodInNonBlockingContext")
class SocketTcpClient(
    private val host: String,
    private val port: Int,
    private val isSsl: Boolean = true
) : TcpClient {

    private val coroutineScope = CoroutineScope(Dispatchers.IO)

    private var socket: Socket? = null
    private var source: BufferedSource? = null
    private var sink: BufferedSink? = null

    override var isShutdown: Boolean = false
        private set


    override suspend fun connect() = withThisContext {
        log("connect")
        if (isConnected) return@withThisContext

        socket = if (isSsl) SSLSocketFactory.getDefault().createSocket(host, port) else Socket(host, port)
        source = socket!!.source().buffer()
        sink = socket!!.sink().buffer()
    }

    override fun shutdown() {
        if (isShutdown) return
        isShutdown = true

        socket?.apply {
            close()
            source = null
            sink = null
        }
        coroutineScope.cancel()
        log("shutdown")
    }

    override val isConnected: Boolean
        get() = socket?.isConnected == true

    override suspend fun sendString(string: String) = withThisContext {
        sink!!.writeUtf8(string).flush().also { log("string sent: $string") }
        Unit
    }

    override suspend fun sendBytes(bytes: ByteArray) = withThisContext {
        sink!!.write(bytes).flush().also { log("bytes sent: ${bytes.toHexString()}") }
        Unit
    }

    override suspend fun readString(): String? = withThisContext {
        source!!.readUtf8Line().also { log("string read: $it") }
    }

    override suspend fun readBytes(count: Long): ByteArray = withThisContext {
        source!!.readByteArray(count).also { log("bytes read: ${it.toHexString()}") }
    }

    override suspend fun readByte(): Byte = withThisContext {
        source!!.readByte().also { log("byte read: ${it.toHexString()}") }
    }

    private suspend fun <T> withThisContext(block: suspend CoroutineScope.() -> T) =
        withContext(coroutineScope.coroutineContext, block)

    private fun log(msg: String) {
        Log.d("SocketTcpClient", "[${Thread.currentThread().name}] $msg")
    }

    private fun logError(msg: String, throwable: Throwable? = null) {
        Log.e("SocketTcpClient", "[${Thread.currentThread().name}] $msg", throwable)
    }

}