adb

gradle/libs.versions.toml

@@ -12,7 +12,7 @@ packageVersion = "1.0.0"
 #
 gradle = "8.4.0"
 kotlin = "1.9.22"
-compose = "1.6.11"
+compose = "1.7.0-alpha03"
 okhttp3 = "5.0.0-alpha.12"
 materialKolor = "1.7.0"
 #compose-webview = "1.9.20"

src/main/kotlin/eu/mjdev/dadb/Dadb.kt

@@ -0,0 +1,329 @@
+package eu.mjdev.dadb
+
+import eu.mjdev.dadb.*
+import eu.mjdev.dadb.adbserver.AdbServer
+import eu.mjdev.dadb.forwarding.TcpForwardDescriptor
+import eu.mjdev.dadb.forwarding.TcpForwarder
+import eu.mjdev.dadb.helpers.*
+import okio.*
+import java.io.File
+import java.nio.file.Files
+
+interface Dadb : AutoCloseable {
+    @Throws(IOException::class)
+    fun open(destination: String): AdbStream
+
+    fun supportsFeature(feature: String): Boolean
+
+    @Throws(IOException::class)
+    fun shell(command: String): AdbShellResponse {
+        openShell(command).use { stream ->
+            return stream.readAll()
+        }
+    }
+
+    @Throws(IOException::class)
+    fun openShell(command: String = ""): AdbShellStream {
+        val stream = open("shell,v2,raw:$command")
+        return AdbShellStream(stream)
+    }
+
+    @Throws(IOException::class)
+    fun push(src: File, remotePath: String, mode: Int = readMode(src), lastModifiedMs: Long = src.lastModified()) {
+        push(src.source(), remotePath, mode, lastModifiedMs)
+    }
+
+    @Throws(IOException::class)
+    fun push(source: Source, remotePath: String, mode: Int, lastModifiedMs: Long) {
+        openSync().use { stream ->
+            stream.send(source, remotePath, mode, lastModifiedMs)
+        }
+    }
+
+    @Throws(IOException::class)
+    fun pull(dst: File, remotePath: String) {
+        pull(dst.sink(append = false), remotePath)
+    }
+
+    @Throws(IOException::class)
+    fun pull(sink: Sink, remotePath: String) {
+        openSync().use { stream ->
+            stream.recv(sink, remotePath)
+        }
+    }
+
+    @Throws(IOException::class)
+    fun openSync(): AdbSyncStream {
+        val stream = open("sync:")
+        return AdbSyncStream(stream)
+    }
+
+    @Throws(IOException::class)
+    fun install(file: File, vararg options: String) {
+        if (supportsFeature("cmd")) {
+            install(file.source(), file.length(), *options)
+        } else {
+            pmInstall(file, *options)
+        }
+    }
+
+    @Throws(IOException::class)
+    fun install(source: Source, size: Long, vararg options: String) {
+        if (supportsFeature("cmd")) {
+            execCmd("package", "install", "-S", size.toString(), *options).use { stream ->
+                stream.sink.writeAll(source)
+                stream.sink.flush()
+                val response = stream.source.readString(Charsets.UTF_8)
+                if (!response.startsWith("Success")) {
+                    throw IOException("Install failed: $response")
+                }
+            }
+        } else {
+            val tempFile = kotlin.io.path.createTempFile()
+            val fileSink = tempFile.sink().buffer()
+            fileSink.writeAll(source)
+            fileSink.flush()
+            pmInstall(tempFile.toFile(), *options)
+        }
+    }
+
+    private fun pmInstall(file: File, vararg options: String) {
+        val fileName = file.name
+        val remotePath = "/data/local/tmp/$fileName"
+        push(file, remotePath)
+        shell("pm install ${options.joinToString(" ")} \"$remotePath\"")
+    }
+
+    @Throws(IOException::class)
+    fun installMultiple(apks: List<File>, vararg options: String) {
+        // http://aospxref.com/android-12.0.0_r3/xref/packages/modules/adb/client/adb_install.cpp#538
+        if (supportsFeature("cmd")) {
+            val totalLength = apks.map { it.length() }.reduce { acc, l -> acc + l }
+            execCmd("package", "install-create", "-S", totalLength.toString(), *options).use { createStream ->
+                val response = createStream.source.readString(Charsets.UTF_8)
+                if (!response.startsWith("Success")) {
+                    throw IOException("connect error for create: $response")
+                }
+                val pattern = """\[(\w+)]""".toRegex()
+                val sessionId =
+                    pattern.find(response)?.groups?.get(1)?.value ?: throw IOException("failed to create session")
+
+                var error: String? = null
+                apks.forEach { apk ->
+                    // install write every apk file to stream
+                    execCmd(
+                        "package",
+                        "install-write",
+                        "-S",
+                        apk.length().toString(),
+                        sessionId,
+                        apk.name,
+                        "-",
+                        *options
+                    ).use { writeStream ->
+                        writeStream.sink.writeAll(apk.source())
+                        writeStream.sink.flush()
+
+                        val writeResponse = writeStream.source.readString(Charsets.UTF_8)
+                        if (!writeResponse.startsWith("Success")) {
+                            error = writeResponse
+                            return@forEach
+                        }
+                    }
+                }
+
+                // commit the session
+                val finalCommand = if (error == null) "install-commit" else "install-abandon"
+                execCmd("package", finalCommand, sessionId, *options).use { commitStream ->
+                    val finalResponse = commitStream.source.readString(Charsets.UTF_8)
+                    if (!finalResponse.startsWith("Success")) {
+                        throw IOException("failed to finalize session: $commitStream")
+                    }
+                }
+
+                if (error != null) {
+                    throw IOException("Install failed: $error")
+                }
+            }
+        } else {
+            val totalLength = apks.map { it.length() }.reduce { acc, l -> acc + l }
+            // step1: create session
+            val response = shell("pm install-create -S $totalLength ${options.joinToString(" ")}")
+            if (!response.allOutput.startsWith("Success")) {
+                throw IOException("pm create session failed: $response")
+            }
+
+            val pattern = """\[(\w+)]""".toRegex()
+            val sessionId =
+                pattern.find(response.allOutput)?.groups?.get(1)?.value ?: throw IOException("failed to create session")
+            var error: String? = null
+
+            val fileNames = apks.map { it.name }
+            val remotePaths = fileNames.map { "/data/local/tmp/$it" }
+
+            // step2: write apk to the session
+            apks.zip(remotePaths).forEachIndexed { index, pair ->
+                val apk = pair.first
+                val remotePath = pair.second
+
+                try {
+                    // we should push the apk files to device, when push failed, it would stop the installation
+                    push(apk, remotePath)
+                } catch (t: IOException) {
+                    error = t.message
+                    return@forEachIndexed
+                }
+
+                // pm install-write -S APK_SIZE SESSION_ID INDEX PATH
+                val writeResponse = shell("pm install-write -S ${apk.length()} $sessionId $index $remotePath")
+                if (!writeResponse.allOutput.startsWith("Success")) {
+                    error = writeResponse.allOutput
+                    return@forEachIndexed
+                }
+            }
+
+            // step3: commit or abandon the session
+            val finalCommand = if (error == null) "pm install-commit $sessionId" else "pm install-abandon $sessionId"
+            val finalResponse = shell(finalCommand)
+            if (!finalResponse.allOutput.startsWith("Success")) {
+                throw IOException("failed to finalize session: $finalResponse")
+            }
+            if (error != null) {
+                throw IOException("Install failed: $error");
+            }
+        }
+    }
+
+    @Throws(IOException::class)
+    fun uninstall(packageName: String) {
+        val response = shell("cmd package uninstall $packageName")
+        if (response.exitCode != 0) {
+            throw IOException("Uninstall failed: ${response.allOutput}")
+        }
+    }
+
+    @Throws(IOException::class)
+    fun execCmd(vararg command: String): AdbStream {
+        if (!supportsFeature("cmd")) throw UnsupportedOperationException("cmd is not supported on this version of Android")
+        val destination = (listOf("exec:cmd") + command).joinToString(" ")
+        return open(destination)
+    }
+
+    @Throws(IOException::class)
+    fun abbExec(vararg command: String): AdbStream {
+        if (!supportsFeature("abb_exec")) throw UnsupportedOperationException("abb_exec is not supported on this version of Android")
+        val destination = "abb_exec:${command.joinToString("\u0000")}"
+        return open(destination)
+    }
+
+    @Throws(IOException::class)
+    fun root() {
+        val response = restartAdb(this, "root:")
+        if (!response.startsWith("restarting") && !response.contains("already")) {
+            throw IOException("Failed to restart adb as root: $response")
+        }
+        waitRootOrClose(this, root = true)
+    }
+
+    @Throws(IOException::class)
+    fun unroot() {
+        val response = restartAdb(this, "unroot:")
+        if (!response.startsWith("restarting") && !response.contains("not running as root")) {
+            throw IOException("Failed to restart adb as root: $response")
+        }
+        waitRootOrClose(this, root = false)
+    }
+
+    @Throws(InterruptedException::class)
+    fun tcpForward(targetPort: Int, hostPort: Int): TcpForwardDescriptor {
+        val forwarder = TcpForwarder(this, targetPort, hostPort)
+        val localPort = forwarder.start()
+
+        return TcpForwardDescriptor(forwarder, localPort)
+    }
+
+    @Throws(InterruptedException::class)
+    fun tcpForward(targetPort: Int): TcpForwardDescriptor {
+        val forwarder = TcpForwarder(this, targetPort)
+        val localPort = forwarder.start()
+
+        return TcpForwardDescriptor(forwarder, localPort)
+    }
+
+    companion object {
+        private const val MIN_EMULATOR_PORT = 5555
+        private const val MAX_EMULATOR_PORT = 5683
+
+        @JvmStatic
+        @JvmOverloads
+        fun create(
+            host: String,
+            port: Int,
+            keyPair: AdbKeyPair? = AdbKeyPair.readDefault(),
+            connectTimeout: Int = 0,
+            socketTimeout: Int = 0
+        ): Dadb = DadbImpl(host, port, keyPair, connectTimeout, socketTimeout)
+
+        @JvmStatic
+        @JvmOverloads
+        fun discover(
+            host: String = "localhost",
+            keyPair: AdbKeyPair? = AdbKeyPair.readDefault()
+        ): List<Dadb> {
+            return list(host, keyPair)
+        }
+
+        @JvmStatic
+        @JvmOverloads
+        fun list(host: String = "localhost", keyPair: AdbKeyPair? = AdbKeyPair.readDefault()): List<Dadb> {
+            val dadbs = AdbServer.listDadbs(adbServerHost = host)
+            if (dadbs.isNotEmpty()) return dadbs
+
+            return (MIN_EMULATOR_PORT..MAX_EMULATOR_PORT).mapNotNull { port ->
+                val dadb = create(host, port, keyPair)
+                val response = try {
+                    dadb.shell("echo success").allOutput
+                } catch (ignore: Throwable) {
+                    null
+                }
+                if (response == "success\n") {
+                    dadb
+                } else {
+                    null
+                }
+            }
+        }
+
+        private fun waitRootOrClose(dadb: Dadb, root: Boolean) {
+            while (true) {
+                try {
+                    val response = dadb.shell("getprop service.adb.root")
+                    val propValue = if (root) 1 else 0
+                    if (response.output == "$propValue\n") return
+                } catch (e: IOException) {
+                    return
+                }
+            }
+        }
+
+        private fun restartAdb(dadb: Dadb, destination: String): String {
+            dadb.open(destination).use { stream ->
+                return stream.source.readUntil('\n'.code.toByte()).readString(Charsets.UTF_8)
+            }
+        }
+
+        private fun BufferedSource.readUntil(endByte: Byte): Buffer {
+            val buffer = Buffer()
+            while (true) {
+                val b = readByte()
+                buffer.writeByte(b.toInt())
+                if (b == endByte) return buffer
+            }
+        }
+
+        private fun readMode(file: File): Int {
+            return Files.getAttribute(file.toPath(), "unix:mode") as? Int
+                ?: throw RuntimeException("Unable to read file mode")
+        }
+    }
+}

