Convert screen encoder to async processor

Contrary to the other tasks (controller and audio capture/encoding), the
screen encoder was executed synchronously. As a consequence,
scrcpy-server could not terminate until the screen encoder returned.

Convert it to an async processor. This allows to terminate on controller
error, and this paves the way to disable video mirroring.

PR #3978 <https://github.com/Genymobile/scrcpy/pull/3978>

server/src/main/java/com/genymobile/scrcpy/AsyncProcessor.java

@@ -1,7 +1,16 @@
 package com.genymobile.scrcpy;
 
 public interface AsyncProcessor {
-    void start();
+    interface TerminationListener {
+        /**
+         * Notify processor termination
+         *
+         * @param fatalError {@code true} if this must cause the termination of the whole scrcpy-server.
+         */
+        void onTerminated(boolean fatalError);
+    }
+
+    void start(TerminationListener listener);
     void stop();
     void join() throws InterruptedException;
 }

server/src/main/java/com/genymobile/scrcpy/AudioEncoder.java

@@ -115,16 +115,22 @@ public final class AudioEncoder implements AsyncProcessor {
     }
 
     @Override
-    public void start() {
+    public void start(TerminationListener listener) {
         thread = new Thread(() -> {
+            boolean fatalError = false;
             try {
                 encode();
-            } catch (ConfigurationException | AudioCaptureForegroundException e) {
+            } catch (ConfigurationException e) {
+                // Do not print stack trace, a user-friendly error-message has already been logged
+                fatalError = true;
+            } catch (AudioCaptureForegroundException e) {
                 // Do not print stack trace, a user-friendly error-message has already been logged
             } catch (IOException e) {
                 Ln.e("Audio encoding error", e);
+                fatalError = true;
             } finally {
                 Ln.d("Audio encoder stopped");
+                listener.onTerminated(fatalError);
             }
         });
         thread.start();

server/src/main/java/com/genymobile/scrcpy/AudioRawRecorder.java

@@ -54,16 +54,19 @@ public final class AudioRawRecorder implements AsyncProcessor {
     }
 
     @Override
-    public void start() {
+    public void start(TerminationListener listener) {
         thread = new Thread(() -> {
+            boolean fatalError = false;
             try {
                 record();
             } catch (AudioCaptureForegroundException e) {
                 // Do not print stack trace, a user-friendly error-message has already been logged
             } catch (IOException e) {
                 Ln.e("Audio recording error", e);
+                fatalError = true;
             } finally {
                 Ln.d("Audio recorder stopped");
+                listener.onTerminated(fatalError);
             }
         });
         thread.start();

server/src/main/java/com/genymobile/scrcpy/Controller.java

@@ -85,7 +85,7 @@ public class Controller implements AsyncProcessor {
     }
 
     @Override
-    public void start() {
+    public void start(TerminationListener listener) {
         thread = new Thread(() -> {
             try {
                 control();
@@ -93,6 +93,7 @@ public class Controller implements AsyncProcessor {
                 // this is expected on close
             } finally {
                 Ln.d("Controller stopped");
+                listener.onTerminated(true);
             }
         });
         thread.start();

server/src/main/java/com/genymobile/scrcpy/ScreenEncoder.java

@@ -16,7 +16,7 @@ import java.nio.ByteBuffer;
 import java.util.List;
 import java.util.concurrent.atomic.AtomicBoolean;
 
-public class ScreenEncoder implements Device.RotationListener {
+public class ScreenEncoder implements Device.RotationListener, AsyncProcessor {
 
     private static final int DEFAULT_I_FRAME_INTERVAL = 10; // seconds
     private static final int REPEAT_FRAME_DELAY_US = 100_000; // repeat after 100ms
@@ -39,6 +39,9 @@ public class ScreenEncoder implements Device.RotationListener {
     private boolean firstFrameSent;
     private int consecutiveErrors;
 
+    private Thread thread;
+    private final AtomicBoolean stopped = new AtomicBoolean();
+
     public ScreenEncoder(Device device, Streamer streamer, int videoBitRate, int maxFps, List<CodecOption> codecOptions, String encoderName,
             boolean downsizeOnError) {
         this.device = device;
@@ -55,11 +58,11 @@ public class ScreenEncoder implements Device.RotationListener {
         rotationChanged.set(true);
     }
 
-    public boolean consumeRotationChange() {
+    private boolean consumeRotationChange() {
         return rotationChanged.getAndSet(false);
     }
 
-    public void streamScreen() throws IOException, ConfigurationException {
+    private void streamScreen() throws IOException, ConfigurationException {
         Codec codec = streamer.getCodec();
         MediaCodec mediaCodec = createMediaCodec(codec, encoderName);
         MediaFormat format = createFormat(codec.getMimeType(), videoBitRate, maxFps, codecOptions);
@@ -163,9 +166,14 @@ public class ScreenEncoder implements Device.RotationListener {
 
     private boolean encode(MediaCodec codec, Streamer streamer) throws IOException {
         boolean eof = false;
+        boolean alive = true;
         MediaCodec.BufferInfo bufferInfo = new MediaCodec.BufferInfo();
 
         while (!consumeRotationChange() && !eof) {
+            if (stopped.get()) {
+                alive = false;
+                break;
+            }
             int outputBufferId = codec.dequeueOutputBuffer(bufferInfo, -1);
             try {
                 if (consumeRotationChange()) {
@@ -193,7 +201,7 @@ public class ScreenEncoder implements Device.RotationListener {
             }
         }
 
-        return !eof;
+        return !eof && alive;
     }
 
     private static MediaCodec createMediaCodec(Codec codec, String encoderName) throws IOException, ConfigurationException {
@@ -267,4 +275,38 @@ public class ScreenEncoder implements Device.RotationListener {
             SurfaceControl.closeTransaction();
         }
     }
+
+    @Override
+    public void start(TerminationListener listener) {
+        thread = new Thread(() -> {
+            try {
+                streamScreen();
+            } catch (ConfigurationException e) {
+                // Do not print stack trace, a user-friendly error-message has already been logged
+            } catch (IOException e) {
+                // Broken pipe is expected on close, because the socket is closed by the client
+                if (!IO.isBrokenPipe(e)) {
+                    Ln.e("Video encoding error", e);
+                }
+            } finally {
+                Ln.d("Screen streaming stopped");
+                listener.onTerminated(true);
+            }
+        });
+        thread.start();
+    }
+
+    @Override
+    public void stop() {
+        if (thread != null) {
+            stopped.set(true);
+        }
+    }
+
+    @Override
+    public void join() throws InterruptedException {
+        if (thread != null) {
+            thread.join();
+        }
+    }
 }

server/src/main/java/com/genymobile/scrcpy/Server.java

@@ -9,6 +9,35 @@ import java.util.List;
 
 public final class Server {
 
+    private static class Completion {
+        private int running;
+        private boolean fatalError;
+
+        Completion(int running) {
+            this.running = running;
+        }
+
+        synchronized void addCompleted(boolean fatalError) {
+            --running;
+            if (fatalError) {
+                this.fatalError = true;
+            }
+            if (running == 0 || this.fatalError) {
+                notify();
+            }
+        }
+
+        synchronized void await() {
+            try {
+                while (running > 0 && !fatalError) {
+                    wait();
+                }
+            } catch (InterruptedException e) {
+                // ignore
+            }
+        }
+    }
+
     private Server() {
         // not instantiable
     }
@@ -122,22 +151,17 @@ public final class Server {
                     options.getSendFrameMeta());
             ScreenEncoder screenEncoder = new ScreenEncoder(device, videoStreamer, options.getVideoBitRate(), options.getMaxFps(),
                     options.getVideoCodecOptions(), options.getVideoEncoder(), options.getDownsizeOnError());
+            asyncProcessors.add(screenEncoder);
 
+            Completion completion = new Completion(asyncProcessors.size());
             for (AsyncProcessor asyncProcessor : asyncProcessors) {
-                asyncProcessor.start();
+                asyncProcessor.start((fatalError) -> {
+                    completion.addCompleted(fatalError);
+                });
             }
 
-            try {
+            completion.await();
-                // synchronous
-                screenEncoder.streamScreen();
-            } catch (IOException e) {
-                // Broken pipe is expected on close, because the socket is closed by the client
-                if (!IO.isBrokenPipe(e)) {
-                    Ln.e("Video encoding error", e);
-                }
-            }
         } finally {
-            Ln.d("Screen streaming stopped");
             initThread.interrupt();
             for (AsyncProcessor asyncProcessor : asyncProcessors) {
                 asyncProcessor.stop();