Repository: beeware/briefcase
Pull request #1687 (2024-03-08 01:23:46)

Capture emulator `stdout` during startup
## Changes
- An attempt to resolve https://github.com/beeware/briefcase/issues/1573
- One user in Discord has reported that reading `stdout` resolves the issue
- Introduces `PopenOutputStreamer` for encapsulating the output stream `Thread` and adds the ability to capture instead of print output
- Adds `Subprocess.stream_output_non_blocking()` alongside the existing `Subprocess.steam_output()` to facilitate callers streaming (and conditionally capturing output) while doing other tasks

## PR Checklist:
<!--- Go over all the following points, and put an `x` in all the boxes that apply. -->
- [x] All new features have been tested
- [x] All new features have been documented
- [X] I have read the **CONTRIBUTING.md** file
- [X] I will abide by the code of conduct


Patch:
diff --git a/changes/1687.bugfix.rst b/changes/1687.bugfix.rst
new file mode 100644
index 00000000..c0e1d42c
--- /dev/null
+++ b/changes/1687.bugfix.rst
@@ -0,0 +1,1 @@
+On Windows, the Android emulator will always open without needing to press CTRL-C.
diff --git a/src/briefcase/integrations/android_sdk.py b/src/briefcase/integrations/android_sdk.py
index 11a9a7d9..1b18c048 100644
--- a/src/briefcase/integrations/android_sdk.py
+++ b/src/briefcase/integrations/android_sdk.py
@@ -1288,14 +1288,9 @@ In future, you can specify this device by running:
         if extra_args is None:
             extra_args = []
 
+        # Start the emulator
         emulator_popen = self.tools.subprocess.Popen(
-            [
-                os.fsdecode(self.emulator_path),
-                f"@{avd}",
-                "-dns-server",
-                "8.8.8.8",
-            ]
-            + extra_args,
+            [self.emulator_path, f"@{avd}", "-dns-server", "8.8.8.8"] + extra_args,
             env=self.env,
             stdout=subprocess.PIPE,
             stderr=subprocess.STDOUT,
@@ -1303,6 +1298,17 @@ In future, you can specify this device by running:
             start_new_session=True,
         )
 
+        # Start capturing the emulator's output
+        # On Windows, the emulator can block until stdout is read and the emulator will
+        # not actually run until the user sends CTRL+C to Briefcase (#1573). This
+        # avoids that scenario while also ensuring emulator output is always available
+        # to print in the console if other issues occur.
+        emulator_streamer = self.tools.subprocess.stream_output_non_blocking(
+            label="Android emulator",
+            popen_process=emulator_popen,
+            capture_output=True,
+        )
+
         # wrap AVD name in quotes since '@' is a special char in PowerShell
         emulator_command = " ".join(
             f'"{arg}"' if arg.startswith("@") else arg
@@ -1330,12 +1336,11 @@ You can also start the emulator manually by running:
         failed_startup_error_msg = f"{{prologue}}\n{general_error_msg}"
 
         # The boot process happens in 2 phases.
-        # First, the emulator appears in the device list. However, it's
-        # not ready until the boot process has finished. To determine
-        # the boot status, we need the device ID, and an ADB connection.
+        # First, the emulator appears in the device list. However, it's not ready until
+        # the boot process has finished. To determine the boot status, we need the
+        # device ID, and an ADB connection.
 
-        # Phase 1: Wait for the device to appear so we can get an
-        # ADB instance for the new device.
+        # Phase 1: Wait for the device to appear so we can get an ADB instance for it.
         try:
             with self.tools.input.wait_bar("Starting emulator...") as keep_alive:
                 adb = None
@@ -1388,14 +1393,7 @@ You can also start the emulator manually by running:
                 "Emulator output log for startup failure",
                 prefix=self.name,
             )
-            try:
-                # if the emulator exited, this should return its output immediately
-                self.tools.logger.info(emulator_popen.communicate(timeout=1)[0])
-            except subprocess.TimeoutExpired:
-                self.tools.logger.info(
-                    "Briefcase failed to retrieve emulator output "
-                    "(this is expected if the emulator is running)"
-                )
+            self.tools.logger.info(emulator_streamer.captured_output)
 
             # Provide troubleshooting steps if user gives up on the emulator starting
             if isinstance(e, KeyboardInterrupt):
@@ -1412,6 +1410,8 @@ run Briefcase again. The running emulator can then be selected from the list.
                 self.tools.logger.info(general_error_msg)
 
             raise
+        finally:
+            emulator_streamer.request_stop()
 
         # Return the device ID and full name.
         return device, full_name
diff --git a/src/briefcase/integrations/subprocess.py b/src/briefcase/integrations/subprocess.py
index 05f05388..33ddb392 100644
--- a/src/briefcase/integrations/subprocess.py
+++ b/src/briefcase/integrations/subprocess.py
@@ -4,6 +4,7 @@ import contextlib
 import json
 import operator
 import os
+import queue
 import shlex
 import subprocess
 import sys
@@ -56,7 +57,7 @@ def is_process_dead(pid: int) -> bool:
     monitoring of a PID to identify when the process goes from existing to not existing.
 
     :param pid: integer value to be checked if assigned as a PID.
-    :return: True if PID does not exist; False otherwise.
+    :returns: True if PID does not exist; False otherwise.
     """
     return not psutil.pid_exists(pid)
 
@@ -75,7 +76,7 @@ def get_process_id_by_command(
         takes a filepath to a directory for an application; therefore, the actual
         running process will be running a command within that directory.
     :param logger: optional Log to show messages about process matching to users
-    :return: PID if found else None
+    :returns: PID if found else None
     """
     matching_procs = []
     # retrieve command line, creation time, and ID for all running processes.