src/main/kotlin/eu/mjdev/dadb/DadbDiscover.kt

@@ -0,0 +1,44 @@
+package eu.mjdev.dadb
+
+import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.delay
+import kotlinx.coroutines.launch
+
+@Suppress("MemberVisibilityCanBePrivate")
+class AdbDiscover(
+    coroutineScope: CoroutineScope? = null,
+    discoverDelay: Long = 1000L,
+    onAdded: suspend (Dadb) -> Unit,
+    onRemoved: suspend (Dadb) -> Unit,
+) {
+    val devices = mutableMapOf<String, Dadb>()
+
+    init {
+        coroutineScope?.launch {
+            while (true) {
+                val newDevices = Dadb.discover()
+                if (newDevices.isNotEmpty()) {
+                    newDevices.forEach { entry ->
+                        if (!devices.containsKey(entry.toString())) {
+                            devices[entry.toString()] = entry
+                            onAdded(entry)
+                        }
+                    }
+                }
+                delay(discoverDelay)
+            }
+        }
+    }
+
+    companion object {
+        fun adbDevicesHandler(
+            coroutineScope: CoroutineScope? = null,
+            onRemoved: suspend (Dadb) -> Unit = {},
+            onAdded: suspend (Dadb) -> Unit = {},
+        ) = AdbDiscover(
+            coroutineScope = coroutineScope,
+            onAdded = onAdded,
+            onRemoved = onRemoved
+        )
+    }
+}

src/main/kotlin/eu/mjdev/dadb/adbserver/AdbBinary.kt

@@ -0,0 +1,82 @@
+package eu.mjdev.dadb.adbserver
+
+import java.io.File
+import java.io.IOException
+import java.net.Socket
+import java.util.*
+
+internal object AdbBinary {
+    private val ADB_BINARY: File? by lazy { find() }
+
+    fun tryStartServer(adbServerHost: String, adbServerPort: Int): Boolean {
+        return try {
+            ensureServerRunning(adbServerHost, adbServerPort)
+            true
+        } catch (ignore: Exception) {
+            false
+        }
+    }
+
+    fun ensureServerRunning(adbServerHost: String, adbServerPort: Int) {
+        if (!isServerRunning(adbServerHost, adbServerPort)) {
+            if (adbServerHost != "localhost") {
+                throw IOException("No running adb server found at $adbServerHost:$adbServerPort.")
+            }
+            val adbBinary = ADB_BINARY ?: throw IOException(
+                "No running adb server found at $adbServerHost:$adbServerPort and unable to discover an adb binary."
+            )
+            startServer(adbBinary, adbServerPort)
+            // Immediately after starting the adb server, emulators show as offline.
+            // This is a hack to work around this behavior.
+            Thread.sleep(200)
+        }
+    }
+
+    private fun startServer(adbBinary: File, adbServerPort: Int) {
+        val process = ProcessBuilder(adbBinary.absolutePath, "-P", adbServerPort.toString(), "start-server")
+            .redirectErrorStream(true)
+            .start()
+        val exitCode = process.waitFor()
+        if (exitCode != 0) {
+            val output = process.inputStream.bufferedReader().readText()
+            throw IOException("Failed to start adb server on port $adbServerPort: $output")
+        }
+    }
+
+    private fun isServerRunning(adbServerHost: String, adbServerPort: Int): Boolean {
+        return try {
+            Socket(adbServerHost, adbServerPort).close()
+            true
+        } catch (e: Exception) {
+            false
+        }
+    }
+
+    private fun find(): File? {
+        return findViaWhich() ?: findViaAndroidHome()
+    }
+
+    private fun findViaWhich(): File? {
+        val which = if (isWindows()) "where" else "which"
+        val process = ProcessBuilder(which, "adb").start()
+        if (process.waitFor() != 0) return null
+        val output = process.inputStream.bufferedReader().use { r ->
+            r.readLine().trim()
+        }
+        val file = File(output)
+        if (!file.exists()) return null
+        return file
+    }
+
+    private fun findViaAndroidHome(): File? {
+        val androidEnvHome = System.getenv("ANDROID_HOME") ?: System.getenv("ANDROID_SDK_ROOT") ?: return null
+        val adbName = if (isWindows()) "adb.exe" else "adb"
+        val adbFile = File(androidEnvHome).resolve("platform-tools").resolve(adbName)
+        if (!adbFile.exists()) return null
+        return adbFile
+    }
+
+    private fun isWindows(): Boolean {
+        return System.getProperty("os.name")?.lowercase(Locale.ENGLISH)?.contains("win") == true
+    }
+}

