Merge pull request #76 from Malinskiy/feature/extract-ktor

feat(adam): extract ktor transport into a separate module

adam/build.gradle.kts

@@ -152,7 +152,6 @@ dependencies {
     implementation(Libraries.annotations)
     implementation(kotlin("stdlib-jdk8", version = Versions.kotlin))
     implementation(Libraries.coroutines)
-    implementation(Libraries.ktorNetwork)
     implementation(Libraries.logging)
     api(Libraries.protobufLite)
     api(Libraries.grpcProtobufLite)

adam/src/integrationTest/kotlin/com/malinskiy/adam/extension/FileExtension.kt

@@ -0,0 +1,42 @@
+/*
+ * Copyright (C) 2022 Anton Malinskiy
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.malinskiy.adam.extension
+
+import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.channels.ReceiveChannel
+import kotlinx.coroutines.channels.produce
+import java.io.File
+
+fun CoroutineScope.sequentialRead(file: File, sizeChannel: ReceiveChannel<Int>): ReceiveChannel<ByteArray> = produce(capacity = 1) {
+    file.inputStream().buffered().use { stream ->
+        var position = 0
+        for (size in sizeChannel) {
+            val buffer = ByteArray(size)
+            val read = stream.read(buffer)
+            if (read == -1) {
+                close()
+                break
+            } else if (read == buffer.size) {
+                send(buffer)
+            } else {
+                send(buffer.copyOf(read))
+            }
+            position += read
+            println("$position/${file.length()}")
+        }
+    }
+}

adam/src/integrationTest/kotlin/com/malinskiy/adam/integration/feature/CmdE2ETest.kt

@@ -19,6 +19,7 @@ package com.malinskiy.adam.integration.feature
 import assertk.assertThat
 import assertk.assertions.contains
 import com.malinskiy.adam.exception.RequestRejectedException
+import com.malinskiy.adam.extension.sequentialRead
 import com.malinskiy.adam.request.Feature
 import com.malinskiy.adam.request.misc.ExecInRequest
 import com.malinskiy.adam.request.pkg.AtomicInstallPackageRequest
@@ -31,7 +32,8 @@ import com.malinskiy.adam.request.pkg.multi.ApkSplitInstallationPackage
 import com.malinskiy.adam.request.pkg.multi.SingleFileInstallationPackage
 import com.malinskiy.adam.rule.AdbDeviceRule
 import com.malinskiy.adam.rule.DeviceType
-import io.ktor.util.cio.readChannel
+import kotlinx.coroutines.channels.Channel
+import kotlinx.coroutines.channels.ReceiveChannel
 import kotlinx.coroutines.delay
 import kotlinx.coroutines.runBlocking
 import org.junit.After
@@ -89,10 +91,14 @@ class CmdE2ETest {
     fun testExecIn() {
         runBlocking {
             val testFile = File(javaClass.getResource("/app-debug.apk").toURI())
+
+            val blockSizeChannel = Channel<Int>(capacity = 1)
+            val channel: ReceiveChannel<ByteArray> = sequentialRead(testFile, blockSizeChannel)
             val success = client.execute(
                 ExecInRequest(
                     "cmd package install -S ${testFile.length()}",
-                    testFile.readChannel()
+                    channel,
+                    blockSizeChannel
                 ),
                 adb.deviceSerial
             )

adam/src/main/kotlin/com/malinskiy/adam/request/misc/ExecInRequest.kt

@@ -16,31 +16,39 @@
 
 package com.malinskiy.adam.request.misc
 
-import com.malinskiy.adam.extension.copyTo
+import com.malinskiy.adam.Const
 import com.malinskiy.adam.extension.readStatus
 import com.malinskiy.adam.request.ComplexRequest
 import com.malinskiy.adam.transport.Socket
-import com.malinskiy.adam.transport.withMaxFilePacketBuffer
-import io.ktor.utils.io.ByteReadChannel
+import kotlinx.coroutines.channels.ReceiveChannel
+import kotlinx.coroutines.channels.SendChannel
 
 /**
  * Executes the command and provides the channel as the input to the command. Does not return anything
  */
-class ExecInRequest(private val cmd: String, private val channel: ByteReadChannel, socketIdleTimeout: Long? = null) :
+class ExecInRequest(
+    private val cmd: String,
+    private val channel: ReceiveChannel<ByteArray>,
+    private val sizeChannel: SendChannel<Int>,
+    socketIdleTimeout: Long? = null
+) :
     ComplexRequest<Unit>(socketIdleTimeout = socketIdleTimeout) {
     override suspend fun readElement(socket: Socket) {
-        withMaxFilePacketBuffer {
-            val buffer = array()
-            while (true) {
-                val available = channel.copyTo(buffer, 0, buffer.size)
-                when {
-                    available > 0 -> socket.writeFully(buffer, 0, available)
-                    else -> break
-                }
+        while (true) {
+            if (!channel.isClosedForReceive) {
+                //Should not request more if read channel is already closed
+                sizeChannel.send(Const.MAX_FILE_PACKET_LENGTH)
+            }
+            val result = channel.tryReceive()
+            when {
+                result.isSuccess && result.getOrThrow().isNotEmpty() -> socket.writeFully(result.getOrThrow(), 0, result.getOrThrow().size)
+                result.isClosed -> break
+                result.isFailure -> continue
+                else -> break
             }
-            //Have to poll
-            socket.readStatus()
         }
+        //Have to poll
+        socket.readStatus()
     }
 
     override fun serialize() = createBaseRequest("exec:$cmd")

adam/src/main/kotlin/com/malinskiy/adam/transport/BufferFactory.kt

@@ -105,4 +105,3 @@ inline fun <R> withMaxFilePacketBuffer(block: ByteBuffer.() -> R): R {
         AdamMaxFilePacketPool.recycle(instance)
     }
 }
-

adam/src/test/kotlin/com/malinskiy/adam/request/misc/ExecInRequestTest.kt

@@ -19,28 +19,18 @@ package com.malinskiy.adam.request.misc
 import assertk.assertThat
 import assertk.assertions.isEqualTo
 import com.malinskiy.adam.extension.toRequestString
-import io.ktor.utils.io.ByteChannel
-import io.ktor.utils.io.cancel
-import org.junit.After
-import org.junit.Before
+import kotlinx.coroutines.channels.Channel
+import kotlinx.coroutines.channels.produce
+import kotlinx.coroutines.runBlocking
 import org.junit.Test
 
 class ExecInRequestTest {
-    lateinit var channel: ByteChannel
-
-    @Before
-    fun prepare() {
-        channel = ByteChannel()
-    }
-
-    @After
-    fun teardown() {
-        channel.cancel()
-    }
-
     @Test
     fun testSerialize() {
-        assertThat(ExecInRequest("cmd package install", channel).serialize().toRequestString())
-            .isEqualTo("0018exec:cmd package install")
+        runBlocking {
+            val receiveChannel = produce<ByteArray> {}
+            assertThat(ExecInRequest("cmd package install", receiveChannel, Channel()).serialize().toRequestString())
+                .isEqualTo("0018exec:cmd package install")
+        }
     }
 }

docs/_docs/shell/shell.md

@@ -59,11 +59,15 @@ launch {
 Executes the command and provides the channel as the input to the command. Does not return anything
 
 ```kotlin
-val testFile = File(javaClass.getResource("/app-debug.apk").toURI())
+val blockSizeChannel = Channel<Int>(capacity = 1)
+//You have to implement the function below for applicable source of data that you have. 
+//Testing code in adam has an example for a file
+val channel: ReceiveChannel<ByteArray> = someFunctionThatProducesByteArrayInResponseToRequestsOverBlockSizeChannel(blockSizeChannel)
 val success = client.execute(
     request = ExecInRequest(
         cmd = "cmd package install -S ${testFile.length()}",
-        channel = testFile.readChannel()
+        channel = testFile.readChannel(),
+        sizeChannel = blockSizeChannel
     ),
     serial = "emulator-5554"
 )

settings.gradle.kts

@@ -21,6 +21,7 @@ pluginManagement {
     }
 }
 include(":adam")
+include(":transport-ktor")
 include(":android-junit4")
 include(":android-junit4-test-annotation-producer")
 include(":android-testrunner-contract")

transport-ktor/build.gradle.kts

@@ -0,0 +1,35 @@
+/*
+ * Copyright (C) 2021 Anton Malinskiy
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+plugins {
+    kotlin("jvm")
+}
+
+java {
+    sourceCompatibility = JavaVersion.VERSION_1_8
+    targetCompatibility = JavaVersion.VERSION_1_8
+}
+
+tasks.withType(org.jetbrains.kotlin.gradle.tasks.KotlinCompile::class) {
+    kotlinOptions.jvmTarget = "1.8"
+    kotlinOptions.apiVersion = "1.5"
+}
+
+dependencies {
+    implementation(project(":adam"))
+    implementation(Libraries.ktorNetwork)
+    implementation(Libraries.logging)
+}