1 files changed, 47 insertions, 7 deletions

diff --git a/app/src/main/java/invalid/lena/scrcpy/ControlStream.java b/app/src/main/java/invalid/lena/scrcpy/ControlStream.java
index eea6c7d..a231b92 100644
--- a/app/src/main/java/invalid/lena/scrcpy/ControlStream.java
+++ b/app/src/main/java/invalid/lena/scrcpy/ControlStream.java
@@ -5,6 +5,7 @@ import java.io.InputStream;
 import java.io.OutputStream;
 import java.nio.charset.StandardCharsets;
 import java.util.concurrent.LinkedBlockingDeque;
+import java.util.concurrent.atomic.AtomicBoolean;
 
 // Bidirectional bridge to the scrcpy control socket.
 //
@@ -36,7 +37,9 @@ public final class ControlStream {
 
     private final InputStream  in;
     private final OutputStream out;
+    private final Runnable     onFatalError;
     private final LinkedBlockingDeque<byte[]> outbox = new LinkedBlockingDeque<>(MAX_QUEUED);
+    private final AtomicBoolean fatalReported = new AtomicBoolean();
 
     private volatile InboundSink sink;
     private Thread writer;
@@ -44,8 +47,13 @@ public final class ControlStream {
     private volatile boolean stop;
 
     public ControlStream(InputStream in, OutputStream out) {
+        this(in, out, null);
+    }
+
+    public ControlStream(InputStream in, OutputStream out, Runnable onFatalError) {
         this.in = in;
         this.out = out;
+        this.onFatalError = onFatalError;
     }
 
     // Wired after construction: the Controller that consumes inbound
@@ -64,8 +72,8 @@ public final class ControlStream {
     public void stop() {
         stop = true;
         outbox.clear();
-        if (writer != null) writer.interrupt();
-        if (reader != null) reader.interrupt();
+        stopThread(writer);
+        stopThread(reader);
     }
 
     public void send(byte[] msg) {
@@ -79,14 +87,24 @@ public final class ControlStream {
                 if (outbox.remove(b)) break;
             }
         }
-        if (!outbox.offerLast(msg)) {
-            Log.w("control: outbox full, dropping msg type=%d", msg.length > 0 ? msg[0] & 0xff : -1);
-        }
+        if (outbox.offerLast(msg)) return;
+
+        // A queue containing only state transitions is unhealthy, but
+        // silently dropping UP/CANCEL/key events leaves input stuck on the
+        // target. Fail the control stream visibly instead.
+        stop = true;
+        if (writer != null) writer.interrupt();
+        if (reader != null) reader.interrupt();
+        try { out.close(); } catch (IOException ignored) {}
+        try { in.close(); } catch (IOException ignored) {}
+        Log.e("control: outbox saturated with non-droppable events");
+        reportFatal();
     }
 
     // ---- writer ----
 
     public void runWriter() {
+        if (writer == null) writer = Thread.currentThread();
         try {
             while (!stop) {
                 byte[] msg = outbox.takeFirst();
@@ -95,7 +113,10 @@ public final class ControlStream {
             }
         } catch (InterruptedException ignored) {
         } catch (IOException e) {
-            if (!stop) Log.e(e, "control writer");
+            if (!stop) {
+                Log.e(e, "control writer");
+                reportFatal();
+            }
         } finally {
             Log.i("control writer: end");
         }
@@ -104,6 +125,7 @@ public final class ControlStream {
     // ---- reader ----
 
     public void runReader() {
+        if (reader == null) reader = Thread.currentThread();
         try {
             byte[] tmp = new byte[12];
             while (!stop) {
@@ -144,11 +166,29 @@ public final class ControlStream {
                 }
             }
         } catch (IOException e) {
-            if (!stop) Log.e(e, "control reader");
+            if (!stop) {
+                Log.e(e, "control reader");
+                reportFatal();
+            }
         } catch (Exception e) {
             Log.e(e, "control reader unexpected");
+            if (!stop) reportFatal();
         } finally {
             Log.i("control reader: end");
         }
     }
+
+    private void reportFatal() {
+        if (onFatalError != null && fatalReported.compareAndSet(false, true)) {
+            onFatalError.run();
+        }
+    }
+
+    private static void stopThread(Thread t) {
+        if (t == null) return;
+        t.interrupt();
+        if (t == Thread.currentThread()) return;
+        try { t.join(1_000); }
+        catch (InterruptedException e) { Thread.currentThread().interrupt(); }
+    }
 }