src/main/kotlin/eu/mjdev/dadb/adbserver/AdbServer.kt

@@ -0,0 +1,159 @@
+package eu.mjdev.dadb.adbserver
+
+import eu.mjdev.dadb.Dadb
+import eu.mjdev.dadb.helpers.AdbStream
+import okio.buffer
+import okio.sink
+import okio.source
+import java.io.DataInputStream
+import java.io.DataOutputStream
+import java.io.IOException
+import java.io.OutputStreamWriter
+import java.net.Socket
+import java.nio.charset.StandardCharsets
+
+object AdbServer {
+    /**
+     * Experimental API
+     *
+     * Possible deviceQuery values:
+     *
+     * host:transport:<serial-number>
+     *     Ask to switch the connection to the device/emulator identified by
+     *     <serial-number>. After the OKAY response, every client request will
+     *     be sent directly to the adbd daemon running on the device.
+     *     (Used to implement the -s option)
+     *
+     * host:transport-usb
+     *     Ask to switch the connection to one device connected through USB
+     *     to the host machine. This will fail if there are more than one such
+     *     devices. (Used to implement the -d convenience option)
+     *
+     * host:transport-local
+     *     Ask to switch the connection to one emulator connected through TCP.
+     *     This will fail if there is more than one such emulator instance
+     *     running. (Used to implement the -e convenience option)
+     *
+     * host:transport-any
+     *     Another host:transport variant. Ask to switch the connection to
+     *     either the device or emulator connect to/running on the host.
+     *     Will fail if there is more than one such device/emulator available.
+     *     (Used when neither -s, -d or -e are provided)
+     */
+    @JvmStatic
+    @JvmOverloads
+    fun createDadb(
+        adbServerHost: String = "localhost",
+        adbServerPort: Int = 5037,
+        deviceQuery: String = "host:transport-any"
+    ): Dadb {
+        val name = deviceQuery
+            .removePrefix("host:") // Use the device query without the host: prefix
+            .removePrefix("transport:") // If it's a serial-number, just show that
+        return AdbServerDadb(adbServerHost, adbServerPort, deviceQuery, name)
+    }
+
+    /**
+     * Returns a list of serial numbers of connected devices.
+     */
+    @JvmStatic
+    @JvmOverloads
+    fun listDadbs(
+        adbServerHost: String = "localhost",
+        adbServerPort: Int = 5037,
+    ): List<Dadb> {
+        if (!AdbBinary.tryStartServer(adbServerHost, adbServerPort)) {
+            return emptyList()
+        }
+        val output = Socket(adbServerHost, adbServerPort).use { socket ->
+            send(socket, "host:devices")
+            readString(DataInputStream(socket.getInputStream()))
+        }
+        return output.lines()
+            .filter { it.isNotBlank() }
+            .mapNotNull {
+                val parts = it.split("\t")
+                if (parts.size != 2) {
+                    null
+                } else {
+                    parts[0]
+                }
+            }
+            .map { createDadb(adbServerHost, adbServerPort, "host:transport:${it}") }
+    }
+
+    internal fun readString(inputStream: DataInputStream): String {
+        val encodedLength = readString(inputStream, 4)
+        val length = encodedLength.toInt(16)
+        return readString(inputStream, length)
+    }
+
+    internal fun send(socket: Socket, command: String) {
+        val inputStream = DataInputStream(socket.getInputStream())
+        val outputStream = DataOutputStream(socket.getOutputStream())
+
+        writeString(outputStream, command)
+
+        val response = readString(inputStream, 4)
+        if (response != "OKAY") {
+            val error = readString(inputStream)
+            throw IOException("Command failed ($command): $error")
+        }
+    }
+
+    private fun writeString(outputStream: DataOutputStream, string: String) {
+        OutputStreamWriter(outputStream, StandardCharsets.UTF_8).apply {
+            write(String.format("%04x", string.toByteArray().size))
+            write(string)
+            flush()
+        }
+    }
+
+    private fun readString(inputStream: DataInputStream, length: Int): String {
+        val responseBuffer = ByteArray(length)
+        inputStream.readFully(responseBuffer)
+        return String(responseBuffer, StandardCharsets.UTF_8)
+    }
+}
+
+private class AdbServerDadb constructor(
+    private val host: String,
+    private val port: Int,
+    private val deviceQuery: String,
+    private val name: String,
+) : Dadb {
+
+    private val supportedFeatures: Set<String>
+
+    init {
+        supportedFeatures = open("host:features").use {
+            val features = AdbServer.readString(DataInputStream(it.source.inputStream()))
+            features.split(",").toSet()
+        }
+    }
+
+    override fun open(destination: String): AdbStream {
+        AdbBinary.ensureServerRunning(host, port)
+        val socket = Socket(host, port)
+        AdbServer.send(socket, deviceQuery)
+        AdbServer.send(socket, destination)
+        return object : AdbStream {
+
+            override val source = socket.source().buffer()
+
+            override val sink = socket.sink().buffer()
+
+            override fun close() = socket.close()
+        }
+    }
+
+    override fun supportsFeature(feature: String): Boolean {
+        return feature in supportedFeatures
+    }
+
+    override fun close() {}
+
+    override fun toString(): String {
+        return name
+    }
+}

src/main/kotlin/eu/mjdev/dadb/forwarding/TcpForwarder.kt

@@ -0,0 +1,167 @@
+package eu.mjdev.dadb.forwarding
+
+import eu.mjdev.dadb.Dadb
+import eu.mjdev.dadb.helpers.log
+import okio.*
+import java.io.IOException
+import java.io.InterruptedIOException
+import java.net.ServerSocket
+import java.net.SocketException
+import java.util.concurrent.ExecutorService
+import java.util.concurrent.Executors
+import java.util.concurrent.TimeUnit
+import java.util.concurrent.TimeoutException
+import kotlin.concurrent.thread
+
+data class TcpForwardDescriptor(private val resource: AutoCloseable, val localPort: Int) : AutoCloseable {
+    override fun close() {
+        resource.close()
+    }
+}
+
+internal class TcpForwarder(
+    private val dadb: Dadb,
+    private val targetPort: Int,
+    private val hostPort: Int? = null,
+) : AutoCloseable {
+
+    private var state: State = State.STOPPED
+    private var serverThread: Thread? = null
+    private var server: ServerSocket? = null
+    private var clientExecutor: ExecutorService? = null
+
+    fun start(): Int {
+        check(state == State.STOPPED) { "Forwarder is already started at port $hostPort" }
+
+        moveToState(State.STARTING)
+
+        clientExecutor = Executors.newCachedThreadPool()
+        serverThread = thread {
+            try {
+                handleForwarding()
+            } catch (ignored: SocketException) {
+                // Do nothing
+            } catch (e: IOException) {
+                log { "could not start TCP port forwarding: ${e.message}" }
+            } finally {
+                moveToState(State.STOPPED)
+            }
+        }
+
+        waitFor(10, 5000) {
+            state == State.STARTED
+        }
+
+        return server!!.localPort
+    }
+
+    private fun handleForwarding() {
+        val serverRef = ServerSocket(hostPort ?: 0)
+        server = serverRef
+
+        moveToState(State.STARTED)
+
+        while (!Thread.interrupted()) {
+            val client = serverRef.accept()
+
+            clientExecutor?.execute {
+                val adbStream = dadb.open("tcp:$targetPort")
+
+                val readerThread = thread {
+                    forward(
+                        client.getInputStream().source(),
+                        adbStream.sink
+                    )
+                }
+
+                try {
+                    forward(
+                        adbStream.source,
+                        client.sink().buffer()
+                    )
+                } finally {
+                    adbStream.close()
+                    client.close()
+
+                    readerThread.interrupt()
+                }
+            }
+        }
+    }
+
+    override fun close() {
+        if (state == State.STOPPED || state == State.STOPPING) {
+            return
+        }
+
+        // Make sure that we are not stopping the server while it is in a transient state
+        // to avoid surprises
+        waitFor(10, 5000) {
+            state == State.STARTED
+        }
+
+        moveToState(State.STOPPING)
+
+        server?.close()
+        server = null
+        serverThread?.interrupt()
+        serverThread = null
+        clientExecutor?.shutdown()
+        clientExecutor?.awaitTermination(5, TimeUnit.SECONDS)
+        clientExecutor = null
+
+        waitFor(10, 5000) {
+            state == State.STOPPED
+        }
+    }
+
+    private fun forward(source: Source, sink: BufferedSink) {
+        try {
+            while (!Thread.interrupted()) {
+                try {
+                    if (source.read(sink.buffer, 256) >= 0) {
+                        sink.flush()
+                    } else {
+                        return
+                    }
+                } catch (ignored: IOException) {
+                    // Do nothing
+                }
+            }
+        } catch (ignored: InterruptedException) {
+            // Do nothing
+        } catch (ignored: InterruptedIOException) {
+            // do nothing
+        }
+    }
+
+    private fun moveToState(state: State) {
+        this.state = state
+    }
+
+    private enum class State {
+        STARTING,
+        STARTED,
+        STOPPING,
+        STOPPED
+    }
+
+    private fun waitFor(intervalMs: Int, timeoutMs: Int, test: () -> Boolean) {
+        val start = System.currentTimeMillis()
+        var lastCheck = start
+        while (!test()) {
+            val now = System.currentTimeMillis()
+            val timeSinceStart = now - start
+            val timeSinceLastCheck = now - lastCheck
+            if (timeoutMs in 0..timeSinceStart) {
+                throw TimeoutException()
+            }
+            val sleepTime = intervalMs - timeSinceLastCheck
+            if (sleepTime > 0) {
+                Thread.sleep(sleepTime)
+            }
+            lastCheck = System.currentTimeMillis()
+        }
+    }
+
+}

