fix(usb): make sure usb + ktor works

Malinskiy · Jan 21, 2021 · 6a94700 · 6a94700
1 parent 9bb8811
commit 6a94700
Show file tree

Hide file tree

Showing 5 changed files with 44 additions and 12 deletions.
diff --git a/src/integrationTest/kotlin/com/malinskiy/adam/integration/ApkE2ETest.kt b/src/integrationTest/kotlin/com/malinskiy/adam/integration/ApkE2ETest.kt
@@ -25,6 +25,7 @@ import com.malinskiy.adam.request.shell.v1.ShellCommandRequest
 import com.malinskiy.adam.request.sync.v1.PushFileRequest
 import com.malinskiy.adam.rule.AdbDeviceRule
 import kotlinx.coroutines.channels.receiveOrNull
+import kotlinx.coroutines.debug.junit4.CoroutinesTimeout
 import kotlinx.coroutines.delay
 import kotlinx.coroutines.runBlocking
 import org.junit.After
@@ -41,6 +42,10 @@ class ApkE2ETest {
     val adb = AdbDeviceRule()
     val client = adb.adb
 
+    @Rule
+    @JvmField
+    val timeout = CoroutinesTimeout.seconds(60)
+
     @Before
     fun setup() {
         runBlocking {

diff --git a/src/main/kotlin/com/malinskiy/adam/extension/ByteReadChannel.kt b/src/main/kotlin/com/malinskiy/adam/extension/ByteReadChannel.kt
@@ -37,4 +37,27 @@ suspend fun ByteReadChannel.copyTo(socket: Socket, buffer: ByteArray): Long {
         }
     }
     return processed
-}
+}
+
+/**
+ * Copies up to limit bytes into transformer using buffer. If limit is null - copy until EOF
+ */
+suspend fun ByteReadChannel.copyTo(buffer: ByteArray, offset: Int, limit: Int): Int {
+    var processed = 0
+    loop@ while (true) {
+        val toRead = (buffer.size - offset) - processed
+        val available = readAvailable(buffer, offset + processed, toRead)
+        when {
+            processed == limit -> break@loop
+            available < 0 && processed != 0 -> {
+                break@loop
+            }
+            available < 0 -> return available
+            available > 0 -> {
+                processed += available
+            }
+            else -> continue@loop
+        }
+    }
+    return processed
+}
diff --git a/src/main/kotlin/com/malinskiy/adam/request/misc/ExecInRequest.kt b/src/main/kotlin/com/malinskiy/adam/request/misc/ExecInRequest.kt
@@ -16,7 +16,6 @@
 
 package com.malinskiy.adam.request.misc
 
-import com.malinskiy.adam.Const
 import com.malinskiy.adam.extension.copyTo
 import com.malinskiy.adam.extension.readStatus
 import com.malinskiy.adam.request.ComplexRequest

diff --git a/src/main/kotlin/com/malinskiy/adam/request/sync/base/BasePushFileRequest.kt b/src/main/kotlin/com/malinskiy/adam/request/sync/base/BasePushFileRequest.kt
@@ -25,7 +25,7 @@ import com.malinskiy.adam.extension.write
 import com.malinskiy.adam.request.AsyncChannelRequest
 import com.malinskiy.adam.request.ValidationResponse
 import com.malinskiy.adam.transport.Socket
-import com.malinskiy.adam.transport.withMaxPacketBuffer
+import com.malinskiy.adam.transport.withMaxFilePacketBuffer
 import io.ktor.util.cio.*
 import io.ktor.utils.io.*
 import kotlinx.coroutines.Dispatchers
@@ -46,10 +46,10 @@ abstract class BasePushFileRequest(
         get() = mode.toInt(8) and "0777".toInt(8)
 
     override suspend fun readElement(socket: Socket, sendChannel: SendChannel<Double>): Boolean {
-        withMaxPacketBuffer {
+        withMaxFilePacketBuffer {
             val data = array()
-            val available = fileReadChannel.copyTo(data, 8, Const.MAX_FILE_PACKET_LENGTH)
-            when {
+            val available = fileReadChannel.copyTo(data, 0, data.size)
+            return when {
                 available < 0 -> {
                     Const.Message.DONE.copyInto(data)
                     (local.lastModified() / 1000).toInt().toByteArray().copyInto(data, destinationOffset = 4)
@@ -59,22 +59,27 @@ abstract class BasePushFileRequest(
 
                     if (transportResponse.okay) {
                         sendChannel.send(1.0)
-                        return true
+                        true
                     } else {
                         throw PushFailedException("adb didn't acknowledge the file transfer: ${transportResponse.message ?: ""}")
                     }
                 }
                 available > 0 -> {
-                    Const.Message.DATA.copyInto(data)
-                    available.toByteArray().reversedArray().copyInto(data, destinationOffset = 4)
+                    socket.writeFully(Const.Message.DATA)
+                    socket.writeFully(available.toByteArray().reversedArray())
                     /**
                      * USB devices are very picky about the size of the DATA buffer. Using the adb's default
                      */
-                    socket.writeFully(data, 0, available + 8)
+                    /**
+                     * USB devices are very picky about the size of the DATA buffer. Using the adb's default
+                     */
+                    socket.writeFully(data, 0, available)
                     currentPosition += available
                     sendChannel.send(currentPosition.toDouble() / totalBytes)
+                    false
                 }
-            else -> null
+                else -> false
+            }
         }
     }
 

diff --git a/src/main/kotlin/com/malinskiy/adam/transport/KtorSocket.kt b/src/main/kotlin/com/malinskiy/adam/transport/KtorSocket.kt
@@ -35,7 +35,7 @@ class KtorSocket(private val ktorSocket: RealKtorSocket) : Socket {
     override suspend fun writeFully(byteBuffer: ByteBuffer) = writeChannel.writeFully(byteBuffer)
     override suspend fun writeFully(toByteArray: ByteArray, offset: Int, limit: Int) = writeChannel.writeFully(toByteArray, offset, limit)
     override suspend fun readAvailable(buffer: ByteArray, offset: Int, limit: Int): Int {
-        if (readChannel.availableForRead == 0) return 0
+        if (!readChannel.isClosedForRead && readChannel.availableForRead == 0) return 0
         return readChannel.readAvailable(buffer, offset, limit)
     }