clidispatch: add start_pager API for IO

Summary:
Implement the core API to start pager in native Rust. For now it is only
enabled for the entire command if `--pager=always` is set.

Reviewed By: DurhamG

Differential Revision: D20849644

fbshipit-source-id: 860b4e18d841da607864c3447d78dbac126f5f18

eden/scm/exec/hgmain/src/main.rs

@@ -71,5 +71,6 @@ fn main() {
             code = 255;
         }
     }
+    drop(io);
     std::process::exit(code as i32);
 }

eden/scm/lib/clidispatch/Cargo.toml

@@ -7,9 +7,11 @@ edition = "2018"
 [dependencies]
 anyhow = "1.0.20"
 blackbox = { path = "../blackbox" }
-configparser = { path = "../configparser" }
 cliparser = { path = "../cliparser" }
+configparser = { path = "../configparser" }
 indexedlog = { path = "../indexedlog" }
+pipe = "0.2"
+streampager = "0.7"
 thiserror = "1.0.5"
 thrift-types = { path = "../thrift-types" }
 tracing = "0.1"

eden/scm/lib/clidispatch/src/dispatch.rs

@@ -252,6 +252,10 @@ pub fn dispatch(command_table: &CommandTable, args: Vec<String>, io: &mut IO) ->
 
     initialize_blackbox(&optional_repo)?;
 
+    if global_opts.pager == "always" {
+        io.start_pager()?;
+    }
+
     let handler = def.func();
     match handler {
         CommandFunc::Repo(f) => {

eden/scm/lib/clidispatch/src/io.rs

@@ -5,13 +5,19 @@
  * GNU General Public License version 2.
  */
 
+use pipe::pipe;
 use std::any::Any;
 use std::io;
+use std::mem;
+use std::thread::{spawn, JoinHandle};
+use streampager::Pager;
 
 pub struct IO {
     pub input: Box<dyn Read>,
     pub output: Box<dyn Write>,
     pub error: Option<Box<dyn Write>>,
+
+    pager_handle: Option<JoinHandle<streampager::Result<()>>>,
 }
 
 pub trait Read: io::Read + Any {
@@ -45,6 +51,7 @@ impl IO {
             input: Box::new(input),
             output: Box::new(output),
             error: error.map(|e| Box::new(e) as Box<dyn Write>),
+            pager_handle: None,
         }
     }
 
@@ -77,12 +84,47 @@ impl IO {
             input: Box::new(io::stdin()),
             output: Box::new(io::stdout()),
             error: Some(Box::new(io::stderr())),
+            pager_handle: None,
         }
     }
+
+    pub fn start_pager(&mut self) -> io::Result<()> {
+        if self.pager_handle.is_some() {
+            return Ok(());
+        }
+
+        let mut pager = Pager::new_using_system_terminal()
+            .map_err(|e| io::Error::new(io::ErrorKind::Other, e))?;
+        let (out_read, out_write) = pipe();
+        let (err_read, err_write) = pipe();
+
+        self.flush()?;
+        self.output = Box::new(out_write);
+        self.error = Some(Box::new(err_write));
+
+        self.pager_handle = Some(spawn(|| {
+            pager
+                .add_output_stream(out_read, "")?
+                .add_error_stream(err_read, "")?;
+            pager.run()?;
+            Ok(())
+        }));
+
+        Ok(())
+    }
 }
 
 impl Drop for IO {
     fn drop(&mut self) {
         let _ = self.flush();
+        // Drop the output and error. This sends EOF to pager.
+        self.output = Box::new(Vec::new());
+        self.error = None;
+        // Wait for the pager.
+        let mut handle = None;
+        mem::swap(&mut handle, &mut self.pager_handle);
+        if let Some(handle) = handle {
+            let _ = handle.join();
+        }
     }
 }