feat(timeout): add timeout overrides

src/main/kotlin/com/malinskiy/adam/AndroidDebugBridgeClient.kt

@@ -48,7 +48,11 @@ class AndroidDebugBridgeClient(
             val requestSimpleClassName = request.javaClass.simpleName
             throw RequestValidationException("Request $requestSimpleClassName did not pass validation: ${validationResponse.message}")
         }
-        socketFactory.tcp(socketAddress).use { socket ->
+
+        socketFactory.tcp(
+            socketAddress = socketAddress,
+            idleTimeout = request.socketIdleTimeout
+        ).use { socket ->
             serial?.let {
                 SetDeviceRequest(it).handshake(socket)
             }
@@ -63,7 +67,10 @@ class AndroidDebugBridgeClient(
             throw RequestValidationException("Request $requestSimpleClassName did not pass validation: ${validationResponse.message}")
         }
         return scope.produce {
-            socketFactory.tcp(socketAddress).use { socket ->
+            socketFactory.tcp(
+                socketAddress = socketAddress,
+                idleTimeout = request.socketIdleTimeout
+            ).use { socket ->
                 var backChannel = request.channel
 
                 try {
@@ -103,7 +110,10 @@ class AndroidDebugBridgeClient(
     }
 
     suspend fun execute(request: EmulatorCommandRequest): String {
-        socketFactory.tcp(request.address).use { socket ->
+        socketFactory.tcp(
+            socketAddress = request.address,
+            idleTimeout = request.idleTimeoutOverride
+        ).use { socket ->
             return request.process(socket)
         }
     }
@@ -128,15 +138,15 @@ class AndroidDebugBridgeClientFactory {
     var host: InetAddress? = null
     var coroutineContext: CoroutineContext? = null
     var socketFactory: SocketFactory? = null
-    var socketTimeout: Duration? = null
+    var idleTimeout: Duration? = null
 
     fun build(): AndroidDebugBridgeClient {
         return AndroidDebugBridgeClient(
             port = port ?: DiscoverAdbSocketInteractor().execute(),
             host = host ?: InetAddress.getByName(Const.DEFAULT_ADB_HOST),
             socketFactory = socketFactory ?: KtorSocketFactory(
                 coroutineContext = coroutineContext ?: Dispatchers.IO,
-                socketTimeout = socketTimeout?.toMillis() ?: 30_000
+                idleTimeout = idleTimeout?.toMillis() ?: 30_000
             )
         )
     }

src/main/kotlin/com/malinskiy/adam/request/AsyncChannelRequest.kt

@@ -26,8 +26,9 @@ import kotlinx.coroutines.channels.SendChannel
  */
 abstract class AsyncChannelRequest<T : Any?, I : Any?>(
     val channel: ReceiveChannel<I>? = null,
-    target: Target = NonSpecifiedTarget
-) : Request(target) {
+    target: Target = NonSpecifiedTarget,
+    socketIdleTimeout: Long? = null
+) : Request(target, socketIdleTimeout) {
 
     /**
      * Called after the initial OKAY confirmation

src/main/kotlin/com/malinskiy/adam/request/ComplexRequest.kt

@@ -22,7 +22,8 @@ import com.malinskiy.adam.transport.Socket
  * This type of request starts with single serialized request
  * and then proceed to do several reads and writes that have dynamic size
  */
-abstract class ComplexRequest<T : Any?>(target: Target = NonSpecifiedTarget) : Request(target) {
+abstract class ComplexRequest<T : Any?>(target: Target = NonSpecifiedTarget, socketIdleTimeout: Long? = null) :
+    Request(target, socketIdleTimeout) {
     /**
      * Some requests ignore the initial OKAY/FAIL response and instead stream the actual response
      * To implement these we allow overriding this method

src/main/kotlin/com/malinskiy/adam/request/Request.kt

@@ -27,8 +27,13 @@ import java.io.UnsupportedEncodingException
 /**
  * By default all requests are targeted at adb daemon itself
  * @see [Target]
+ *
+ * @param socketIdleTimeout override for socket idle timeout
  */
-open abstract class Request(val target: Target = HostTarget) {
+open abstract class Request(
+    val target: Target = HostTarget,
+    val socketIdleTimeout: Long? = null
+) {
 
     /**
      * Some requests require a device serial to be passed to the request itself by means of <host-prefix>

src/main/kotlin/com/malinskiy/adam/request/device/AsyncDeviceMonitorRequest.kt

@@ -21,16 +21,10 @@ import com.malinskiy.adam.request.AsyncChannelRequest
 import com.malinskiy.adam.request.HostTarget
 import com.malinskiy.adam.transport.Socket
 import kotlinx.coroutines.channels.SendChannel
-import java.net.SocketTimeoutException
 
-class AsyncDeviceMonitorRequest : AsyncChannelRequest<List<Device>, Unit>(target = HostTarget) {
+class AsyncDeviceMonitorRequest : AsyncChannelRequest<List<Device>, Unit>(target = HostTarget, socketIdleTimeout = Long.MAX_VALUE) {
     override suspend fun readElement(socket: Socket, sendChannel: SendChannel<List<Device>>): Boolean {
-        val data = try {
-            socket.readProtocolString()
-        } catch (e: SocketTimeoutException) {
-            //Unfortunately there is no way to check if the socket timeout was cause by no device changes or real network timeout
-            return false
-        }
+        val data = socket.readProtocolString()
 
         sendChannel.send(
             data.lines()

src/main/kotlin/com/malinskiy/adam/request/emu/EmulatorCommandRequest.kt

@@ -40,7 +40,8 @@ class EmulatorCommandRequest(
     private val cmd: String,
     val address: InetSocketAddress,
     private val authToken: String? = null,
-    private val cleanResponse: Boolean = true
+    private val cleanResponse: Boolean = true,
+    val idleTimeoutOverride: Long? = null
 ) {
     private suspend fun readAuthToken(): String? {
         val authTokenFile = File(System.getProperty("user.home"), ".emulator_console_auth_token")

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

@@ -26,7 +26,8 @@ import io.ktor.utils.io.*
 /**
  * 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) : ComplexRequest<Unit>() {
+class ExecInRequest(private val cmd: String, private val channel: ByteReadChannel, socketIdleTimeout: Long? = null) :
+    ComplexRequest<Unit>(socketIdleTimeout = socketIdleTimeout) {
     override suspend fun readElement(socket: Socket) {
         withMaxFilePacketBuffer {
             val buffer = array()

src/main/kotlin/com/malinskiy/adam/request/shell/v1/ChanneledShellCommandRequest.kt

@@ -26,8 +26,9 @@ import kotlinx.coroutines.channels.SendChannel
 
 open class ChanneledShellCommandRequest(
     val cmd: String,
-    target: Target = NonSpecifiedTarget
-) : AsyncChannelRequest<String, Unit>(target = target) {
+    target: Target = NonSpecifiedTarget,
+    socketIdleTimeout: Long? = null
+) : AsyncChannelRequest<String, Unit>(target = target, socketIdleTimeout = socketIdleTimeout) {
 
     override suspend fun readElement(socket: Socket, sendChannel: SendChannel<String>): Boolean {
         withMaxFilePacketBuffer {

src/main/kotlin/com/malinskiy/adam/request/shell/v2/ShellCommandRequest.kt

@@ -16,6 +16,7 @@
 
 package com.malinskiy.adam.request.shell.v2
 
-open class ShellCommandRequest(cmd: String) : SyncShellCommandRequest<ShellCommandResult>(cmd) {
+open class ShellCommandRequest(cmd: String, socketIdleTimeout: Long? = null) :
+    SyncShellCommandRequest<ShellCommandResult>(cmd, socketIdleTimeout = socketIdleTimeout) {
     override fun convertResult(response: ShellCommandResult) = response
 }

src/main/kotlin/com/malinskiy/adam/request/shell/v2/SyncShellCommandRequest.kt

@@ -28,8 +28,8 @@ import io.ktor.utils.io.*
 /**
  * shell v2 service required for this request
  */
-abstract class SyncShellCommandRequest<T : Any?>(val cmd: String, target: Target = NonSpecifiedTarget) :
-    ComplexRequest<T>(target) {
+abstract class SyncShellCommandRequest<T : Any?>(val cmd: String, target: Target = NonSpecifiedTarget, socketIdleTimeout: Long? = null) :
+    ComplexRequest<T>(target, socketIdleTimeout) {
 
     private val stdoutBuilder = StringBuilder()
     private val stderrBuilder = StringBuilder()

src/main/kotlin/com/malinskiy/adam/request/testrunner/TestRunnerRequest.kt

@@ -37,6 +37,7 @@ import kotlinx.coroutines.channels.SendChannel
  * @param userId Specify user instrumentation runs in; current user if not specified
  * @param abi Launch the instrumented process with the selected ABI. This assumes that the process supports the selected ABI.
  * @param profilingOutputPath write profiling data to <FILE>
+ * @param socketIdleTimeout override for socket idle timeout. This should be longer than the longest test
  *
  * @see https://android.googlesource.com/platform/frameworks/base/+/master/cmds/am/src/com/android/commands/am/Am.java#155
  */
@@ -52,7 +53,8 @@ class TestRunnerRequest(
     private val profilingOutputPath: String? = null,
     private val outputLogPath: String? = null,
     private val protobuf: Boolean = false,
-) : AsyncChannelRequest<List<TestEvent>, Unit>() {
+    socketIdleTimeout: Long? = Long.MAX_VALUE
+) : AsyncChannelRequest<List<TestEvent>, Unit>(socketIdleTimeout = socketIdleTimeout) {
 
     private val transformer: ProgressiveResponseTransformer<List<TestEvent>?> by lazy {
         if (protobuf) {

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

@@ -24,15 +24,16 @@ import kotlin.coroutines.CoroutineContext
 
 class KtorSocketFactory(
     coroutineContext: CoroutineContext,
-    private val socketTimeout: Long
+    private val connectTimeout: Long = 10_000,
+    private val idleTimeout: Long = 30_000
 ) : SocketFactory {
     private val selectorManager: SelectorManager = ActorSelectorManager(coroutineContext)
 
-    override suspend fun tcp(socketAddress: InetSocketAddress): Socket {
+    override suspend fun tcp(socketAddress: InetSocketAddress, connectTimeout: Long?, idleTimeout: Long?): Socket {
         return KtorSocket(aSocket(selectorManager)
                               .tcp()
                               .connect(socketAddress) {
-                                  socketTimeout = this@KtorSocketFactory.socketTimeout
+                                  socketTimeout = idleTimeout ?: this@KtorSocketFactory.idleTimeout
                               })
     }
 }

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

@@ -18,12 +18,15 @@ package com.malinskiy.adam.transport
 
 import java.net.InetSocketAddress
 
-class NioSocketFactory : SocketFactory {
-    override suspend fun tcp(socketAddress: InetSocketAddress): Socket {
+class NioSocketFactory(
+    private val connectTimeout: Long = 10_000,
+    private val idleTimeout: Long = 30_000
+) : SocketFactory {
+    override suspend fun tcp(socketAddress: InetSocketAddress, connectTimeout: Long?, idleTimeout: Long?): Socket {
         val nioSocket = NioSocket(
             socketAddress = socketAddress,
-            connectTimeout = 10_000,
-            idleTimeout = 10_000,
+            connectTimeout = connectTimeout ?: this.connectTimeout,
+            idleTimeout = idleTimeout ?: this.idleTimeout,
         )
         nioSocket.connect()
         return nioSocket

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

@@ -19,5 +19,5 @@ package com.malinskiy.adam.transport
 import java.net.InetSocketAddress
 
 interface SocketFactory {
-    suspend fun tcp(socketAddress: InetSocketAddress): Socket
+    suspend fun tcp(socketAddress: InetSocketAddress, connectTimeout: Long? = null, idleTimeout: Long? = null): Socket
 }