@@ -116,7 +117,7 @@ def ensure_console_is_safe(sub_method):
 
         :param sub: Bound Subprocess object
         :param args: command line to run in subprocess
-        :return: the return value for the Subprocess method
+        :returns: the return value for the Subprocess method
         """
         # Just run the command if no dynamic elements are active
         if not sub.tools.input.is_console_controlled:
@@ -145,6 +146,121 @@ def ensure_console_is_safe(sub_method):
     return inner
 
 
+class PopenOutputStreamer(threading.Thread):
+    def __init__(
+        self,
+        label: str,
+        popen_process: subprocess.Popen,
+        logger: Log,
+        capture_output: bool = False,
+        filter_func: Callable[[str], Iterator[str]] | None = None,
+    ):
+        """Thread for streaming stdout for a Popen process.
+
+        :param label: Descriptive name for process being streamed
+        :param popen_process: Popen process with stdout to stream
+        :param logger: logger for printing to console
+        :param capture_output: Retain process output in ``output_queue`` via a
+            ``queue.Queue`` instead of printing to console
+        :param filter_func: a callable that will be invoked on every line of output
+            that is streamed; see ``Subprocess.stream_output`` for details
+        """
+        super().__init__(name=f"{label} output streamer", daemon=True)
+
+        self.popen_process = popen_process
+        self.logger = logger
+        self.capture_output = capture_output
+        self.filter_func = filter_func
+
+        # arbitrarily large maxsize to prevent unbounded memory use if things go south
+        self.output_queue = queue.Queue(maxsize=10_000_000)
+        self.stop_flag = threading.Event()
+
+    def run(self):
+        """Stream output for a Popen process."""
+        try:
+            while output_line := self._readline():
+                # The stop_flag is intentionally checked both at the top and bottom of
+                # this loop; if the flag was set during the call to readline(), then
+                # processing the output is skipped altogether. And if the flag is set
+                # as a consequence of filter_func(), the streamer still exits before
+                # calling readline() again and potentially blocking indefinitely.
+                if not self.stop_flag.is_set():
+                    filtered_output, stop_streaming = self._filter(output_line)
+
+                    for filtered_line in filtered_output:
+                        if self.capture_output:
+                            self.output_queue.put_nowait(filtered_line)
+                        else:
+                            self.logger.info(filtered_line)
+
+                    if stop_streaming:
+                        self.stop_flag.set()
+
+                if self.stop_flag.is_set():
+                    break
+        except Exception as e:
+            self.logger.error(f"Error while streaming output: {type(e).__name__}: {e}")
+            self.logger.capture_stacktrace("Output thread")
+
+    def request_stop(self):
+        """Set the stop flag to cause the streamer to exit.
+
+        If the streamer is currently blocking on `readline()` because the process'
+        stdout buffer is empty, then the streamer will not exit until `readline()`
+        returns or until Briefcase exits.
+        """
+        self.stop_flag.set()
+
+    @property
+    def captured_output(self) -> str:
+        """The captured output from the process."""
+        output = []
+        while not self.output_queue.empty():
+            with contextlib.suppress(queue.Empty):
+                output.append(self.output_queue.get_nowait())
+                self.output_queue.task_done()
+        return "".join(output)
+
+    def _readline(self) -> str:
+        """Read a line of output from the process while blocking.
+
+        Calling readline() for stdout always returns at least a newline, i.e. "\n",
+        UNLESS the process is exiting or already exited; in that case, an empty string
+        is returned.
+
+        :returns: one line of output or "" if nothing more can be read from stdout
+        """
+        try:
+            return ensure_str(self.popen_process.stdout.readline())
+        except ValueError as e:
+            # Catch ValueError if stdout is unexpectedly closed; this can
+            # happen, for instance, if the user starts spamming CTRL+C.
+            if "I/O operation on closed file" in str(e):
+                self.logger.warning(
+                    "WARNING: stdout was unexpectedly closed while streaming output"
+                )
+                return ""
+            else:
+                raise
+
+    def _filter(self, line: str) -> tuple[list[str], bool]:
+        """Run filter function over output from process."""
+        filtered_output = []
+        stop_streaming = False
+
+        if self.filter_func is not None:
+            try:
+                for filtered_line in self.filter_func(line.strip("\n")):
+                    filtered_output.append(filtered_line)
+            except StopStreaming:
+                stop_streaming = True
+        else:
+            filtered_output.append(line)
+
+        return filtered_output, stop_streaming
+
+
 class NativeAppContext(Tool):
     """A wrapper around subprocess for use as an app-bound tool."""
 
@@ -409,7 +525,7 @@ class Subprocess(Tool):
             user. Can raise StopStreaming to terminate the output stream.
         :param kwargs: keyword args for ``subprocess.run()``
         :raises ValueError: if a filter function is provided when in non-streaming mode.
-        :return: ``CompletedProcess`` for invoked process
+        :returns: ``CompletedProcess`` for invoked process
         """
 
         # Stream the output unless the caller explicitly disables it. When a
@@ -617,19 +733,19 @@ class Subprocess(Tool):
         :param label: A description of the content being streamed; used for to provide
             context in logging messages.
         :param popen_process: a running Popen process with output to print
