Allow capturing of a task's stdout and log

src/libextra/test.rs

@@ -34,6 +34,7 @@ use std::io;
 use std::io::File;
 use std::io::Writer;
 use std::io::stdio::StdWriter;
+use std::str;
 use std::task;
 use std::to_str::ToStr;
 use std::f64;
@@ -350,7 +351,7 @@ struct ConsoleTestState<T> {
     ignored: uint,
     measured: uint,
     metrics: MetricMap,
-    failures: ~[TestDesc],
+    failures: ~[(TestDesc, ~[u8])],
     max_name_len: uint, // number of columns to fill when aligning names
 }
 
@@ -484,14 +485,27 @@ impl<T: Writer> ConsoleTestState<T> {
     }
 
     pub fn write_failures(&mut self) {
-        self.write_plain("\nfailures:\n");
         let mut failures = ~[];
-        for f in self.failures.iter() {
+        let mut fail_out  = ~"";
+        for &(ref f, ref stdout) in self.failures.iter() {
             failures.push(f.name.to_str());
+            if stdout.len() > 0 {
+                fail_out.push_str(format!("---- {} stdout ----\n\t",
+                                  f.name.to_str()));
+                let output = str::from_utf8(*stdout);
+                fail_out.push_str(output.replace("\n", "\n\t"));
+                fail_out.push_str("\n");
+            }
+        }
+        if fail_out.len() > 0 {
+            self.write_plain("\n");
+            self.write_plain(fail_out);
         }
+
+        self.write_plain("\nfailures:\n");
         sort::tim_sort(failures);
         for name in failures.iter() {
-            self.write_plain(format!("    {}\n", name.to_str()));
+            self.write_plain(format!("    {}\n", name.as_slice()));
         }
     }
 
@@ -612,7 +626,7 @@ pub fn run_tests_console(opts: &TestOpts,
         match (*event).clone() {
             TeFiltered(ref filtered_tests) => st.write_run_start(filtered_tests.len()),
             TeWait(ref test, padding) => st.write_test_start(test, padding),
-            TeResult(test, result) => {
+            TeResult(test, result, stdout) => {
                 st.write_log(&test, &result);
                 st.write_result(&result);
                 match result {
@@ -634,7 +648,7 @@ pub fn run_tests_console(opts: &TestOpts,
                     }
                     TrFailed => {
                         st.failed += 1;
-                        st.failures.push(test);
+                        st.failures.push((test, stdout));
                     }
                 }
             }
@@ -696,7 +710,7 @@ fn should_sort_failures_before_printing_them() {
         measured: 0u,
         max_name_len: 10u,
         metrics: MetricMap::new(),
-        failures: ~[test_b, test_a]
+        failures: ~[(test_b, ~[]), (test_a, ~[])]
     };
 
     st.write_failures();
@@ -716,10 +730,10 @@ fn use_color() -> bool { return get_concurrency() == 1; }
 enum TestEvent {
     TeFiltered(~[TestDesc]),
     TeWait(TestDesc, NamePadding),
-    TeResult(TestDesc, TestResult),
+    TeResult(TestDesc, TestResult, ~[u8] /* stdout */),
 }
 
-type MonitorMsg = (TestDesc, TestResult);
+type MonitorMsg = (TestDesc, TestResult, ~[u8] /* stdout */);
 
 fn run_tests(opts: &TestOpts,
              tests: ~[TestDescAndFn],
@@ -762,11 +776,11 @@ fn run_tests(opts: &TestOpts,
             pending += 1;
         }
 
-        let (desc, result) = p.recv();
+        let (desc, result, stdout) = p.recv();
         if concurrency != 1 {
             callback(TeWait(desc.clone(), PadNone));
         }
-        callback(TeResult(desc, result));
+        callback(TeResult(desc, result, stdout));
         pending -= 1;
     }
 
@@ -775,8 +789,8 @@ fn run_tests(opts: &TestOpts,
     for b in filtered_benchs_and_metrics.move_iter() {
         callback(TeWait(b.desc.clone(), b.testfn.padding()));
         run_test(!opts.run_benchmarks, b, ch.clone());
-        let (test, result) = p.recv();
-        callback(TeResult(test, result));
+        let (test, result, stdout) = p.recv();
+        callback(TeResult(test, result, stdout));
     }
 }
 
@@ -865,7 +879,7 @@ pub fn run_test(force_ignore: bool,
     let TestDescAndFn {desc, testfn} = test;
 
     if force_ignore || desc.ignore {
-        monitor_ch.send((desc, TrIgnored));
+        monitor_ch.send((desc, TrIgnored, ~[]));
         return;
     }
 
@@ -875,40 +889,43 @@ pub fn run_test(force_ignore: bool,
         let testfn_cell = ::std::cell::Cell::new(testfn);
         do task::spawn {
             let mut task = task::task();
+            let (mut reader, writer) = io::comm::stream();
             task.name(match desc.name {
                 DynTestName(ref name) => SendStrOwned(name.clone()),
                 StaticTestName(name) => SendStrStatic(name),
             });
+            task.opts.stdout = Some(~writer as ~Writer);
             let result_future = task.future_result();
             task.spawn(testfn_cell.take());
 
+            let stdout = reader.read_to_end();
             let task_result = result_future.recv();
             let test_result = calc_result(&desc, task_result.is_ok());
-            monitor_ch.send((desc.clone(), test_result));
+            monitor_ch.send((desc.clone(), test_result, stdout));
         }
     }
 
     match testfn {
         DynBenchFn(bencher) => {
             let bs = ::test::bench::benchmark(|harness| bencher.run(harness));
-            monitor_ch.send((desc, TrBench(bs)));
+            monitor_ch.send((desc, TrBench(bs), ~[]));
             return;
         }
         StaticBenchFn(benchfn) => {
             let bs = ::test::bench::benchmark(benchfn);
-            monitor_ch.send((desc, TrBench(bs)));
+            monitor_ch.send((desc, TrBench(bs), ~[]));
             return;
         }
         DynMetricFn(f) => {
             let mut mm = MetricMap::new();
             f(&mut mm);
-            monitor_ch.send((desc, TrMetrics(mm)));
+            monitor_ch.send((desc, TrMetrics(mm), ~[]));
             return;
         }
         StaticMetricFn(f) => {
             let mut mm = MetricMap::new();
             f(&mut mm);
-            monitor_ch.send((desc, TrMetrics(mm)));
+            monitor_ch.send((desc, TrMetrics(mm), ~[]));
             return;
         }
         DynTestFn(f) => run_test_inner(desc, monitor_ch, f),

src/libstd/io/comm.rs

@@ -0,0 +1,142 @@
+// Copyright 2013 The Rust Project Developers. See the COPYRIGHT
+// file at the top-level directory of this distribution and at
+// http://rust-lang.org/COPYRIGHT.
+//
+// Licensed under the Apache License, Version 2.0 <LICENSE-APACHE or
+// http://www.apache.org/licenses/LICENSE-2.0> or the MIT license
+// <LICENSE-MIT or http://opensource.org/licenses/MIT>, at your
+// option. This file may not be copied, modified, or distributed
+// except according to those terms.
+
+//! Readers and Writers using channels as the underlying transport
+//!
+//! This module is useful for communication between tasks through Readers and
+//! Writers. This implements the Reader/Writer interface on top of the Port/Chan
+//! types to provide a blocking `read` method and an always-asynchronous `write`
+//! method.
+//!
+//! # Example
+//!
+//!     use std::io::comm;
+//!
+//!     let (mut reader, writer) = comm::stream();
+//!
+//!     do spawn {
+//!         let mut writer = writer;
+//!         writer.write([1, 2, 3]);
+//!     }
+//!
+//!     assert_eq!(reader.read_to_end(), ~[1, 2, 3]);
+
+use cmp;
+use comm;
+use comm::{GenericChan, GenericPort, Port, Chan};
+use io::{Writer, Reader};
+use option::{Some, None, Option};
+use vec;
+use vec::{ImmutableVector, CopyableVector};
+
+/// Creates a new connected (reader, writer) pair. All data written on the
+/// writer will show up on the reader.
+pub fn stream() -> (PortReader, ChanWriter) {
+    let (port, chan) = comm::stream();
+    (PortReader { prev: None, port: port, pos: 0 }, ChanWriter { chan: chan })
+}
+
+/// Wrapper struct around an internal communication port to read data from
+pub struct PortReader {
+    priv prev: Option<~[u8]>,
+    priv pos: uint,
+    priv port: Port<~[u8]>,
+}
+
+/// Wrapper struct around a channel to implement a `Writer` interface.
+pub struct ChanWriter {
+    priv chan: Chan<~[u8]>,
+}
+
+impl Writer for ChanWriter {
+    fn write(&mut self, data: &[u8]) {
+        self.chan.send(data.to_owned());
+    }
+}
+
+impl Reader for PortReader {
+    fn read(&mut self, data: &mut [u8]) -> Option<uint> {
+        let mut offset = 0;
+        while offset < data.len() {
+            match self.prev {
+                Some(ref b) => {
+                    let dst = data.mut_slice_from(offset);
+                    let src = b.slice_from(self.pos);
+                    let amt = cmp::min(dst.len(), src.len());
+                    vec::bytes::copy_memory(dst, src, amt);
+                    self.pos += amt;
+                    offset += amt;
+                    if self.pos < b.len() {
+                        break
+                    }
+                }
+                None => {}
+            }
+            self.prev = self.port.try_recv();
+            self.pos = 0;
+            if self.prev.is_none() { break }
+        }
+        if offset == 0 {
+            None
+        } else {
+            Some(offset)
+        }
+    }
+
+    fn eof(&mut self) -> bool { false }
+}
+
+#[cfg(test)]
+mod tests {
+    use prelude::*;
+
+    use io::comm::stream;
+    use util;
+
+    #[test]
+    fn smoke() {
+        let (mut reader, mut writer) = stream();
+        writer.write([1, 2, 3, 4]);
+        util::ignore(writer);
+
+        assert_eq!(reader.read_byte(), Some(1));
+        let mut buf = [0, 0];
+        assert_eq!(reader.read(buf), Some(2));
+        assert_eq!(buf[0], 2);
+        assert_eq!(buf[1], 3);
+
+        assert_eq!(reader.read(buf), Some(1));
+        assert_eq!(buf[0], 4);
+
+        assert_eq!(reader.read(buf), None);
+    }
+
+    #[test]
+    fn parallel() {
+        let (mut reader, writer) = stream();
+        do spawn {
+            let mut writer = writer;
+            writer.write([1, 2]);
+            writer.write([3]);
+            writer.write([4]);
+        }
+
+        assert_eq!(reader.read_byte(), Some(1));
+        let mut buf = [0, 0];
+        assert_eq!(reader.read(buf), Some(2));
+        assert_eq!(buf[0], 2);
+        assert_eq!(buf[1], 3);
+
+        assert_eq!(reader.read(buf), Some(1));
+        assert_eq!(buf[0], 4);
+
+        assert_eq!(reader.read(buf), None);
+    }
+}

src/libstd/io/mod.rs

@@ -298,6 +298,8 @@ pub mod flate;
 /// Interop between byte streams and pipes. Not sure where it belongs
 pub mod comm_adapters;
 
+pub mod comm;
+
 /// Extension traits
 pub mod extensions;
 

src/libstd/logging.rs

@@ -92,8 +92,10 @@ start, print out all modules registered for logging, and then exit.
 use fmt;
 use option::*;
 use rt::local::Local;
-use rt::logging::{Logger, StdErrLogger};
 use rt::task::Task;
+use std::io;
+use std::io::buffered::LineBufferedWriter;
+use std::io::stdio;
 
 /// This function is called directly by the compiler when using the logging
 /// macros. This function does not take into account whether the log level
@@ -110,18 +112,18 @@ pub fn log(_level: u32, args: &fmt::Arguments) {
                 // Lazily initialize the local task's logger
                 match (*local).logger {
                     // Use the available logger if we have one
-                    Some(ref mut logger) => { logger.log(args); }
+                    Some(ref mut logger) => { fmt::writeln(*logger, args); }
                     None => {
-                        let mut logger = StdErrLogger::new();
-                        logger.log(args);
-                        (*local).logger = Some(logger);
+                        let mut logger = LineBufferedWriter::new(stdio::stderr());
+                        fmt::writeln(&mut logger as &mut io::Writer, args);
+                        (*local).logger = Some(~logger as ~io::Writer);
                     }
                 }
             }
             // If there's no local task, then always log to stderr
             None => {
-                let mut logger = StdErrLogger::new();
-                logger.log(args);
+                let mut logger = stdio::stderr();
+                fmt::writeln(&mut logger as &mut io::Writer, args);
             }
         }
     }