Process stderr from test-proxy differently

Resolves #2715. On Windows, listening on stderr was keeping the process alive and not shutting down immediately after all test processes were done and the proxy dropped.

Author: heaths

sdk/core/azure_core_test/src/proxy/mod.rs

@@ -28,8 +28,9 @@ use serde::Serializer;
 #[cfg(not(target_arch = "wasm32"))]
 use std::process::ExitStatus;
 use std::{fmt, str::FromStr, sync::Arc};
+use tokio::sync::Notify;
 #[cfg(not(target_arch = "wasm32"))]
-use tokio::process::Child;
+use tokio::{io::Lines, process::Child};
 
 const ABSTRACTION_IDENTIFIER: HeaderName = HeaderName::from_static("x-abstraction-identifier");
 const RECORDING_ID: HeaderName = HeaderName::from_static("x-recording-id");
@@ -45,6 +46,8 @@ pub use bootstrap::start;
 pub struct Proxy {
     #[cfg(not(target_arch = "wasm32"))]
     command: Option<Child>,
+    #[cfg(not(target_arch = "wasm32"))]
+    done: Option<Arc<Notify>>,
     endpoint: Option<Url>,
     client: Option<Client>,
 }
@@ -76,34 +79,57 @@ impl Proxy {
                 )
             })?;
 
-        let mut stdout = command
-            .stdout
-            .take()
-            .ok_or_else(|| azure_core::Error::message(ErrorKind::Io, "no stdout pipe"))?;
+        let mut stdout = BufReader::new(
+            command
+                .stdout
+                .take()
+                .ok_or_else(|| azure_core::Error::message(ErrorKind::Io, "no stdout pipe"))?,
+        )
+        .lines();
         // Take stderr now but we won't listen until after start up, such that messages should buffer.
-        let mut stderr = command
-            .stderr
-            .take()
-            .ok_or_else(|| azure_core::Error::message(ErrorKind::Io, "no stderr pipe"))?;
+        // let mut stderr = BufReader::new(
+        //     command
+        //         .stderr
+        //         .take()
+        //         .ok_or_else(|| azure_core::Error::message(ErrorKind::Io, "no stderr pipe"))?,
+        // )
+        // .lines();
+
+        let done = Arc::new(Notify::new());
         self.command = Some(command);
+        self.done = Some(done.clone());
 
         // Wait until the service is listening on a port.
         self.wait_till_listening(&mut stdout).await?;
 
         // Then spawn a thread to keep pumping messages to stdout and stderr.
         // The pipe will be closed when the process is shut down, which will terminate the task.
-        tokio::spawn(async move {
-            let mut reader = BufReader::new(&mut stdout).lines();
-            while let Some(line) = reader.next_line().await.unwrap_or(None) {
+        let mut stdout_task = tokio::spawn(async move {
+            while let Some(line) = stdout.next_line().await.unwrap_or(None) {
                 // Trace useful lines that test-proxy writes to stdout.
                 trace_line(Level::TRACE, &line);
             }
         });
+        // let mut stderr_task = tokio::spawn(async move {
+        //     while let Some(line) = stderr.next_line().await.unwrap_or(None) {
+        //         // Trace useful lines that test-proxy writes to stderr.
+        //         trace_line(Level::TRACE, &line);
+        //     }
+        // });
         tokio::spawn(async move {
-            let mut reader = BufReader::new(&mut stderr).lines();
-            while let Some(line) = reader.next_line().await.unwrap_or(None) {
-                // Trace useful lines that test-proxy writes to stdout.
-                trace_line(Level::ERROR, &line);
+            loop {
+                tokio::select! {
+                    _ = &mut stdout_task => {
+                        eprintln!("stdout");
+                    }
+                    // _ = &mut stderr_task => {
+                    //     eprintln!("stderr");
+                    // },
+                    _ = done.notified() => {
+                        eprintln!("done");
+                        break;
+                    },
+                };
             }
         });
 
@@ -138,6 +164,8 @@ impl Proxy {
     ///
     /// Waits until the process is killed.
     pub async fn stop(&mut self) -> Result<()> {
+        tracing::warn!("stopping");
+        self.done();
         if let Some(command) = &mut self.command {
             tracing::debug!(pid = ?command.id(), "stopping");
             return Ok(command.kill().await?);
@@ -147,16 +175,20 @@ impl Proxy {
 
     /// Waits for the Test Proxy service to exit, return the process exit code when completed.
     pub async fn wait(&mut self) -> Result<ExitStatus> {
+        tracing::warn!("waiting");
+        self.done();
         if let Some(command) = &mut self.command {
             return Ok(command.wait().await?);
         }
         Ok(ExitStatus::default())
     }
 
-    async fn wait_till_listening(&mut self, stdout: &mut ChildStdout) -> Result<()> {
+    async fn wait_till_listening(
+        &mut self,
+        stdout: &mut Lines<BufReader<ChildStdout>>,
+    ) -> Result<()> {
         let pid = self.command.as_ref().and_then(Child::id);
-        let mut reader = BufReader::new(stdout).lines();
-        while let Some(line) = reader.next_line().await? {
+        while let Some(line) = stdout.next_line().await? {
             const RUNNING_PATTERN: &str = "Running proxy version is Azure.Sdk.Tools.TestProxy ";
             const LISTENING_PATTERN: &str = "Now listening on: ";
 
@@ -192,6 +224,13 @@ impl Proxy {
 
         Ok(())
     }
+
+    fn done(&self) {
+        if let Some(done) = &self.done {
+            eprintln!("signalling done");
+            done.notify_one();
+        }
+    }
 }
 
 impl Proxy {
@@ -201,6 +240,8 @@ impl Proxy {
         Ok(Self {
             #[cfg(not(target_arch = "wasm32"))]
             command: None,
+            #[cfg(not(target_arch = "wasm32"))]
+            done: None,
             endpoint: Some(endpoint.clone()),
             client: Some(Client::new(endpoint)?),
         })
@@ -221,6 +262,8 @@ impl Drop for Proxy {
     ///
     /// Does not wait until the process is killed.
     fn drop(&mut self) {
+        eprintln!("dropping Proxy");
+        self.done();
         if let Some(command) = &mut self.command {
             let _ = command.start_kill();
         }
@@ -273,7 +316,7 @@ impl Default for ProxyOptions {
         Self {
             auto: true,
             insecure: false,
-            auto_shutdown_in_seconds: 300,
+            auto_shutdown_in_seconds: 10,
         }
     }
 }