fix: implement flow control for adb

app/src/main/java/io/github/miuzarte/scrcpyforandroid/nativecore/DirectAdbClient.kt

@@ -660,9 +660,10 @@ internal class DirectAdbConnection(
      */
     fun openStream(service: String): AdbSocketStream {
         val localId = nextLocalId.getAndIncrement()
-        val stream = AdbSocketStream(localId) { command, arg0, arg1, data ->
+        val flowControlWindow = Storage.appSettings.bundleState.value.adbFlowControlWindow
+        val stream = AdbSocketStream(localId, { command, arg0, arg1, data ->
             sendMsg(command, arg0, arg1, data)
-        }
+        }, flowControlWindow)
         streams[localId] = stream
         sendMsg(A_OPEN, localId, 0, (service + "\u0000").toByteArray(Charsets.UTF_8))
         try {
@@ -959,6 +960,7 @@ internal class DirectAdbConnection(
 class AdbSocketStream(
     val localId: Int,
     private val sender: (cmd: Int, arg0: Int, arg1: Int, `data`: ByteArray) -> Unit,
+    private val flowControlWindow: Int = 0,
 ): Closeable {
 
     companion object {
@@ -976,6 +978,10 @@ class AdbSocketStream(
     private val latchOk = AtomicBoolean(false)
     private val queue = LinkedBlockingQueue<Any>()
 
+    // need notifyAll() / wait()
+    private val writeLock = java.lang.Object()
+    @Volatile private var inflightWrites = 0
+
     private object EndOfStreamMarker
 
     val inputStream: InputStream = InStream()
@@ -987,6 +993,12 @@ class AdbSocketStream(
             latchOk.set(true)
             latch.countDown()
         }
+        if (flowControlWindow > 0) {
+            synchronized(writeLock) {
+                if (inflightWrites > 0) inflightWrites--
+                writeLock.notifyAll()
+            }
+        }
     }
 
     internal fun onData(data: ByteArray) {
@@ -997,6 +1009,9 @@ class AdbSocketStream(
         closed = true
         queue.offer(EndOfStreamMarker)
         latch.countDown()
+        if (flowControlWindow > 0) {
+            synchronized(writeLock) { writeLock.notifyAll() }
+        }
     }
 
     fun awaitOpen(timeoutMs: Long) {
@@ -1016,6 +1031,9 @@ class AdbSocketStream(
             sender(A_CLSE, localId, remoteId, ByteArray(0))
         }
         queue.offer(EndOfStreamMarker)
+        if (flowControlWindow > 0) {
+            synchronized(writeLock) { writeLock.notifyAll() }
+        }
     }
 
     private inner class InStream: InputStream() {
@@ -1054,6 +1072,15 @@ class AdbSocketStream(
         override fun write(b: ByteArray, off: Int, len: Int) {
             if (closed) throw IOException("ADB stream closed")
             if (len == 0) return
+            if (flowControlWindow > 0) {
+                synchronized(writeLock) {
+                    while (inflightWrites >= flowControlWindow && !closed) {
+                        writeLock.wait()
+                    }
+                    if (closed) throw IOException("ADB stream closed")
+                    inflightWrites++
+                }
+            }
             sender(A_WRTE, localId, remoteId, b.copyOfRange(off, off + len))
         }
 

app/src/main/java/io/github/miuzarte/scrcpyforandroid/pages/FileManagerScreen.kt

@@ -629,6 +629,7 @@ private fun PathJumpDialog(
                         onConfirm()
                     },
                     modifier = Modifier.weight(1f),
+                    colors = ButtonDefaults.textButtonColorsPrimary(),
                 )
             }
         }

app/src/main/java/io/github/miuzarte/scrcpyforandroid/pages/FullscreenControlScreen.kt

@@ -71,6 +71,10 @@ fun FullscreenControlScreen(
     val listingsRefreshBusy by scrcpy.listings.refreshBusyState.collectAsState()
     val listingsRefreshVersion by scrcpy.listings.refreshVersionState.collectAsState()
 
+    LaunchedEffect(currentSession) {
+        if (currentSession == null) onBack()
+    }
+
     val asBundleShared by appSettings.bundleState.collectAsState()
     val asBundleSharedLatest by rememberUpdatedState(asBundleShared)
     var asBundle by rememberSaveable(asBundleShared) { mutableStateOf(asBundleShared) }

app/src/main/java/io/github/miuzarte/scrcpyforandroid/pages/SettingsScreen.kt

@@ -1121,6 +1121,30 @@ fun SettingsPage(
                         )
                     },
                 )
+                ArrowSlider(
+                    title = stringResource(R.string.pref_title_adb_flow_control),
+                    summary = stringResource(R.string.pref_summary_adb_flow_control),
+                    value = asBundle.adbFlowControlWindow.toFloat(),
+                    onValueChange = {
+                        asBundle = asBundle.copy(
+                            adbFlowControlWindow = it.roundToInt(),
+                        )
+                    },
+                    valueRange = 0f..4f,
+                    steps = 4 - 0 - 1,
+                    zeroStateText = stringResource(R.string.pref_adb_flow_control_disabled),
+                    displayFormatter = { it.roundToInt().toString() },
+                    inputInitialValue = asBundle.adbFlowControlWindow.toString(),
+                    inputFilter = { input -> input.filter(Char::isDigit) },
+                    inputValueRange = 0f..4f,
+                    onInputConfirm = { input ->
+                        input.toIntOrNull()?.let {
+                            asBundle = asBundle.copy(
+                                adbFlowControlWindow = it.coerceIn(0, 4),
+                            )
+                        }
+                    },
+                )
             }
         }
 

app/src/main/java/io/github/miuzarte/scrcpyforandroid/scrcpy/Scrcpy.kt

@@ -8,6 +8,7 @@ import android.view.KeyEvent
 import androidx.core.net.toUri
 import com.github.promeg.pinyinhelper.Pinyin
 import io.github.miuzarte.scrcpyforandroid.NativeCoreFacade
+import io.github.miuzarte.scrcpyforandroid.R
 import io.github.miuzarte.scrcpyforandroid.constants.Defaults
 import io.github.miuzarte.scrcpyforandroid.nativecore.AdbSocketStream
 import io.github.miuzarte.scrcpyforandroid.nativecore.NativeAdbService
@@ -58,11 +59,13 @@ class Scrcpy(
     private val lowLatency: Boolean = false,
 ) {
 
+    private val backgroundScope = CoroutineScope(SupervisorJob() + Dispatchers.IO)
     private val session = Session(
         ::handleRemoteClipboardText,
         ::updateCurrentSessionSize,
-    )
+    ) {
-    private val backgroundScope = CoroutineScope(SupervisorJob() + Dispatchers.IO)
+        backgroundScope.launch { stop(StopReason.REMOTE_DISCONNECTED) }
+    }
     private val clipboardSyncLock = Any()
     private val clipboardManager by lazy {
         appContext.getSystemService(Context.CLIPBOARD_SERVICE) as? ClipboardManager
@@ -86,6 +89,12 @@ class Scrcpy(
     @Volatile
     private var isRunning: Boolean = false
 
+    enum class StopReason { USER, REMOTE_DISCONNECTED }
+
+    @Volatile
+    @JvmField
+    var lastStopReason: StopReason = StopReason.USER
+
     @Volatile
     @JvmField
     var flexDisplay: Boolean = false
@@ -310,13 +319,14 @@ class Scrcpy(
         }
     }
 
-    suspend fun stop(): Boolean = withContext(Dispatchers.IO) {
+    suspend fun stop(reason: StopReason = StopReason.USER): Boolean = withContext(Dispatchers.IO) {
         if (!isRunning) {
             Log.w(TAG, "stop(): No active session to stop")
             return@withContext false
         }
 
-        Log.i(TAG, "stop(): Stopping scrcpy session")
+        lastStopReason = reason
+        Log.i(TAG, "stop(): Stopping scrcpy session (reason=$reason)")
 
         return@withContext try {
             session.clearVideoConsumer()
@@ -335,6 +345,10 @@ class Scrcpy(
             flexDisplay = false
             _currentSessionState.value = null
             stopClipboardSync()
+            if (reason == StopReason.REMOTE_DISCONNECTED) {
+                logEvent(R.string.vm_session_disconnected, level = Log.WARN)
+                AppRuntime.snackbar(R.string.vm_session_disconnected)
+            }
             Log.i(TAG, "stop(): Session stopped successfully")
             true
         } catch (e: Exception) {
@@ -979,12 +993,16 @@ class Scrcpy(
     class Session(
         private val onRemoteClipboardText: (String) -> Unit,
         private val onVideoSessionSize: (Int, Int) -> Unit,
+        private val onSessionDied: () -> Unit,
     ) {
         private val mutex = Mutex()
 
         @Volatile
         private var activeSession: ActiveSession? = null
 
+        @Volatile
+        private var controlChannelAlive: Boolean = false
+
         private val videoConsumers = linkedSetOf<(VideoPacket) -> Unit>()
 
         @Volatile
@@ -1138,6 +1156,7 @@ class Scrcpy(
                     controlWriter = controlWriter,
                 )
                 activeSession = newSession
+                controlChannelAlive = options.control
                 if (options.control && options.clipboardAutosync)
                     startControlReaderThread(newSession)
 
@@ -1220,6 +1239,7 @@ class Scrcpy(
                         }
                     }
                 } finally {
+                    if (activeSession === session) onSessionDied()
                 }
             }
         }
@@ -1265,6 +1285,7 @@ class Scrcpy(
                         }
                     }
                 } finally {
+                    if (activeSession === session) onSessionDied()
                 }
             }
         }
@@ -1272,19 +1293,11 @@ class Scrcpy(
         suspend fun clearAudioConsumer() = mutex.withLock { audioConsumers.clear() }
 
         suspend fun startApp(name: String) = mutex.withLock {
-            try {
+            withControlWriter("startApp") { startApp(name) }
-                requireControlWriter().startApp(name)
-            } catch (e: IllegalStateException) {
-                Log.w(TAG, "startApp(): control channel not available", e)
-            }
         }
 
         suspend fun resizeDisplay(width: Int, height: Int) = mutex.withLock {
-            try {
+            withControlWriter("resizeDisplay") { resizeDisplay(width, height) }
-                requireControlWriter().resizeDisplay(width, height)
-            } catch (e: IllegalStateException) {
-                Log.w(TAG, "resizeDisplay(): control channel not available", e)
-            }
         }
 
         suspend fun injectKeycode(
@@ -1293,27 +1306,15 @@ class Scrcpy(
             repeat: Int = 0,
             metaState: Int = 0,
         ) = mutex.withLock {
-            try {
+            withControlWriter("injectKeycode") { injectKeycode(action, keycode, repeat, metaState) }
-                requireControlWriter().injectKeycode(action, keycode, repeat, metaState)
-            } catch (e: IllegalStateException) {
-                Log.w(TAG, "injectKeycode(): control channel not available", e)
-            }
         }
 
         suspend fun injectText(text: String) = mutex.withLock {
-            try {
+            withControlWriter("injectText") { injectText(text) }
-                requireControlWriter().injectText(text)
-            } catch (e: IllegalStateException) {
-                Log.w(TAG, "injectText(): control channel not available", e)
-            }
         }
 
         suspend fun setClipboard(text: String, paste: Boolean) = mutex.withLock {
-            try {
+            withControlWriter("setClipboard") { setClipboard(text, paste) }
-                requireControlWriter().setClipboard(text, paste)
-            } catch (e: IllegalStateException) {
-                Log.w(TAG, "setClipboard(): control channel not available", e)
-            }
         }
 
         suspend fun injectTouch(
@@ -1327,8 +1328,8 @@ class Scrcpy(
             actionButton: Int,
             buttons: Int,
         ) = mutex.withLock {
-            try {
+            withControlWriter("injectTouch") {
-                requireControlWriter().injectTouch(
+                injectTouch(
                     action,
                     pointerId,
                     x,
@@ -1339,8 +1340,6 @@ class Scrcpy(
                     actionButton,
                     buttons,
                 )
-            } catch (e: IllegalStateException) {
-                Log.w(TAG, "injectTouch(): control channel not available", e)
             }
         }
 
@@ -1353,8 +1352,8 @@ class Scrcpy(
             vScroll: Float,
             buttons: Int,
         ) = mutex.withLock {
-            try {
+            withControlWriter("injectScroll") {
-                requireControlWriter().injectScroll(
+                injectScroll(
                     x,
                     y,
                     screenWidth,
@@ -1363,25 +1362,15 @@ class Scrcpy(
                     vScroll,
                     buttons,
                 )
-            } catch (e: IllegalStateException) {
-                Log.w(TAG, "injectScroll(): control channel not available", e)
             }
         }
 
         suspend fun pressBackOrTurnScreenOn(action: Int = KeyEvent.ACTION_DOWN) = mutex.withLock {
-            try {
+            withControlWriter("pressBackOrTurnScreenOn") { pressBackOrTurnScreenOn(action) }
-                requireControlWriter().pressBackOrTurnScreenOn(action)
-            } catch (e: IllegalStateException) {
-                Log.w(TAG, "pressBackOrScreenOn(): control channel not available", e)
-            }
         }
 
         suspend fun setDisplayPower(on: Boolean) = mutex.withLock {
-            try {
+            withControlWriter("setDisplayPower") { setDisplayPower(on) }
-                requireControlWriter().setDisplayPower(on)
-            } catch (e: IllegalStateException) {
-                Log.w(TAG, "setDisplayPower(): control channel not available", e)
-            }
         }
 
         suspend fun stop() = mutex.withLock {
@@ -1391,6 +1380,7 @@ class Scrcpy(
         private fun stopInternal() {
             val session = activeSession ?: return
             activeSession = null
+            controlChannelAlive = false
             videoConsumers.clear()
             audioConsumers.clear()
 
@@ -1431,6 +1421,23 @@ class Scrcpy(
                 ?: throw IllegalStateException("scrcpy control channel not available")
         }
 
+        private inline fun withControlWriter(tag: String, block: ControlWriter.() -> Unit) {
+            if (!controlChannelAlive) {
+                Log.w(TAG, "$tag(): control channel not alive")
+                return
+            }
+            try {
+                requireControlWriter().block()
+            } catch (e: IllegalStateException) {
+                controlChannelAlive = false
+                Log.w(TAG, "$tag(): control channel not available", e)
+            } catch (e: IOException) {
+                controlChannelAlive = false
+                Log.w(TAG, "$tag(): control channel I/O failed, marking dead", e)
+                onSessionDied()
+            }
+        }
+
         private fun startControlReaderThread(session: ActiveSession) {
             val controlInput = session.controlInput ?: return
             val controlStream = session.controlStream ?: return
@@ -1478,6 +1485,7 @@ class Scrcpy(
                         }
                     }
                 } finally {
+                    if (activeSession === session) onSessionDied()
                 }
             }
         }

app/src/main/java/io/github/miuzarte/scrcpyforandroid/scrcpy/TouchEventHandler.kt

@@ -5,7 +5,9 @@ import android.view.InputDevice
 import android.view.MotionEvent
 import androidx.compose.ui.geometry.Offset
 import androidx.compose.ui.unit.IntSize
+import kotlinx.coroutines.CancellationException
 import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.Job
 import kotlinx.coroutines.launch
 import kotlin.math.roundToInt
 
@@ -58,6 +60,7 @@ class TouchEventHandler(
     private val eventPositions = HashMap<Int, Offset>(10)
     private val eventPressures = HashMap<Int, Float>(10)
     private val justPressedPointerIds = HashSet<Int>(10)
+    private val pendingMoveJobs = HashMap<Int, Job>(10)
 
     fun handleMotionEvent(event: MotionEvent): Boolean {
         if (touchAreaSize.width == 0 || touchAreaSize.height == 0) {
@@ -263,6 +266,7 @@ class TouchEventHandler(
 
     private fun releasePointer(pointerId: Int, bounds: ContentBounds) {
         if (!activePointerIds.contains(pointerId)) return
+        pendingMoveJobs.remove(pointerId)?.cancel()
         val pos = activePointerPositions[pointerId] ?: Offset.Zero
         val (x, y) = mapToDevice(pos.x, pos.y, bounds)
         coroutineScope.launch {
@@ -370,10 +374,12 @@ class TouchEventHandler(
             activePointerPositions[pointerId] = raw
             val (x, y) = mapToDevice(raw.x, raw.y, bounds)
             activePointerDevicePositions[pointerId] = x to y
-            coroutineScope.launch {
+            pendingMoveJobs[pointerId]?.cancel()
+            pendingMoveJobs[pointerId] = coroutineScope.launch {
                 runCatching {
                     onInjectTouch(UiMotionActions.MOVE, pointerId.toLong(), x, y, pressure, 0, 0)
                 }.onFailure { e ->
+                    if (e is CancellationException) throw e
                     Log.w(
                         FULLSCREEN_TOUCH_LOG_TAG,
                         "handlePointerMove failed for pointerId=$pointerId",

app/src/main/java/io/github/miuzarte/scrcpyforandroid/storage/AppSettings.kt

@@ -261,6 +261,10 @@ class AppSettings(context: Context): Settings(context, "AppSettings") {
             booleanPreferencesKey("adb_auto_load_app_list_on_connect"),
             false,
         )
+        val ADB_FLOW_CONTROL_WINDOW = Pair(
+            intPreferencesKey("adb_flow_control_window"),
+            0,
+        )
 
         // Terminal
         val TERMINAL_FONT_SIZE_SP = Pair(
@@ -349,6 +353,7 @@ class AppSettings(context: Context): Settings(context, "AppSettings") {
         val adbAutoReconnectPairedDevice: Boolean,
         val adbMdnsLanDiscovery: Boolean,
         val adbAutoLoadAppListOnConnect: Boolean,
+        val adbFlowControlWindow: Int,
 
         // Terminal
         val terminalFontSizeSp: Float,
@@ -413,6 +418,7 @@ class AppSettings(context: Context): Settings(context, "AppSettings") {
         bundleField(ADB_AUTO_RECONNECT_PAIRED_DEVICE) { it.adbAutoReconnectPairedDevice },
         bundleField(ADB_MDNS_LAN_DISCOVERY) { it.adbMdnsLanDiscovery },
         bundleField(ADB_AUTO_LOAD_APP_LIST_ON_CONNECT) { it.adbAutoLoadAppListOnConnect },
+        bundleField(ADB_FLOW_CONTROL_WINDOW) { it.adbFlowControlWindow },
 
         // Terminal
         bundleField(TERMINAL_FONT_SIZE_SP) { it.terminalFontSizeSp },
@@ -482,6 +488,7 @@ class AppSettings(context: Context): Settings(context, "AppSettings") {
         adbAutoReconnectPairedDevice = preferences.read(ADB_AUTO_RECONNECT_PAIRED_DEVICE),
         adbMdnsLanDiscovery = preferences.read(ADB_MDNS_LAN_DISCOVERY),
         adbAutoLoadAppListOnConnect = preferences.read(ADB_AUTO_LOAD_APP_LIST_ON_CONNECT),
+        adbFlowControlWindow = preferences.read(ADB_FLOW_CONTROL_WINDOW),
 
         // Terminal
         terminalFontSizeSp = preferences.read(TERMINAL_FONT_SIZE_SP),

app/src/main/res/values-zh/strings.xml

@@ -239,6 +239,9 @@
     <string name="cd_export_adb_key">导出</string>
     <string name="pref_adb_key_reset_confirm_title">重置 ADB 密钥</string>
     <string name="pref_adb_key_reset_confirm_summary">重置后，所有已授权的设备都需要重新授权。\n是否继续？</string>
+    <string name="pref_title_adb_flow_control">ADB 流控窗口</string>
+    <string name="pref_summary_adb_flow_control">1 = 严格流控\n数值越大吞吐越高但可能触发 adbd 流控保护</string>
+    <string name="pref_adb_flow_control_disabled">禁用</string>
     <string name="pref_title_auto_discovery">配对时自动启用发现服务</string>
     <string name="pref_summary_auto_discovery">打开配对弹窗后自动搜索可用配对端口</string>
     <string name="pref_title_auto_reconnect">自动重连已配对设备</string>
@@ -570,6 +573,7 @@
     <string name="vm_legacy_paste_injected">已使用 legacy paste 注入本机剪贴板文本</string>
     <string name="vm_clipboard_synced_paste">已同步本机剪贴板到设备并触发粘贴</string>
     <string name="vm_clipboard_paste_failed">本机剪贴板粘贴失败</string>
+    <string name="vm_session_disconnected">Scrcpy 连接已断开</string>
     <string name="vm_ime_text_failed">输入法文本提交失败：%1$s</string>
     <string name="vm_adb_disconnected_device">ADB 已断开：%1$s</string>
     <string name="vm_mdns_updated">mDNS 发现新端口，已更新快速设备：%1$s:%2$s -> %1$s:%3$s</string>

app/src/main/res/values/strings.xml

@@ -239,6 +239,9 @@
     <string name="cd_export_adb_key">Export</string>
     <string name="pref_adb_key_reset_confirm_title">Reset ADB key</string>
     <string name="pref_adb_key_reset_confirm_summary">After resetting, all previously authorized devices will need to be re-authorized.\nContinue?</string>
+    <string name="pref_title_adb_flow_control">ADB flow control window</string>
+    <string name="pref_summary_adb_flow_control">1 = strict flow control\nHigher values increase throughput but may trigger adbd flow control protection</string>
+    <string name="pref_adb_flow_control_disabled">Disabled</string>
     <string name="pref_title_auto_discovery">Auto enable discovery service when pairing</string>
     <string name="pref_summary_auto_discovery">Automatically search for available pairing ports after opening the pairing dialog</string>
     <string name="pref_title_auto_reconnect">Auto reconnect paired devices</string>
@@ -570,6 +573,7 @@
     <string name="vm_legacy_paste_injected">Used legacy paste to inject local clipboard text</string>
     <string name="vm_clipboard_synced_paste">Synced local clipboard to device and triggered paste</string>
     <string name="vm_clipboard_paste_failed">Local clipboard paste failed</string>
+    <string name="vm_session_disconnected">Scrcpy session disconnected</string>
     <string name="vm_ime_text_failed">IME text submission failed: %1$s</string>
     <string name="vm_adb_disconnected_device">ADB disconnected: %1$s</string>
     <string name="vm_mdns_updated">mDNS discovered new port, updated quick device: %1$s:%2$s -> %1$s:%3$s</string>