-        :param stop_func: a Callable that returns True when output streaming should stop
+        :param stop_func: A Callable that returns True when output streaming should stop
             and the popen_process should be terminated.
-        :param filter_func: a callable that will be invoked on every line of output that
+        :param filter_func: A callable that will be invoked on every line of output that
             is streamed. The function accepts the "raw" line of input (stripped of any
             trailing newline); it returns a generator that yields the filtered output
             that should be displayed to the user. Can raise StopStreaming to terminate
             the output stream.
         """
-        output_streamer = threading.Thread(
-            name=f"{label} output streamer",
-            target=self._stream_output_thread,
-            args=(popen_process, filter_func),
-            daemon=True,
+        output_streamer = PopenOutputStreamer(
+            label=label,
+            popen_process=popen_process,
+            logger=self.tools.logger,
+            filter_func=filter_func,
         )
         try:
             output_streamer.start()
@@ -653,52 +769,39 @@ class Subprocess(Tool):
                     "Log stream hasn't terminated; log output may be corrupted."
                 )
 
-    def _stream_output_thread(
+    def stream_output_non_blocking(
         self,
-        popen_process: subprocess.Popen,
-        filter_func: Callable[[str], Iterator[str]],
-    ):
-        """Stream output for a Popen process in a Thread.
+        label: str,
+        popen_process: Popen,
+        capture_output: bool = False,
+        filter_func: Callable[[str], Iterator[str]] | None = None,
+    ) -> PopenOutputStreamer:
+        """Stream the output of a Popen process without blocking.
+
+        This is useful for streaming or capturing the output of a process in the
+        background. In this way, the process' output can be shown to users while the
+        main thread monitors other activities; alternatively, the output of the process
+        can be captured to be retrieved later in the event of an error, for instance.
 
-        :param popen_process: popen process to stream stdout
-        :param filter_func: a callable that will be invoked on every line
-            of output that is streamed; see ``stream_output`` for details.
+        :param label: A description of the content being streamed; used for to provide
+            context in logging messages.
+        :param popen_process: A running Popen process with output to print
+        :param capture_output: Retain process output instead of printing to the console
+        :param filter_func: A callable that will be invoked on every line of output that
+            is streamed. The function accepts the "raw" line of input (stripped of any
+            trailing newline); it returns a generator that yields the filtered output
+            that should be displayed to the user. Can raise StopStreaming to terminate
+            the output stream.
         """