src/main/kotlin/eu/mjdev/dadb/helpers/AdbConnection.kt

@@ -0,0 +1,135 @@
+package eu.mjdev.dadb.helpers
+
+import okio.Sink
+import okio.Source
+import okio.sink
+import okio.source
+import org.jetbrains.annotations.TestOnly
+import java.io.Closeable
+import java.io.IOException
+import java.net.Socket
+import java.util.*
+
+internal class AdbConnection internal constructor(
+    adbReader: AdbReader,
+    private val adbWriter: AdbWriter,
+    private val closeable: Closeable?,
+    private val supportedFeatures: Set<String>,
+    private val version: Int,
+    private val maxPayloadSize: Int
+) : AutoCloseable {
+
+    private val random = Random()
+    private val messageQueue = AdbMessageQueue(adbReader)
+
+    @Throws(IOException::class)
+    fun open(destination: String): AdbStream {
+        val localId = newId()
+        messageQueue.startListening(localId)
+        try {
+            adbWriter.writeOpen(localId, destination)
+            val message = messageQueue.take(localId, Constants.CMD_OKAY)
+            val remoteId = message.arg0
+            return AdbStreamImpl(messageQueue, adbWriter, maxPayloadSize, localId, remoteId)
+        } catch (e: Throwable) {
+            messageQueue.stopListening(localId)
+            throw e
+        }
+    }
+
+    fun supportsFeature(feature: String): Boolean {
+        return supportedFeatures.contains(feature)
+    }
+
+    private fun newId(): Int {
+        return random.nextInt()
+    }
+
+    @TestOnly
+    internal fun ensureEmpty() {
+        messageQueue.ensureEmpty()
+    }
+
+    override fun close() {
+        try {
+            messageQueue.close()
+            adbWriter.close()
+            closeable?.close()
+        } catch (ignore: Throwable) {
+        }
+    }
+
+    companion object {
+
+        fun connect(socket: Socket, keyPair: AdbKeyPair? = null): AdbConnection {
+            val source = socket.source()
+            val sink = socket.sink()
+            return connect(source, sink, keyPair, socket)
+        }
+
+        private fun connect(
+            source: Source,
+            sink: Sink,
+            keyPair: AdbKeyPair? = null,
+            closeable: Closeable? = null
+        ): AdbConnection {
+            val adbReader = AdbReader(source)
+            val adbWriter = AdbWriter(sink)
+
+            try {
+                return connect(adbReader, adbWriter, keyPair, closeable)
+            } catch (t: Throwable) {
+                adbReader.close()
+                adbWriter.close()
+                throw t
+            }
+        }
+
+        private fun connect(
+            adbReader: AdbReader,
+            adbWriter: AdbWriter,
+            keyPair: AdbKeyPair?,
+            closeable: Closeable?
+        ): AdbConnection {
+            adbWriter.writeConnect()
+
+            var message = adbReader.readMessage()
+
+            if (message.command == Constants.CMD_AUTH) {
+                checkNotNull(keyPair) { "Authentication required but no KeyPair provided" }
+                check(message.arg0 == Constants.AUTH_TYPE_TOKEN) { "Unsupported auth type: $message" }
+
+                val signature = keyPair.signPayload(message)
+                adbWriter.writeAuth(Constants.AUTH_TYPE_SIGNATURE, signature)
+
+                message = adbReader.readMessage()
+                if (message.command == Constants.CMD_AUTH) {
+                    adbWriter.writeAuth(Constants.AUTH_TYPE_RSA_PUBLIC, keyPair.publicKeyBytes)
+                    message = adbReader.readMessage()
+                }
+            }
+
+            if (message.command != Constants.CMD_CNXN) throw IOException("Connection failed: $message")
+
+            val connectionString = parseConnectionString(String(message.payload))
+            val version = message.arg0
+            val maxPayloadSize = message.arg1
+
+            return AdbConnection(adbReader, adbWriter, closeable, connectionString.features, version, maxPayloadSize)
+        }
+
+        // ie: "device::ro.product.name=sdk_gphone_x86;ro.product.model=Android SDK built for x86;ro.product.device=generic_x86;features=fixed_push_symlink_timestamp,apex,fixed_push_mkdir,stat_v2,abb_exec,cmd,abb,shell_v2"
+        private fun parseConnectionString(connectionString: String): ConnectionString {
+            val keyValues = connectionString.substringAfter("device::")
+                .split(";")
+                .map { it.split("=") }
+                .mapNotNull { if (it.size != 2) null else it[0] to it[1] }
+                .toMap()
+            if ("features" !in keyValues) throw IOException("Failed to parse features from connection string: $connectionString")
+            val features = keyValues.getValue("features").split(",").toSet()
+            return ConnectionString(features)
+        }
+    }
+}
+
+private data class ConnectionString(val features: Set<String>)

src/main/kotlin/eu/mjdev/dadb/helpers/AdbKeyPair.kt

