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 <#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 @@ private void outputThread(MediaCodec mediaCodec) throws IOException, Interrupted
     }
 
     @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 @@ private void record() throws IOException, AudioCaptureForegroundException {
     }
 
     @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,14 +85,15 @@ private void control() throws IOException {
     }
 
     @Override
-    public void start() {
+    public void start(TerminationListener listener) {
         thread = new Thread(() -> {
             try {
                 control();
             } catch (IOException e) {
                 // 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.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 void onRotationChanged(int rotation) {
         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 @@ private static int chooseMaxSizeFallback(Size failedSize) {
 
     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 @@ private boolean encode(MediaCodec codec, Streamer streamer) throws IOException {
             }
         }
 
-        return !eof;
+        return !eof && alive;
     }
 
     private static MediaCodec createMediaCodec(Codec codec, String encoderName) throws IOException, ConfigurationException {
@@ -267,4 +275,38 @@ private static void setDisplaySurface(IBinder display, Surface surface, int orie
             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 @@
 
 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 @@ private static void scrcpy(Options options) throws IOException, ConfigurationExc
                     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 {
-                // 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);
-                }
-            }
+            completion.await();
         } finally {
-            Ln.d("Screen streaming stopped");
             initThread.interrupt();
             for (AsyncProcessor asyncProcessor : asyncProcessors) {
                 asyncProcessor.stop();