-        try:
-            while True:
-                try:
-                    output_line = ensure_str(popen_process.stdout.readline())
-                except ValueError as e:
-                    # Catch ValueError if stdout is unexpectedly closed; this can
-                    # happen, for instance, if the user starts spamming CTRL+C.
-                    if "I/O operation on closed file" in str(e):
-                        self.tools.logger.warning(
-                            "WARNING: stdout was unexpectedly closed while streaming output"
-                        )
-                        return
-                    else:
-                        raise
-
-                # readline should always return at least a newline (ie \n) UNLESS
-                # the underlying process is exiting/gone; then "" is returned.
-                if output_line:
-                    if filter_func is not None:
-                        try:
-                            for filtered_output in filter_func(
-                                output_line.rstrip("\n")
-                            ):
-                                self.tools.logger.info(filtered_output)
-                        except StopStreaming:
-                            return
-                    else:
-                        self.tools.logger.info(output_line)
-                else:
-                    return
-        except Exception as e:
-            self.tools.logger.error(
-                f"Error while streaming output: {e.__class__.__name__}: {e}"
-            )
-            self.tools.logger.capture_stacktrace("Output thread")
+        output_streamer = PopenOutputStreamer(
+            label=label,
+            popen_process=popen_process,
+            logger=self.tools.logger,
+            capture_output=capture_output,
+            filter_func=filter_func,
+        )
+        output_streamer.start()
+        return output_streamer
 
     def cleanup(self, label: str, popen_process: subprocess.Popen):
         """Clean up after a Popen process, gracefully terminating if possible; forcibly


Test patch:
diff --git a/tests/integrations/android_sdk/AndroidSDK/test_start_emulator.py b/tests/integrations/android_sdk/AndroidSDK/test_start_emulator.py
index a0339daf..096e9364 100644
--- a/tests/integrations/android_sdk/AndroidSDK/test_start_emulator.py
+++ b/tests/integrations/android_sdk/AndroidSDK/test_start_emulator.py
@@ -1,4 +1,3 @@
-import os
 import subprocess
 from unittest.mock import MagicMock, call
 
@@ -109,7 +108,7 @@ def test_start_emulator(mock_tools, android_sdk):
     # The process was started.
     mock_tools.subprocess.Popen.assert_called_with(
         [
-            os.fsdecode(android_sdk.emulator_path),
+            android_sdk.emulator_path,
             "@idleEmulator",
             "-dns-server",
             "8.8.8.8",
@@ -121,6 +120,12 @@ def test_start_emulator(mock_tools, android_sdk):
         start_new_session=True,
     )
 
+    mock_tools.subprocess.stream_output_non_blocking.assert_called_with(
+        label="Android emulator",
+        popen_process=emu_popen,
+        capture_output=True,
+    )
+
     # There were 5 calls to run
     android_sdk.mock_run.assert_has_calls(
         [
@@ -190,7 +195,7 @@ def test_start_emulator_fast_start(mock_tools, android_sdk):
     # The process was started.
     mock_tools.subprocess.Popen.assert_called_with(
         [
-            os.fsdecode(android_sdk.emulator_path),
+            android_sdk.emulator_path,
             "@idleEmulator",
             "-dns-server",
             "8.8.8.8",
@@ -202,6 +207,12 @@ def test_start_emulator_fast_start(mock_tools, android_sdk):
         start_new_session=True,
     )
 
+    mock_tools.subprocess.stream_output_non_blocking.assert_called_with(
+        label="Android emulator",
+        popen_process=emu_popen,
+        capture_output=True,
+    )
+
     # There were 3 calls to run
     android_sdk.mock_run.assert_has_calls(
         [
@@ -285,7 +296,7 @@ def test_emulator_fail_to_start(mock_tools, android_sdk):
     # The process was started.
     mock_tools.subprocess.Popen.assert_called_with(
         [
-            os.fsdecode(android_sdk.emulator_path),
+            android_sdk.emulator_path,
             "@idleEmulator",
             "-dns-server",
             "8.8.8.8",
@@ -297,6 +308,12 @@ def test_emulator_fail_to_start(mock_tools, android_sdk):
         start_new_session=True,
     )
 
+    mock_tools.subprocess.stream_output_non_blocking.assert_called_with(
+        label="Android emulator",
+        popen_process=emu_popen,
+        capture_output=True,
+    )
+
     # There were 2 calls to run, both to get AVD name
     android_sdk.mock_run.assert_has_calls(
         [
@@ -389,7 +406,7 @@ def test_emulator_fail_to_boot(mock_tools, android_sdk):
     # The process was started.
     mock_tools.subprocess.Popen.assert_called_with(
         [
-            os.fsdecode(android_sdk.emulator_path),
+            android_sdk.emulator_path,
             "@idleEmulator",
             "-dns-server",
             "8.8.8.8",
@@ -401,6 +418,12 @@ def test_emulator_fail_to_boot(mock_tools, android_sdk):
         start_new_session=True,
     )
 
+    mock_tools.subprocess.stream_output_non_blocking.assert_called_with(
+        label="Android emulator",
+        popen_process=emu_popen,
+        capture_output=True,
+    )
+
     # There were 6 calls to run before failure
     android_sdk.mock_run.assert_has_calls(
         [
@@ -423,20 +446,7 @@ def test_emulator_fail_to_boot(mock_tools, android_sdk):
     assert "Android emulator was unable to boot!" in exc_info.value.msg
 
 
-@pytest.mark.parametrize(
-    "emulator_comm, expected_log",
-    [
-        (
-            ("This is stdout", "this is stderr"),
-            "Emulator output log for startup failure\nThis is stdout",
-        ),
-        (
-            subprocess.TimeoutExpired(cmd="emulator", timeout=1),
-            "Emulator output log for startup failure\nBriefcase failed",
-        ),
-    ],
-)
-def test_emulator_ctrl_c(mock_tools, android_sdk, emulator_comm, expected_log, capsys):
+def test_emulator_ctrl_c(mock_tools, android_sdk, capsys):
     """If emulator startup is interrupted by the user, an error is displayed."""
     # Short circuit device loop by returning no devices
     android_sdk.devices = MagicMock(side_effect=[{}, {}])
@@ -449,7 +459,11 @@ def test_emulator_ctrl_c(mock_tools, android_sdk, emulator_comm, expected_log, c
     mock_tools.subprocess.Popen.return_value = emu_popen
 
     # Mock emulator output
-    emu_popen.communicate.side_effect = [emulator_comm]
+    emu_streamer = MagicMock()
+    mock_tools.subprocess.stream_output_non_blocking = MagicMock(
+        return_value=emu_streamer
+    )
+    emu_streamer.captured_output = "This is stdout"
 
     # Start the emulator
     with pytest.raises(KeyboardInterrupt):
@@ -458,7 +472,7 @@ def test_emulator_ctrl_c(mock_tools, android_sdk, emulator_comm, expected_log, c
     # The process was started.
     mock_tools.subprocess.Popen.assert_called_with(
         [
-            os.fsdecode(android_sdk.emulator_path),
+            android_sdk.emulator_path,
             "@idleEmulator",
             "-dns-server",
             "8.8.8.8",
@@ -470,13 +484,19 @@ def test_emulator_ctrl_c(mock_tools, android_sdk, emulator_comm, expected_log, c
         start_new_session=True,
     )
 
+    mock_tools.subprocess.stream_output_non_blocking.assert_called_with(
+        label="Android emulator",
+        popen_process=emu_popen,
+        capture_output=True,
+    )
+
     # Took a total of 2 naps before KeyboardInterrupt.
     assert android_sdk.sleep.call_count == 2
 
     output = capsys.readouterr().out
 
     # Emulator's log was printed.
-    assert expected_log in output
+    assert "Emulator output log for startup failure\nThis is stdout" in output
 
     # Expected error message was printed.
     assert "Is the Android emulator not starting up properly?" in output
@@ -524,7 +544,7 @@ def test_start_emulator_extra_args(mock_tools, android_sdk):
     # The process was started with the headless test options.
     mock_tools.subprocess.Popen.assert_called_with(
         [
-            os.fsdecode(android_sdk.emulator_path),
+            android_sdk.emulator_path,
             "@idleEmulator",
             "-dns-server",
             "8.8.8.8",
@@ -538,6 +558,12 @@ def test_start_emulator_extra_args(mock_tools, android_sdk):
         start_new_session=True,
     )
 
+    mock_tools.subprocess.stream_output_non_blocking.assert_called_with(
+        label="Android emulator",
+        popen_process=emu_popen,
+        capture_output=True,
+    )
+
     # There were 2 calls to run
     android_sdk.mock_run.assert_has_calls(
         [
diff --git a/tests/integrations/subprocess/test_PopenOutputStreamer.py b/tests/integrations/subprocess/test_PopenOutputStreamer.py
new file mode 100644
index 00000000..5764b79c
--- /dev/null
+++ b/tests/integrations/subprocess/test_PopenOutputStreamer.py
@@ -0,0 +1,333 @@
+import threading
+from io import StringIO
+
+import pytest
+
+from briefcase.console import Log
+from briefcase.integrations import subprocess
+from briefcase.integrations.subprocess import PopenOutputStreamer
+
+
+@pytest.fixture
+def streamer(streaming_process):
+    return PopenOutputStreamer(
+        label="test",
+        popen_process=streaming_process,
+        logger=Log(),
+    )
+
+
+def test_output(streamer, capsys):
+    """Process output is printed."""
+    streamer.start()
+    streamer.join(timeout=5)
+
+    # fmt: off
+    assert capsys.readouterr().out == (
+        "output line 1\n"
+        "\n"
+        "output line 3\n"
+    )
+    # fmt: on
+
+
+def test_stdout_closes_unexpectedly(streamer, monkeypatch, capsys):
+    """Streamer exits from ValueError because stdout was closed."""
+
+    def monkeypatch_ensure_str(value):
+        """Close stdout when ensure_str() runs on output from readline()."""
+        streamer.popen_process.stdout.close()
+        return value
+
+    monkeypatch.setattr(subprocess, "ensure_str", monkeypatch_ensure_str)
+    streamer.popen_process.stdout = StringIO("output line 1\noutput line 2")
+
+    streamer.start()
+    streamer.join(timeout=5)
+
+    assert capsys.readouterr().out == (
+        "output line 1\n"
+        "WARNING: stdout was unexpectedly closed while streaming output\n"
+    )
+
+
+def test_readline_raises_exception(streamer, monkeypatch, capsys):
+    """Streamer aborts if readline() raises ValueError for reasons other than stdout
+    closing."""
+
+    def monkeypatch_ensure_str(value):
+        """Simulate readline() raising an ValueError-derived exception."""
+        raise UnicodeError("readline() exception")
+
+    monkeypatch.setattr(subprocess, "ensure_str", monkeypatch_ensure_str)
+
+    streamer.start()
+    streamer.join(timeout=5)
+
+    assert capsys.readouterr().out == (
+        "Error while streaming output: UnicodeError: readline() exception\n"
+    )
+
+
+def test_request_stop(streamer, capsys):
+    """Requesting the streamer to stop sets the stop flag."""
+    streamer.start()
+    streamer.join(timeout=5)
+
+    # fmt: off
+    assert capsys.readouterr().out == (
+        "output line 1\n"
+        "\n"
+        "output line 3\n"
+    )
+    # fmt: on
+
+    assert not streamer.stop_flag.is_set()
+
+    streamer.request_stop()
+
+    assert streamer.stop_flag.is_set()
+
+
+def test_request_stop_set_immediately(streamer, capsys):
+    """Nothing is printed if stop flag is immediately set."""
+    streamer.request_stop()
+
+    streamer.start()
+    streamer.join(timeout=5)
+
+    assert capsys.readouterr().out == ""
+
+
+def test_request_stop_set_during_output(streamer, monkeypatch, capsys):
+    """Streamer prints nothing more after stop flag is set."""
+
+    def filter_func(value):
+        """Simulate stop flag set while output is being read."""
+        streamer.request_stop()
+        yield value
+
+    streamer.filter_func = filter_func
+
+    streamer.start()
+    streamer.join(timeout=5)
+
+    assert capsys.readouterr().out == "output line 1\n"
+
+
+def test_captured_output(streamer, capsys):
+    """Captured output is available after streaming."""
+    streamer.capture_output = True
+
+    streamer.start()
+    streamer.join(timeout=5)
+
+    assert capsys.readouterr().out == ""
+
+    # fmt: off
+    assert streamer.captured_output == (
+        "output line 1\n"
+        "\n"
+        "output line 3\n"
+    )
+    # fmt: on
+
+
+def test_captured_output_interim(streamer, monkeypatch, capsys):
+    """Captured output is available during streaming."""
+    streamer.capture_output = True
+
+    streamer_is_waiting = threading.Event()
+    continue_streaming = threading.Event()
+
+    def mock_readline():
+        """Wait on asserts while returning output lines."""
+        yield "output line 1\n"
+        yield "output line 2\n"
+        streamer_is_waiting.set()
+        continue_streaming.wait(timeout=5)
+        yield "output line 3\n"
+        yield "output line 4\n"
+        yield ""
+
+    streamer.popen_process.stdout.readline.side_effect = mock_readline()
+
+    streamer.start()
+    streamer_is_waiting.wait(timeout=5)
+
+    assert streamer.captured_output == "output line 1\noutput line 2\n"
+
+    continue_streaming.set()
+    streamer.join(timeout=5)
+
+    assert streamer.captured_output == "output line 3\noutput line 4\n"
+
+
+def test_filter_func(streamer, capsys):
+    """A filter can be added to modify an output stream."""
+
+    # Define a filter function that converts "output" into "filtered"
+    def filter_func(line):
+        yield line.replace("output", "filtered")
+
+    streamer.filter_func = filter_func
+
+    streamer.start()
+    streamer.join(timeout=5)
+
+    # fmt: off
+    # Output has been transformed
+    assert capsys.readouterr().out == (
+        "filtered line 1\n"
+        "\n"
+        "filtered line 3\n"
+    )
+    # fmt: on
+
+
+def test_filter_func_reject(streamer, capsys):
+    """A filter that rejects lines can be added to modify an output stream."""
+
+    # Define a filter function that ignores blank lines
+    def filter_func(line):
+        if len(line) == 0:
+            return
+        yield line
+
+    streamer.filter_func = filter_func
+
+    streamer.start()
+    streamer.join(timeout=5)
+
+    # fmt: off
+    # Output has been transformed
+    assert capsys.readouterr().out == (
+        "output line 1\n"
+        "output line 3\n"
+    )
+    # fmt: on
+
+
+def test_filter_func_line_ends(streamer, sleep_zero, capsys):
+    """Filter functions are not provided the newline."""
+
+    # Define a filter function that redacts lines that end with 1
+    # The newline is *not* included.
+    def filter_func(line):
+        if line.endswith("line 1"):
+            yield line.replace("line 1", "**REDACTED**")
+        else:
+            yield line
+
+    streamer.filter_func = filter_func
+
+    streamer.start()
+    streamer.join(timeout=5)
+
+    # fmt: off
+    # Output has been transformed; newline exists in output
+    assert capsys.readouterr().out == (
+        "output **REDACTED**\n"
+        "\n"
+        "output line 3\n"
+    )
+    # fmt: on
+
+
+def test_filter_func_line_multiple_output(streamer, capsys):
+    """Filter functions can generate multiple lines from a single input."""
+
+    # Define a filter function that adds an extra line of content when the
+    # lines that end with 1
+    def filter_func(line):
+        yield line
+        if line.endswith("line 1"):
+            yield "Extra content!"
+
+    streamer.filter_func = filter_func
+
+    streamer.start()
+    streamer.join(timeout=5)
+
+    # fmt: off
+    # Output has been transformed; newline exists in output
+    assert capsys.readouterr().out == (
+        "output line 1\n"
+        "Extra content!\n"
+        "\n"
+        "output line 3\n"
+    )
+    # fmt: on
+
+
+def test_filter_func_stop_iteration(streamer, capsys):
+    """A filter can indicate that logging should stop."""
+
+    # Define a filter function that converts "output" into "filtered",
+    # and terminates streaming when a blank line is seen.
+    def filter_func(line):
+        if line == "":
+            raise subprocess.StopStreaming()
+        yield line.replace("output", "filtered")
+
+    streamer.filter_func = filter_func
+
+    streamer.start()
+    streamer.join(timeout=5)
+
+    # fmt: off
+    # Output has been transformed, but is truncated when the empty line was received.
+    assert capsys.readouterr().out == (
+        "filtered line 1\n"
+    )
+    # fmt: on
+
+
+def test_filter_func_output_and_stop_iteration(streamer, capsys):
+    """A filter can indicate that logging should stop, and also output content."""
+
+    # Define a filter function that converts "output" into "filtered",
+    # and terminates streaming when a blank line is seen; but outputs
+    # one more line before terminating.
+    def filter_func(line):
+        if line == "":
+            yield "This should be the last line"
+            raise subprocess.StopStreaming()
+        yield line.replace("output", "filtered")
+
+    streamer.filter_func = filter_func
+
+    streamer.start()
+    streamer.join(timeout=5)
+
+    # fmt: off
+    # Output has been transformed, but is truncated when the empty line was received.
+    assert capsys.readouterr().out == (
+        "filtered line 1\n"
+        "This should be the last line\n"
+    )
+    # fmt: on
+
+
+def test_filter_func_line_unexpected_error(streamer, capsys):
+    """If a filter function fails, the error is caught and logged."""
+
+    # Define a filter function that raises a RunTimeError
+    # The newline is *not* included.
+    def filter_func(line):
+        if not line:
+            raise RuntimeError("Like something totally went wrong")
+        yield line
+
+    streamer.filter_func = filter_func
+
+    streamer.start()
+    streamer.join(timeout=5)
+
+    # fmt: off
+    # Exception
+    assert capsys.readouterr().out == (
+        "output line 1\n"
+        "Error while streaming output: RuntimeError: Like something totally went wrong\n"
+    )
+    # fmt: on
diff --git a/tests/integrations/subprocess/test_Subprocess__stream_output.py b/tests/integrations/subprocess/test_Subprocess__stream_output.py
index feb7c951..7dbcfcbf 100644
--- a/tests/integrations/subprocess/test_Subprocess__stream_output.py
+++ b/tests/integrations/subprocess/test_Subprocess__stream_output.py
@@ -1,5 +1,4 @@
 import time
-from io import StringIO
 from threading import Event
 from unittest import mock
 
@@ -37,13 +36,12 @@ def test_output_debug(mock_sub, streaming_process, sleep_zero, capsys):
     mock_sub.stream_output("testing", streaming_process)
 
     # fmt: off
-    expected_output = (
+    assert capsys.readouterr().out == (
         "output line 1\n"
         "\n"
         "output line 3\n"
     )
     # fmt: on
-    assert capsys.readouterr().out == expected_output
 
     mock_sub.cleanup.assert_called_once_with("testing", streaming_process)
 
@@ -58,12 +56,14 @@ def test_keyboard_interrupt(mock_sub, streaming_process, capsys):
     with pytest.raises(KeyboardInterrupt):
         mock_sub.stream_output("testing", streaming_process, stop_func=send_ctrl_c)
 
-    assert (
-        capsys.readouterr().out == "output line 1\n"
+    # fmt: off
+    assert capsys.readouterr().out == (
+        "output line 1\n"
         "\n"
         "output line 3\n"
         "Stopping...\n"
     )
+    # fmt: on
     mock_sub.cleanup.assert_called_once_with("testing", streaming_process)
 
 
@@ -77,6 +77,7 @@ def test_process_exit_with_queued_output(
     streaming_process.poll.side_effect = [None, -3, -3, -3]
 
     mock_sub.stream_output("testing", streaming_process)
+
     # fmt: off
     assert capsys.readouterr().out == (
         "output line 1\n"
@@ -93,6 +94,7 @@ def test_stop_func(mock_sub, streaming_process, stop_func_ret_val, sleep_zero, c
     mock_sub.stream_output(
         "testing", streaming_process, stop_func=lambda: stop_func_ret_val
     )
+
     # fmt: off
     assert capsys.readouterr().out == (
         "output line 1\n"
@@ -128,8 +130,8 @@ def test_stuck_streamer(mock_sub, streaming_process, sleep_zero, monkeypatch, ca
         monkeypatched_streamer_exited.set()
 
     monkeypatch.setattr(
-        mock_sub,
-        "_stream_output_thread",
+        subprocess.PopenOutputStreamer,
+        "run",
         monkeypatched_blocked_streamer,
     )
 
@@ -152,208 +154,3 @@ def test_stuck_streamer(mock_sub, streaming_process, sleep_zero, monkeypatch, ca
         "Log stream hasn't terminated; log output may be corrupted.\n"
     )
     # fmt: on
-
-
-def test_stdout_closes_unexpectedly(mock_sub, streaming_process, monkeypatch, capsys):
-    """Streamer exits from ValueError because stdout was closed."""
-
-    def monkeypatch_ensure_str(value):
-        """Close stdout when ensure_str() runs on output from readline()."""
-        streaming_process.stdout.close()
-        return value
-
-    streaming_process.stdout = StringIO(initial_value="output line 1\noutput line 2")
-    monkeypatch.setattr(subprocess, "ensure_str", monkeypatch_ensure_str)
-
-    mock_sub.stream_output("testing", streaming_process)
-
-    assert capsys.readouterr().out == (
-        "output line 1\n"
-        "WARNING: stdout was unexpectedly closed while streaming output\n"
-    )
-
-
-def test_readline_raises_exception(mock_sub, streaming_process, monkeypatch, capsys):
-    """Streamer aborts if readline() raises ValueError for reasons other than stdout
-    closing."""
-
-    def monkeypatch_ensure_str(value):
-        """Simulate readline() raising an ValueError-derived exception."""
-        raise UnicodeError("readline() exception")
-
-    streaming_process.stdout = StringIO(initial_value="output line 1\noutput line 2")
-    monkeypatch.setattr(subprocess, "ensure_str", monkeypatch_ensure_str)
-
-    mock_sub.stream_output("testing", streaming_process)
-
-    assert capsys.readouterr().out == (
-        "Error while streaming output: UnicodeError: readline() exception\n"
-    )
-
-
-def test_filter_func(mock_sub, streaming_process, sleep_zero, capsys):
-    """A filter can be added to modify an output stream."""
-
-    # Define a filter function that converts "output" into "filtered"
-    def filter_func(line):
-        yield line.replace("output", "filtered")
-
-    mock_sub.stream_output("testing", streaming_process, filter_func=filter_func)
-
-    # fmt: off
-    # Output has been transformed
-    assert capsys.readouterr().out == (
-        "filtered line 1\n"
-        "\n"
-        "filtered line 3\n"
-    )
-    # fmt: on
-
-    mock_sub.cleanup.assert_called_once_with("testing", streaming_process)
-
-
-def test_filter_func_reject(mock_sub, streaming_process, sleep_zero, capsys):
-    """A filter that rejects lines can be added to modify an output stream."""
-
-    # Define a filter function that ignores blank lines
-    def filter_func(line):
-        if len(line) == 0:
-            return
-        yield line
-
-    mock_sub.stream_output("testing", streaming_process, filter_func=filter_func)
-
-    # fmt: off
-    # Output has been transformed
-    assert capsys.readouterr().out == (
-        "output line 1\n"
-        "output line 3\n"
-    )
-    # fmt: on
-
-    mock_sub.cleanup.assert_called_once_with("testing", streaming_process)
-
-
-def test_filter_func_line_ends(mock_sub, streaming_process, sleep_zero, capsys):
-    """Filter functions are not provided the newline."""
-
-    # Define a filter function that redacts lines that end with 1
-    # The newline is *not* included.
-    def filter_func(line):
-        if line.endswith("line 1"):
-            yield line.replace("line 1", "**REDACTED**")
-        else:
-            yield line
-
-    mock_sub.stream_output("testing", streaming_process, filter_func=filter_func)
-
-    # fmt: off
-    # Output has been transformed; newline exists in output
-    assert capsys.readouterr().out == (
-        "output **REDACTED**\n"
-        "\n"
-        "output line 3\n"
-    )
-    # fmt: on
-
-    mock_sub.cleanup.assert_called_once_with("testing", streaming_process)
-
-
-def test_filter_func_line_multiple_output(
-    mock_sub,
-    streaming_process,
-    sleep_zero,
-    capsys,
-):
-    """Filter functions can generate multiple lines from a single input."""
-
-    # Define a filter function that adds an extra line of content when the
-    # lines that end with 1
-    def filter_func(line):
-        yield line
-        if line.endswith("line 1"):
-            yield "Extra content!"
-
-    mock_sub.stream_output("testing", streaming_process, filter_func=filter_func)
-
-    # fmt: off
-    # Output has been transformed; newline exists in output
-    assert capsys.readouterr().out == (
-        "output line 1\n"
-        "Extra content!\n"
-        "\n"
-        "output line 3\n"
-    )
-    # fmt: on
-
-    mock_sub.cleanup.assert_called_once_with("testing", streaming_process)
-
-
-def test_filter_func_stop_iteration(mock_sub, streaming_process, capsys):
-    """A filter can indicate that logging should stop."""
-
-    # Define a filter function that converts "output" into "filtered",
-    # and terminates streaming when a blank line is seen.
-    def filter_func(line):
-        if line == "":
-            raise subprocess.StopStreaming()
-        yield line.replace("output", "filtered")
-
-    mock_sub.stream_output("testing", streaming_process, filter_func=filter_func)
-
-    # fmt: off
-    # Output has been transformed, but is truncated when the empty line was received.
-    assert capsys.readouterr().out == (
-        "filtered line 1\n"
-    )
-    # fmt: on
-
-    mock_sub.cleanup.assert_called_once_with("testing", streaming_process)
-
-
-def test_filter_func_output_and_stop_iteration(mock_sub, streaming_process, capsys):
-    """A filter can indicate that logging should stop, and also output content."""
-
-    # Define a filter function that converts "output" into "filtered",
-    # and terminates streaming when a blank line is seen; but outputs
-    # one more line before terminating.
-    def filter_func(line):
-        if line == "":
-            yield "This should be the last line"
-            raise subprocess.StopStreaming()
-        yield line.replace("output", "filtered")
-
-    mock_sub.stream_output("testing", streaming_process, filter_func=filter_func)
-
-    # fmt: off
-    # Output has been transformed, but is truncated when the empty line was received.
-    assert capsys.readouterr().out == (
-        "filtered line 1\n"
-        "This should be the last line\n"
-    )
-    # fmt: on
-
-    mock_sub.cleanup.assert_called_once_with("testing", streaming_process)
-
-
-def test_filter_func_line_unexpected_error(mock_sub, streaming_process, capsys):
-    """If a filter function fails, the error is caught and logged."""
-
-    # Define a filter function that redacts lines that end with 1
-    # The newline is *not* included.
-    def filter_func(line):
-        if not line:
-            raise RuntimeError("Like something totally went wrong")
-        yield line
-
-    mock_sub.stream_output("testing", streaming_process, filter_func=filter_func)
-
-    # fmt: off
-    # Exception
-    assert capsys.readouterr().out == (
-        "output line 1\n"
-        "Error while streaming output: RuntimeError: Like something totally went wrong\n"
-    )
-    # fmt: on
-
-    mock_sub.cleanup.assert_called_once_with("testing", streaming_process)
diff --git a/tests/integrations/subprocess/test_Subprocess__stream_output_non_blocking.py b/tests/integrations/subprocess/test_Subprocess__stream_output_non_blocking.py
new file mode 100644
index 00000000..cd3c500b
--- /dev/null
+++ b/tests/integrations/subprocess/test_Subprocess__stream_output_non_blocking.py
@@ -0,0 +1,75 @@
+from briefcase.console import LogLevel
+
+
+def test_output(mock_sub, streaming_process, sleep_zero, capsys):
+    """Process output is printed."""
+    streamer = mock_sub.stream_output_non_blocking("testing", streaming_process)
+    streamer.join()
+
+    # fmt: off
+    assert capsys.readouterr().out == (
+        "output line 1\n"
+        "\n"
+        "output line 3\n"
+    )
+    # fmt: on
+
+
+def test_output_debug(mock_sub, streaming_process, sleep_zero, capsys):
+    """Process output is printed; no debug output for only
+    stream_output_non_blocking."""
+    mock_sub.tools.logger.verbosity = LogLevel.DEBUG
+
+    streamer = mock_sub.stream_output_non_blocking("testing", streaming_process)
+    streamer.join()
+
+    # fmt: off
+    assert capsys.readouterr().out == (
+        "output line 1\n"
+        "\n"
+        "output line 3\n"
+    )
+    # fmt: on
+
+
+def test_captured_output(mock_sub, streaming_process, sleep_zero, capsys):
+    """Process output is captured and available later."""
+    streamer = mock_sub.stream_output_non_blocking(
+        "testing",
+        streaming_process,
+        capture_output=True,
+    )
+    streamer.join()
+
+    assert capsys.readouterr().out == ""
+    # fmt: off
+    assert streamer.captured_output == (
+        "output line 1\n"
+        "\n"
+        "output line 3\n"
+    )
+    # fmt: on
+
+
+def test_filter_func(mock_sub, streaming_process, sleep_zero, capsys):
+    """Process output can be filtered with a filter func."""
+
+    def filter_func(line):
+        """Filter out lines without text."""
+        if line == "":
+            return
+        yield line
+
+    streamer = mock_sub.stream_output_non_blocking(
+        "testing",
+        streaming_process,
+        filter_func=filter_func,
+    )
+    streamer.join()
+
+    # fmt: off
+    assert capsys.readouterr().out == (
+        "output line 1\n"
+        "output line 3\n"
+    )
+    # fmt: on