@@ -0,0 +1,163 @@
+package eu.mjdev.dadb.helpers
+
+import java.io.File
+import java.math.BigInteger
+import java.nio.ByteBuffer
+import java.nio.ByteOrder
+import java.security.KeyPairGenerator
+import java.security.PrivateKey
+import java.security.interfaces.RSAPublicKey
+import java.util.*
+import javax.crypto.Cipher
+
+class AdbKeyPair(
+    private val privateKey: PrivateKey,
+    internal val publicKeyBytes: ByteArray
+) {
+
+    internal fun signPayload(message: AdbMessage): ByteArray {
+        val cipher = Cipher.getInstance("RSA/ECB/NoPadding")
+        cipher.init(Cipher.ENCRYPT_MODE, privateKey)
+        cipher.update(SIGNATURE_PADDING)
+        return cipher.doFinal(message.payload, 0, message.payloadLength)
+    }
+
+    companion object {
+
+        private const val KEY_LENGTH_BITS = 2048
+        private const val KEY_LENGTH_BYTES = KEY_LENGTH_BITS / 8
+        private const val KEY_LENGTH_WORDS = KEY_LENGTH_BYTES / 4
+
+        @OptIn(ExperimentalUnsignedTypes::class)
+        private val SIGNATURE_PADDING = ubyteArrayOf(
+            0x00u, 0x01u, 0xffu, 0xffu, 0xffu, 0xffu, 0xffu, 0xffu, 0xffu, 0xffu, 0xffu, 0xffu, 0xffu,
+            0xffu, 0xffu, 0xffu, 0xffu, 0xffu, 0xffu, 0xffu, 0xffu, 0xffu, 0xffu, 0xffu, 0xffu, 0xffu,
+            0xffu, 0xffu, 0xffu, 0xffu, 0xffu, 0xffu, 0xffu, 0xffu, 0xffu, 0xffu, 0xffu, 0xffu, 0xffu,
+            0xffu, 0xffu, 0xffu, 0xffu, 0xffu, 0xffu, 0xffu, 0xffu, 0xffu, 0xffu, 0xffu, 0xffu, 0xffu,
+            0xffu, 0xffu, 0xffu, 0xffu, 0xffu, 0xffu, 0xffu, 0xffu, 0xffu, 0xffu, 0xffu, 0xffu, 0xffu,
+            0xffu, 0xffu, 0xffu, 0xffu, 0xffu, 0xffu, 0xffu, 0xffu, 0xffu, 0xffu, 0xffu, 0xffu, 0xffu,
+            0xffu, 0xffu, 0xffu, 0xffu, 0xffu, 0xffu, 0xffu, 0xffu, 0xffu, 0xffu, 0xffu, 0xffu, 0xffu,
+            0xffu, 0xffu, 0xffu, 0xffu, 0xffu, 0xffu, 0xffu, 0xffu, 0xffu, 0xffu, 0xffu, 0xffu, 0xffu,
+            0xffu, 0xffu, 0xffu, 0xffu, 0xffu, 0xffu, 0xffu, 0xffu, 0xffu, 0xffu, 0xffu, 0xffu, 0xffu,
+            0xffu, 0xffu, 0xffu, 0xffu, 0xffu, 0xffu, 0xffu, 0xffu, 0xffu, 0xffu, 0xffu, 0xffu, 0xffu,
+            0xffu, 0xffu, 0xffu, 0xffu, 0xffu, 0xffu, 0xffu, 0xffu, 0xffu, 0xffu, 0xffu, 0xffu, 0xffu,
+            0xffu, 0xffu, 0xffu, 0xffu, 0xffu, 0xffu, 0xffu, 0xffu, 0xffu, 0xffu, 0xffu, 0xffu, 0xffu,
+            0xffu, 0xffu, 0xffu, 0xffu, 0xffu, 0xffu, 0xffu, 0xffu, 0xffu, 0xffu, 0xffu, 0xffu, 0xffu,
+            0xffu, 0xffu, 0xffu, 0xffu, 0xffu, 0xffu, 0xffu, 0xffu, 0xffu, 0xffu, 0xffu, 0xffu, 0xffu,
+            0xffu, 0xffu, 0xffu, 0xffu, 0xffu, 0xffu, 0xffu, 0xffu, 0xffu, 0xffu, 0xffu, 0xffu, 0xffu,
+            0xffu, 0xffu, 0xffu, 0xffu, 0xffu, 0xffu, 0xffu, 0xffu, 0xffu, 0xffu, 0xffu, 0xffu, 0xffu,
+            0xffu, 0xffu, 0xffu, 0xffu, 0xffu, 0xffu, 0xffu, 0xffu, 0xffu, 0xffu, 0xffu, 0xffu, 0x00u,
+            0x30u, 0x21u, 0x30u, 0x09u, 0x06u, 0x05u, 0x2bu, 0x0eu, 0x03u, 0x02u, 0x1au, 0x05u, 0x00u,
+            0x04u, 0x14u
+        ).toByteArray()
+
+        @JvmStatic
+        fun readDefault(): AdbKeyPair {
+            val privateKeyFile = File(System.getenv("HOME"), ".android/adbkey")
+            val publicKeyFile = File(System.getenv("HOME"), ".android/adbkey.pub")
+
+            if (!privateKeyFile.exists()) {
+                generate(privateKeyFile, publicKeyFile)
+            }
+
+            return read(privateKeyFile, publicKeyFile)
+        }
+
+        @JvmStatic
+        @JvmOverloads
+        fun read(privateKeyFile: File, publicKeyFile: File? = null): AdbKeyPair {
+            val privateKey = PKCS8.parse(privateKeyFile.readBytes())
+            val publicKeyBytes = if (publicKeyFile?.exists() == true) {
+                readAdbPublicKey(publicKeyFile)
+            } else {
+                ByteArray(0)
+            }
+
+            return AdbKeyPair(privateKey, publicKeyBytes)
+        }
+
+        @JvmStatic
+        fun generate(privateKeyFile: File, publicKeyFile: File) {
+            val keyPair = KeyPairGenerator.getInstance("RSA").let {
+                it.initialize(KEY_LENGTH_BITS)
+                it.genKeyPair()
+            }
+
+            privateKeyFile.absoluteFile.parentFile?.mkdirs()
+            publicKeyFile.absoluteFile.parentFile?.mkdirs()
+
+            privateKeyFile.writer().use { out ->
+                val base64 = Base64.getMimeEncoder(64, "\n".toByteArray())
+                out.write("-----BEGIN PRIVATE KEY-----\n")
+                out.write(base64.encodeToString(keyPair.private.encoded))
+                out.write("\n-----END PRIVATE KEY-----")
+            }
+
+            publicKeyFile.writer().use { out ->
+                val base64 = Base64.getEncoder()
+                val bytes = convertRsaPublicKeyToAdbFormat(keyPair.public as RSAPublicKey)
+                out.write(base64.encodeToString(bytes))
+                out.write(" unknown@unknown")
+            }
+        }
+
+        private fun readAdbPublicKey(file: File): ByteArray {
+            val bytes = file.readBytes()
+            val publicKeyBytes = bytes.copyOf(bytes.size + 1)
+            publicKeyBytes[bytes.size] = 0
+            return publicKeyBytes
+        }
+
+        // https://github.com/cgutman/AdbLib/blob/d6937951eb98557c76ee2081e383d50886ce109a/src/com/cgutman/adblib/AdbCrypto.java#L83-L137
+        @Suppress("JoinDeclarationAndAssignment")
+        private fun convertRsaPublicKeyToAdbFormat(pubkey: RSAPublicKey): ByteArray {
+            /*
+             * ADB literally just saves the RSAPublicKey struct to a file.
+             *
+             * typedef struct RSAPublicKey {
+             * int len; // Length of n[] in number of uint32_t
+             * uint32_t n0inv;  // -1 / n[0] mod 2^32
+             * uint32_t n[RSANUMWORDS]; // modulus as little endian array
+             * uint32_t rr[RSANUMWORDS]; // R^2 as little endian array
+             * int exponent; // 3 or 65537
+             * } RSAPublicKey;
+             */
+
+            /* ------ This part is a Java-ified version of RSA_to_RSAPublicKey from adb_host_auth.c ------ */
+            val r32: BigInteger
+            val r: BigInteger
+            var rr: BigInteger
+            var rem: BigInteger
+            var n: BigInteger
+            val n0inv: BigInteger
+            r32 = BigInteger.ZERO.setBit(32)
+            n = pubkey.modulus
+            r = BigInteger.ZERO.setBit(KEY_LENGTH_WORDS * 32)
+            rr = r.modPow(BigInteger.valueOf(2), n)
+            rem = n.remainder(r32)
+            n0inv = rem.modInverse(r32)
+            val myN = IntArray(KEY_LENGTH_WORDS)
+            val myRr = IntArray(KEY_LENGTH_WORDS)
+            var res: Array<BigInteger>
+            for (i in 0 until KEY_LENGTH_WORDS) {
+                res = rr.divideAndRemainder(r32)
+                rr = res[0]
+                rem = res[1]
+                myRr[i] = rem.toInt()
+                res = n.divideAndRemainder(r32)
+                n = res[0]
+                rem = res[1]
+                myN[i] = rem.toInt()
+            }
+
+            /* ------------------------------------------------------------------------------------------- */
+            val bbuf: ByteBuffer = ByteBuffer.allocate(524).order(ByteOrder.LITTLE_ENDIAN)
+            bbuf.putInt(KEY_LENGTH_WORDS)
+            bbuf.putInt(n0inv.negate().toInt())
+            for (i in myN) bbuf.putInt(i)
+            for (i in myRr) bbuf.putInt(i)
+            bbuf.putInt(pubkey.publicExponent.toInt())
+            return bbuf.array()
+        }
+    }
+}

src/main/kotlin/eu/mjdev/dadb/helpers/AdbMessage.kt

@@ -0,0 +1,72 @@
+package eu.mjdev.dadb.helpers
+
+import okio.Buffer
+import okio.BufferedSource
+import java.nio.charset.StandardCharsets
+
+internal class AdbMessage(
+    val command: Int,
+    val arg0: Int,
+    val arg1: Int,
+    val payloadLength: Int,
+    val checksum: Int,
+    val magic: Int,
+    val payload: ByteArray
+) {
+
+    override fun toString() = "${commandStr()}[${argStr(arg0)}, ${argStr(arg1)}] ${payloadStr()}"
+
+    private fun payloadStr(): String {
+        if (payloadLength == 0) return ""
+        return when (command) {
+            Constants.CMD_AUTH -> if (arg0 == Constants.AUTH_TYPE_RSA_PUBLIC) String(payload) else "auth[${payloadLength}]"
+            Constants.CMD_WRTE -> writePayloadStr()
+            Constants.CMD_OPEN -> String(payload, 0, payloadLength - 1)
+            else -> "payload[$payloadLength]"
+        }
+    }
+
+    private fun writePayloadStr(): String {
+        shellPayloadStr()?.let { return it }
+        syncPayloadStr()?.let { return it }
+        return "payload[$payloadLength]"
+    }
+
+    @Suppress("UsePropertyAccessSyntax")
+    private fun shellPayloadStr(): String? {
+        val source: BufferedSource = getSource()
+        if (source.buffer.size < 5) return null
+        val id = source.readByte().toInt()
+        if (id < 0 || id > 3) return null
+        val length = source.readIntLe()
+        if (length != source.buffer.size.toInt()) return null
+        if (id == ID_EXIT) return "[shell] exit(${source.readByte()})"
+        val payload = String(payload, 5, payloadLength - 5)
+        return "[shell] $payload"
+    }
+
+    private fun syncPayloadStr(): String? {
+        val source: BufferedSource = getSource()
+        if (source.buffer.size < 8) return null
+        val id = source.readString(4, StandardCharsets.UTF_8)
+        if (id !in SYNC_IDS) return null
+        val arg = source.readIntLe()
+        return "[sync] $id($arg)"
+    }
+
+    private fun getSource(): BufferedSource {
+        return Buffer().apply { write(payload, 0, payloadLength) }
+    }
+
+    private fun argStr(arg: Int) = String.format("%X", arg)
+
+    private fun commandStr() = when (command) {
+        Constants.CMD_AUTH -> "AUTH";
+        Constants.CMD_CNXN -> "CNXN";
+        Constants.CMD_OPEN -> "OPEN";
+        Constants.CMD_OKAY -> "OKAY";
+        Constants.CMD_CLSE -> "CLSE";
+        Constants.CMD_WRTE -> "WRTE";
+        else -> "????"
+    }
+}

src/main/kotlin/eu/mjdev/dadb/helpers/AdbMessageQueue.kt

@@ -0,0 +1,13 @@
+package eu.mjdev.dadb.helpers
+
+internal class AdbMessageQueue(private val adbReader: AdbReader) : AutoCloseable, MessageQueue<AdbMessage>() {
+    override fun readMessage() = adbReader.readMessage()
+
+    override fun getLocalId(message: AdbMessage) = message.arg1
+
+    override fun getCommand(message: AdbMessage) = message.command
+
+    override fun close() = adbReader.close()
+
+    override fun isCloseCommand(message: AdbMessage) = message.command == Constants.CMD_CLSE
+}

src/main/kotlin/eu/mjdev/dadb/helpers/AdbReader.kt

@@ -0,0 +1,29 @@
+package eu.mjdev.dadb.helpers
+
+import okio.Source
+import okio.buffer
+
+internal class AdbReader(source: Source) : AutoCloseable {
+    private val bufferedSource = source.buffer()
+
+    fun readMessage(): AdbMessage {
+        synchronized(bufferedSource) {
+            bufferedSource.apply {
+                val command = readIntLe()
+                val arg0 = readIntLe()
+                val arg1 = readIntLe()
+                val payloadLength = readIntLe()
+                val checksum = readIntLe()
+                val magic = readIntLe()
+                val payload = readByteArray(payloadLength.toLong())
+                return AdbMessage(command, arg0, arg1, payloadLength, checksum, magic, payload).also {
+                    log { "(${Thread.currentThread().name}) < $it" }
+                }
+            }
+        }
+    }
+
+    override fun close() {
+        bufferedSource.close()
+    }
+}

src/main/kotlin/eu/mjdev/dadb/helpers/AdbShell.kt

@@ -0,0 +1,116 @@
+@file:JvmName("AdbShell")
+
+package eu.mjdev.dadb.helpers
+
+import eu.mjdev.dadb.helpers.AdbShellPacket.*
+import java.io.IOException
+
+const val ID_STDIN = 0
+const val ID_STDOUT = 1
+const val ID_STDERR = 2
+const val ID_EXIT = 3
+const val ID_CLOSE_STDIN = 3
+
+class AdbShellStream(
+    private val stream: AdbStream
+) : AutoCloseable {
+    @Throws(IOException::class)
+    fun readAll(): AdbShellResponse {
+        val output = StringBuilder()
+        val errorOutput = StringBuilder()
+        while (true) {
+            when (val packet = read()) {
+                is Exit -> {
+                    val exitCode = packet.payload[0].toInt()
+                    return AdbShellResponse(output.toString(), errorOutput.toString(), exitCode)
+                }
+
+                is StdOut -> {
+                    output.append(String(packet.payload))
+                }
+
+                is StdError -> {
+                    errorOutput.append(String(packet.payload))
+                }
+            }
+        }
+    }
+
+    @Throws(IOException::class)
+    fun read(): AdbShellPacket {
+        stream.source.apply {
+            val id = checkId(readByte().toInt())
+            val length = checkLength(id, readIntLe())
+            val payload = readByteArray(length.toLong())
+            return when (id) {
+                ID_STDOUT -> StdOut(payload)
+                ID_STDERR -> StdError(payload)
+                ID_EXIT -> Exit(payload)
+                else -> throw IllegalArgumentException("Invalid shell packet id: $id")
+            }
+        }
+    }
+
+    @Throws(IOException::class)
+    fun write(string: String) {
+        write(ID_STDIN, string.toByteArray())
+    }
+
+    @Throws(IOException::class)
+    fun write(id: Int, payload: ByteArray? = null) {
+        stream.sink.apply {
+            writeByte(id)
+            writeIntLe(payload?.size ?: 0)
+            if (payload != null) write(payload)
+            flush()
+        }
+    }
+
+    override fun close() {
+        stream.close()
+    }
+
+    private fun checkId(id: Int): Int {
+        check(id == ID_STDOUT || id == ID_STDERR || id == ID_EXIT) {
+            "Invalid shell packet id: $id"
+        }
+        return id
+    }
+
+    private fun checkLength(id: Int, length: Int): Int {
+        check(length >= 0) { "Shell packet length must be >= 0: $length" }
+        check(id != ID_EXIT || length == 1) { "Shell exit packet does not have payload length == 1: $length" }
+        return length
+    }
+}
+
+sealed class AdbShellPacket(
+    open val payload: ByteArray
+) {
+    abstract val id: Int
+
+    class StdOut(override val payload: ByteArray) : AdbShellPacket(payload) {
+        override val id: Int = ID_STDOUT
+        override fun toString() = "STDOUT: ${String(payload)}"
+    }
+
+    class StdError(override val payload: ByteArray) : AdbShellPacket(payload) {
+        override val id: Int = ID_STDERR
+        override fun toString() = "STDERR: ${String(payload)}"
+    }
+
+    class Exit(override val payload: ByteArray) : AdbShellPacket(payload) {
+        override val id: Int = ID_EXIT
+        override fun toString() = "EXIT: ${payload[0]}"
+    }
+}
+
+class AdbShellResponse(
+    val output: String,
+    val errorOutput: String,
+    val exitCode: Int
+) {
+    val allOutput: String by lazy { "$output$errorOutput" }
+
+    override fun toString() = "Shell response ($exitCode):\n$allOutput"
+}

src/main/kotlin/eu/mjdev/dadb/helpers/AdbStream.kt

@@ -0,0 +1,109 @@
+package eu.mjdev.dadb.helpers
+
+import okio.*
+import java.lang.Integer.min
+import java.nio.ByteBuffer
+
+interface AdbStream : AutoCloseable {
+    val source: BufferedSource
+
+    val sink: BufferedSink
+}
+
+internal class AdbStreamImpl internal constructor(
+    private val messageQueue: AdbMessageQueue,
+    private val adbWriter: AdbWriter,
+    private val maxPayloadSize: Int,
+    private val localId: Int,
+    private val remoteId: Int
+) : AdbStream {
+    private var isClosed = false
+
+    override val source = object : Source {
+
+        private var message: AdbMessage? = null
+        private var bytesRead = 0
+
+        override fun read(sink: Buffer, byteCount: Long): Long {
+            val message = message() ?: return -1
+
+            val bytesRemaining = message.payloadLength - bytesRead
+            val bytesToRead = Math.min(byteCount.toInt(), bytesRemaining)
+
+            sink.write(message.payload, bytesRead, bytesToRead)
+
+            bytesRead += bytesToRead
+
+            check(bytesRead <= message.payloadLength)
+
+            if (bytesRead == message.payloadLength) {
+                this.message = null
+                adbWriter.writeOkay(localId, remoteId)
+            }
+
+            return bytesToRead.toLong()
+        }
+
+        private fun message(): AdbMessage? {
+            message?.let { return it }
+            val nextMessage = nextMessage(Constants.CMD_WRTE)
+            message = nextMessage
+            bytesRead = 0
+            return nextMessage
+        }
+
+        override fun close() {}
+
+        override fun timeout() = Timeout.NONE
+    }.buffer()
+
+    override val sink = object : Sink {
+        private val buffer = ByteBuffer.allocate(maxPayloadSize)
+
+        override fun write(source: Buffer, byteCount: Long) {
+            var remainingBytes = byteCount
+            while (true) {
+                remainingBytes -= writeToBuffer(source, byteCount)
+                if (remainingBytes == 0L) return
+                check(remainingBytes > 0L)
+            }
+        }
+
+        private fun writeToBuffer(source: BufferedSource, byteCount: Long): Int {
+            val bytesToWrite = min(buffer.remaining(), byteCount.toInt())
+            val bytesWritten = source.read(buffer.array(), buffer.position(), bytesToWrite)
+
+            buffer.position(buffer.position() + bytesWritten)
+            if (buffer.remaining() == 0) flush()
+
+            return bytesWritten
+        }
+
+        override fun flush() {
+            adbWriter.writeWrite(localId, remoteId, buffer.array(), 0, buffer.position())
+            buffer.clear()
+        }
+
+        override fun close() {}
+
+        override fun timeout() = Timeout.NONE
+    }.buffer()
+
+    private fun nextMessage(command: Int): AdbMessage? {
+        return try {
+            messageQueue.take(localId, command)
+        } catch (e: IOException) {
+            close()
+            return null
+        }
+    }
+
+    override fun close() {
+        if (isClosed) return
+        isClosed = true
+
+        adbWriter.writeClose(localId, remoteId)
+
+        messageQueue.stopListening(localId)
+    }
+}

src/main/kotlin/eu/mjdev/dadb/helpers/AdbStreamClosed.kt

@@ -0,0 +1,6 @@
+package eu.mjdev.dadb.helpers
+
+import java.io.IOException
+
+internal class AdbStreamClosed(localId: Int) :
+    IOException(String.format("ADB stream is closed for localId: %x", localId))

src/main/kotlin/eu/mjdev/dadb/helpers/AdbSync.kt

@@ -0,0 +1,100 @@
+package eu.mjdev.dadb.helpers
+
+import okio.Buffer
+import okio.Sink
+import okio.Source
+import java.io.IOException
+import java.nio.charset.StandardCharsets
+
+internal const val LIST = "LIST"
+internal const val RECV = "RECV"
+internal const val SEND = "SEND"
+internal const val STAT = "STAT"
+internal const val DATA = "DATA"
+internal const val DONE = "DONE"
+internal const val OKAY = "OKAY"
+internal const val QUIT = "QUIT"
+internal const val FAIL = "FAIL"
+
+internal val SYNC_IDS = setOf(LIST, RECV, SEND, STAT, DATA, DONE, OKAY, QUIT, FAIL)
+
+private class Packet(val id: String, val arg: Int)
+
+class AdbSyncStream(
+    private val stream: AdbStream
+) : AutoCloseable {
+    private val buffer = Buffer()
+
+    @Throws(IOException::class)
+    fun send(source: Source, remotePath: String, mode: Int, lastModifiedMs: Long) {
+        val remote = "$remotePath,$mode"
+        writePacket(SEND, remote.length)
+
+        stream.sink.apply {
+            writeString(remote, StandardCharsets.UTF_8)
+            flush()
+        }
+
+        buffer.clear()
+
+        while (true) {
+            val read = source.read(buffer, 64_000)
+            if (read == -1L) break
+            writePacket(DATA, read.toInt())
+            val sent = stream.sink.writeAll(buffer)
+            check(read == sent)
+        }
+
+        writePacket(DONE, (lastModifiedMs / 1000).toInt())
+
+        stream.sink.flush()
+
+        val packet = readPacket()
+        if (packet.id != OKAY) throw IOException("Unexpected sync packet id: ${packet.id}")
+    }
+
+    @Throws(IOException::class)
+    fun recv(sink: Sink, remotePath: String) {
+        writePacket(RECV, remotePath.length)
+        stream.sink.apply {
+            writeString(remotePath, StandardCharsets.UTF_8)
+            flush()
+        }
+
+        buffer.clear()
+
+        while (true) {
+            val packet = readPacket()
+            if (packet.id == DONE) break
+            if (packet.id == FAIL) {
+                val message = stream.source.readString(packet.arg.toLong(), StandardCharsets.UTF_8)
+                throw IOException("Sync failed: $message")
+            }
+            if (packet.id != DATA) throw IOException("Unexpected sync packet id: ${packet.id}")
+            val chunkSize = packet.arg
+            stream.source.readFully(buffer, chunkSize.toLong())
+            buffer.readAll(sink)
+        }
+
+        sink.flush()
+    }
+
+    private fun writePacket(id: String, arg: Int) {
+        stream.sink.apply {
+            writeString(id, StandardCharsets.UTF_8)
+            writeIntLe(arg)
+            flush()
+        }
+    }
+
+    private fun readPacket(): Packet {
+        val id = stream.source.readString(4, StandardCharsets.UTF_8)
+        val arg = stream.source.readIntLe()
+        return Packet(id, arg)
+    }
+
+    override fun close() {
+        writePacket(QUIT, 0)
+        stream.close()
+    }
+}

src/main/kotlin/eu/mjdev/dadb/helpers/AdbWriter.kt

@@ -0,0 +1,105 @@
+package eu.mjdev.dadb.helpers
+
+import okio.Sink
+import okio.buffer
+import java.nio.ByteBuffer
+
+internal class AdbWriter(sink: Sink) : AutoCloseable {
+    private val bufferedSink = sink.buffer()
+
+    fun writeConnect() = write(
+        Constants.CMD_CNXN,
+        Constants.CONNECT_VERSION,
+        Constants.CONNECT_MAXDATA,
+        Constants.CONNECT_PAYLOAD,
+        0,
+        Constants.CONNECT_PAYLOAD.size
+    )
+
+    fun writeAuth(authType: Int, authPayload: ByteArray) = write(
+        Constants.CMD_AUTH,
+        authType,
+        0,
+        authPayload,
+        0,
+        authPayload.size
+    )
+
+    fun writeOpen(localId: Int, destination: String) {
+        val destinationBytes = destination.toByteArray()
+        val buffer = ByteBuffer.allocate(destinationBytes.size + 1)
+        buffer.put(destinationBytes)
+        buffer.put(0)
+        val payload = buffer.array()
+        write(Constants.CMD_OPEN, localId, 0, payload, 0, payload.size)
+    }
+
+    fun writeWrite(localId: Int, remoteId: Int, payload: ByteArray, offset: Int, length: Int) {
+        write(Constants.CMD_WRTE, localId, remoteId, payload, offset, length)
+    }
+
+    fun writeClose(localId: Int, remoteId: Int) {
+        write(Constants.CMD_CLSE, localId, remoteId, null, 0, 0)
+    }
+
+    fun writeOkay(localId: Int, remoteId: Int) {
+        write(Constants.CMD_OKAY, localId, remoteId, null, 0, 0)
+    }
+
+    fun write(
+        command: Int,
+        arg0: Int,
+        arg1: Int,
+        payload: ByteArray?,
+        offset: Int,
+        length: Int
+    ) {
+        log {
+            "(${Thread.currentThread().name}) > ${
+                AdbMessage(
+                    command,
+                    arg0,
+                    arg1,
+                    length,
+                    0,
+                    0,
+                    payload ?: ByteArray(0)
+                )
+            }"
+        }
+        synchronized(bufferedSink) {
+            bufferedSink.apply {
+                writeIntLe(command)
+                writeIntLe(arg0)
+                writeIntLe(arg1)
+                if (payload == null) {
+                    writeIntLe(0)
+                    writeIntLe(0)
+                } else {
+                    writeIntLe(length)
+                    writeIntLe(payloadChecksum(payload))
+                }
+                writeIntLe(command xor -0x1)
+                if (payload != null) {
+                    write(payload, offset, length)
+                }
+                flush()
+            }
+        }
+    }
+
+    override fun close() {
+        bufferedSink.close()
+    }
+
+    companion object {
+
+        private fun payloadChecksum(payload: ByteArray): Int {
+            var checksum = 0
+            for (byte in payload) {
+                checksum += byte.toUByte().toInt()
+            }
+            return checksum
+        }
+    }
+}

src/main/kotlin/eu/mjdev/dadb/helpers/Constants.kt

@@ -0,0 +1,19 @@
+package eu.mjdev.dadb.helpers
+
+internal object Constants {
+    const val AUTH_TYPE_TOKEN = 1
+    const val AUTH_TYPE_SIGNATURE = 2
+    const val AUTH_TYPE_RSA_PUBLIC = 3
+
+    const val CMD_AUTH = 0x48545541
+    const val CMD_CNXN = 0x4e584e43
+    const val CMD_OPEN = 0x4e45504f
+    const val CMD_OKAY = 0x59414b4f
+    const val CMD_CLSE = 0x45534c43
+    const val CMD_WRTE = 0x45545257
+
+    const val CONNECT_VERSION = 0x01000000
+    const val CONNECT_MAXDATA = 1024 * 1024
+
+    val CONNECT_PAYLOAD = "host::\u0000".toByteArray()
+}

src/main/kotlin/eu/mjdev/dadb/helpers/DadbImpl.kt

@@ -0,0 +1,64 @@
+package eu.mjdev.dadb.helpers
+
+import eu.mjdev.dadb.Dadb
+import org.jetbrains.annotations.TestOnly
+import java.net.InetSocketAddress
+import java.net.Socket
+
+internal class DadbImpl @Throws(IllegalArgumentException::class) constructor(
+    private val host: String,
+    private val port: Int,
+    private val keyPair: AdbKeyPair? = null,
+    private val connectTimeout: Int = 0,
+    private val socketTimeout: Int = 0
+) : Dadb {
+    init {
+        if (port < 0) {
+            throw IllegalArgumentException("port must be >= 0")
+        }
+        if (connectTimeout < 0) {
+            throw IllegalArgumentException("connectTimeout must be >= 0")
+        }
+        if (socketTimeout < 0) {
+            throw IllegalArgumentException("socketTimeout must be >= 0")
+        }
+    }
+
+    private var connection: Pair<AdbConnection, Socket>? = null
+
+    override fun open(destination: String) = connection().open(destination)
+
+    override fun supportsFeature(feature: String): Boolean {
+        return connection().supportsFeature(feature)
+    }
+
+    override fun close() {
+        connection?.first?.close()
+    }
+
+    override fun toString() = "$host:$port"
+
+    @TestOnly
+    fun closeConnection() {
+        connection?.second?.close()
+    }
+
+    @Synchronized
+    private fun connection(): AdbConnection {
+        var connection = connection
+        if (connection == null || connection.second.isClosed) {
+            connection = newConnection()
+            this.connection = connection
+        }
+        return connection.first
+    }
+
+    private fun newConnection(): Pair<AdbConnection, Socket> {
+        val socketAddress = InetSocketAddress(host, port)
+        val socket = Socket()
+        socket.soTimeout = socketTimeout
+        socket.connect(socketAddress, connectTimeout)
+        val adbConnection = AdbConnection.connect(socket, keyPair)
+        return adbConnection to socket
+    }
+}

src/main/kotlin/eu/mjdev/dadb/helpers/Logging.kt

@@ -0,0 +1,9 @@
+package eu.mjdev.dadb.helpers
+
+private val ENABLED = "true" == System.getenv("DADB_LOGGING")
+
+internal fun log(block: () -> String) {
+    if (ENABLED) {
+        println(block())
+    }
+}

src/main/kotlin/eu/mjdev/dadb/helpers/MessageQueue.kt

@@ -0,0 +1,100 @@
+package eu.mjdev.dadb.helpers
+
+import org.jetbrains.annotations.TestOnly
+import java.util.*
+import java.util.concurrent.ConcurrentHashMap
+import java.util.concurrent.ConcurrentLinkedQueue
+import java.util.concurrent.locks.ReentrantLock
+
+internal abstract class MessageQueue<V> {
+    private val readLock = ReentrantLock()
+    private val queueLock = ReentrantLock()
+    private val queueCond = queueLock.newCondition()
+    private val queues = ConcurrentHashMap<Int, ConcurrentHashMap<Int, Queue<V>>>()
+    private val openStreams = ConcurrentHashMap<Int, Boolean>().keySet(true)
+
+    fun take(localId: Int, command: Int): V {
+        while (true) {
+            queueLock.lock {
+                poll(localId, command)?.let { return it }
+                readLock.tryLock({
+                    queueLock.unlock()
+                    read()
+                    queueLock.lock()
+                    queueCond.signalAll()
+                }) { queueCond.await() }
+            }
+        }
+    }
+
+    fun startListening(localId: Int) {
+        openStreams.add(localId)
+        queues.putIfAbsent(localId, ConcurrentHashMap())
+    }
+
+    fun stopListening(localId: Int) {
+        openStreams.remove(localId)
+        queues.remove(localId)
+    }
+
+    @TestOnly
+    fun ensureEmpty() {
+        check(queues.isEmpty()) { "Queues is not empty: ${queues.keys.map { String.format("%X", it) }}" }
+        check(openStreams.isEmpty())
+    }
+
+    protected abstract fun readMessage(): V
+
+    protected abstract fun getLocalId(message: V): Int
+
+    protected abstract fun getCommand(message: V): Int
+
+    protected abstract fun isCloseCommand(message: V): Boolean
+
+    private fun poll(localId: Int, command: Int): V? {
+        val streamQueues = queues[localId] ?: throw IllegalStateException("Not listening for localId: $localId")
+        val message = streamQueues[command]?.poll()
+        if (message == null && !openStreams.contains(localId)) {
+            throw AdbStreamClosed(localId)
+        }
+        return message
+    }
+
+    private fun read() {
+        val message = readMessage()
+        val localId = getLocalId(message)
+
+        if (isCloseCommand(message)) {
+            openStreams.remove(localId)
+            return
+        }
+
+        val streamQueues = queues[localId] ?: return
+
+        val command = getCommand(message)
+        val commandQueue = streamQueues.computeIfAbsent(command) { ConcurrentLinkedQueue() }
+
+        commandQueue.add(message)
+    }
+}
+
+private inline fun <T> ReentrantLock.lock(body: () -> T) {
+    lock()
+    try {
+        body()
+    } finally {
+        if (isHeldByCurrentThread) unlock()
+    }
+}
+
+private inline fun ReentrantLock.tryLock(body: () -> Unit, elseBody: () -> Unit) {
+    return if (tryLock()) {
+        try {
+            body()
+        } finally {
+            if (isHeldByCurrentThread) unlock()
+        }
+    } else {
+        elseBody()
+    }
+}

src/main/kotlin/eu/mjdev/dadb/helpers/PKCS8.kt

@@ -0,0 +1,19 @@
+package eu.mjdev.dadb.helpers
+
+import java.security.KeyFactory
+import java.security.PrivateKey
+import java.security.spec.PKCS8EncodedKeySpec
+import java.util.*
+
+internal object PKCS8 {
+    private const val PREFIX = "-----BEGIN PRIVATE KEY-----"
+    private const val SUFFIX = "-----END PRIVATE KEY-----"
+
+    fun parse(bytes: ByteArray): PrivateKey {
+        val string = String(bytes).replace(PREFIX, "").replace(SUFFIX, "").replace("\n", "")
+        val encoded = Base64.getDecoder().decode(string)
+        val keyFactory = KeyFactory.getInstance("RSA")
+        val keySpec = PKCS8EncodedKeySpec(encoded)
+        return keyFactory.generatePrivate(keySpec)
+    }
+}