-
Notifications
You must be signed in to change notification settings - Fork 703
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Added from_ method constructor for ChildStdin, ChildStdout and ChildS…
…tderr that may interact with mio to fix #387 Uses into_raw_fd to hand over ownership to mio's Pipe{Writer,Reader}
- Loading branch information
Showing
7 changed files
with
306 additions
and
7 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -10,8 +10,10 @@ pub use self::unix::{ | |
| UdpSocket, | ||
| UnixSocket, | ||
| pipe, | ||
| set_nonblock, | ||
| }; | ||
|
|
||
|
|
||
| #[cfg(unix)] | ||
| mod unix; | ||
|
|
||
|
|
||
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,263 @@ | ||
| #![cfg(unix)] | ||
|
|
||
| extern crate mio; | ||
| extern crate bytes; | ||
| use std::mem; | ||
| use mio::*; | ||
| use std::io; | ||
| use mio::unix::{PipeReader, PipeWriter}; | ||
| use std::process::{Command, Stdio, Child}; | ||
|
|
||
|
|
||
| struct SubprocessClient { | ||
| stdin: Option<PipeWriter>, | ||
| stdout: Option<PipeReader>, | ||
| stderr: Option<PipeReader>, | ||
| stdin_token : Token, | ||
| stdout_token : Token, | ||
| stderr_token : Token, | ||
| output : Vec<u8>, | ||
| output_stderr : Vec<u8>, | ||
| input : Vec<u8>, | ||
| input_offset : usize, | ||
| buf : [u8; 65536], | ||
| } | ||
|
|
||
|
|
||
| // Sends a message and expects to receive the same exact message, one at a time | ||
| impl SubprocessClient { | ||
| fn new(stdin: Option<PipeWriter>, stdout : Option<PipeReader>, stderr : Option<PipeReader>, data : &[u8]) -> SubprocessClient { | ||
| SubprocessClient { | ||
| stdin: stdin, | ||
| stdout: stdout, | ||
| stderr: stderr, | ||
| stdin_token : Token(0), | ||
| stdout_token : Token(1), | ||
| stderr_token : Token(2), | ||
| output : Vec::<u8>::new(), | ||
| output_stderr : Vec::<u8>::new(), | ||
| buf : [0; 65536], | ||
| input : data.to_vec(), | ||
| input_offset : 0, | ||
| } | ||
| } | ||
|
|
||
| fn readable(&mut self, event_loop: &mut EventLoop<SubprocessClient>) -> io::Result<()> { | ||
| println!("client socket readable"); | ||
| let mut eof = false; | ||
| match self.stdout { | ||
| None => unreachable!(), | ||
| Some (ref mut stdout) => match stdout.try_read(&mut self.buf[..]) { | ||
| Ok(None) => { | ||
| println!("CLIENT : spurious read wakeup"); | ||
| } | ||
| Ok(Some(r)) => { | ||
| println!("CLIENT : We read {} bytes!", r); | ||
| if r == 0 { | ||
| eof = true; | ||
| } else { | ||
| self.output.extend(&self.buf[0..r]); | ||
| } | ||
| } | ||
| Err(e) => { | ||
| return Err(e); | ||
| } | ||
| } | ||
| }; | ||
| if eof { | ||
| drop(self.stdout.take()); | ||
| match self.stderr { | ||
| None => event_loop.shutdown(), | ||
| Some(_) => {}, | ||
| } | ||
| } | ||
| return Ok(()); | ||
| } | ||
|
|
||
| fn readable_stderr(&mut self, event_loop: &mut EventLoop<SubprocessClient>) -> io::Result<()> { | ||
| println!("client socket readable"); | ||
| let mut eof = false; | ||
| match self.stderr { | ||
| None => unreachable!(), | ||
| Some(ref mut stderr) => match stderr.try_read(&mut self.buf[..]) { | ||
| Ok(None) => { | ||
| println!("CLIENT : spurious read wakeup"); | ||
| } | ||
| Ok(Some(r)) => { | ||
| println!("CLIENT : We read {} bytes!", r); | ||
| if r == 0 { | ||
| eof = true; | ||
| } else { | ||
| self.output_stderr.extend(&self.buf[0..r]); | ||
| } | ||
| } | ||
| Err(e) => { | ||
| return Err(e); | ||
| } | ||
| } | ||
| }; | ||
| if eof { | ||
| drop(self.stderr.take()); | ||
| match self.stdout { | ||
| None => event_loop.shutdown(), | ||
| Some(_) => {}, | ||
| } | ||
| } | ||
| return Ok(()); | ||
| } | ||
|
|
||
| fn writable(&mut self, event_loop: &mut EventLoop<SubprocessClient>) -> io::Result<()> { | ||
| println!("client socket writable"); | ||
| let mut ok = true; | ||
| match self.stdin { | ||
| None => unreachable!(), | ||
| Some(ref mut stdin) => match stdin.try_write(&(&self.input)[self.input_offset..]) { | ||
| Ok(None) => { | ||
| println!("client flushing buf; WOULDBLOCK"); | ||
| }, | ||
| Ok(Some(r)) => { | ||
| if r == 0 { | ||
| println!("CLIENT : we wrote {} bytes!", r); | ||
| ok = false; | ||
| } else { | ||
| self.input_offset += r; | ||
| } | ||
| }, | ||
| Err(e) => { | ||
| println!("not implemented; client err={:?}", e); | ||
| ok = false; | ||
| }, | ||
| } | ||
| } | ||
| if self.input_offset == self.input.len() || !ok { | ||
| drop(self.stdin.take()); | ||
| match self.stderr { | ||
| None => match self.stdout { | ||
| None => event_loop.shutdown(), | ||
| Some(_) => {}, | ||
| }, | ||
| Some(_) => {}, | ||
| } | ||
| } | ||
| return Ok(()); | ||
| } | ||
|
|
||
| } | ||
|
|
||
| impl Handler for SubprocessClient { | ||
| type Timeout = usize; | ||
| type Message = (); | ||
|
|
||
| fn ready(&mut self, event_loop: &mut EventLoop<SubprocessClient>, token: Token, | ||
| events: EventSet) { | ||
| println!("ready {:?} {:?} {:}", token, events, events.is_readable()); | ||
| if token == self.stderr_token { | ||
| let _x = self.readable_stderr(event_loop); | ||
| } else { | ||
| let _x = self.readable(event_loop); | ||
| } | ||
| if token == self.stdin_token { | ||
| let _y = self.writable(event_loop); | ||
| } | ||
| } | ||
| } | ||
|
|
||
|
|
||
|
|
||
|
|
||
| const TEST_DATA : [u8; 1024 * 4096] = [42; 1024 * 4096]; | ||
| pub fn subprocess_communicate(mut process : Child, input : &[u8]) -> (Vec<u8>, Vec<u8>) { | ||
| let mut event_loop = EventLoop::<SubprocessClient>::new().unwrap(); | ||
| let stdin : Option<PipeWriter>; | ||
| let stdin_exists : bool; | ||
| match process.stdin { | ||
| None => stdin_exists = false, | ||
| Some(_) => stdin_exists = true, | ||
| } | ||
| if stdin_exists { | ||
| match PipeWriter::from_stdin(process.stdin.take().unwrap()) { | ||
| Err(e) => panic!(e), | ||
| Ok(pipe) => stdin = Some(pipe), | ||
| } | ||
| } else { | ||
| stdin = None; | ||
| } | ||
| let stdout_exists : bool; | ||
| let stdout : Option<PipeReader>; | ||
| match process.stdout { | ||
| None => stdout_exists = false, | ||
| Some(_) => stdout_exists = true, | ||
| } | ||
| if stdout_exists { | ||
| match PipeReader::from_stdout(process.stdout.take().unwrap()) { | ||
| Err(e) => panic!(e), | ||
| Ok(pipe) => stdout = Some(pipe), | ||
| } | ||
| } else { | ||
| stdout = None; | ||
| } | ||
| let stderr_exists : bool; | ||
| let stderr : Option<PipeReader>; | ||
| match process.stderr { | ||
| None => stderr_exists = false, | ||
| Some(_) => stderr_exists = true, | ||
| } | ||
| if stderr_exists { | ||
| match PipeReader::from_stderr(process.stderr.take().unwrap()) { | ||
| Err(e) => panic!(e), | ||
| Ok(pipe) => stderr = Some(pipe), | ||
| } | ||
| } else { | ||
| stderr = None | ||
| } | ||
| //println!("listen for connections {:?} {:?}", , process.stdout.unwrap().as_raw_fd()); | ||
| let mut subprocess = SubprocessClient::new(stdin, | ||
| stdout, | ||
| stderr, | ||
| input); | ||
| match subprocess.stdout { | ||
| Some(ref sub_stdout) => event_loop.register(sub_stdout, subprocess.stdout_token, EventSet::readable(), | ||
| PollOpt::level()).unwrap(), | ||
| None => {}, | ||
| } | ||
|
|
||
| match subprocess.stderr { | ||
| Some(ref sub_stderr) => event_loop.register(sub_stderr, subprocess.stderr_token, EventSet::readable(), | ||
| PollOpt::level()).unwrap(), | ||
| None => {}, | ||
| } | ||
|
|
||
| // Connect to the server | ||
| match subprocess.stdin { | ||
| Some (ref sub_stdin) => event_loop.register(sub_stdin, subprocess.stdin_token, EventSet::writable(), | ||
| PollOpt::level()).unwrap(), | ||
| None => {}, | ||
| } | ||
|
|
||
| // Start the event loop | ||
| event_loop.run(&mut subprocess).unwrap(); | ||
| let res = process.wait(); | ||
| println!("Final output was {:} {:} {:?}\n", subprocess.output.len(), subprocess.output_stderr.len(), res); | ||
|
|
||
| let ret_stdout = mem::replace(&mut subprocess.output, Vec::<u8>::new()); | ||
| let ret_stderr = mem::replace(&mut subprocess.output_stderr, Vec::<u8>::new()); | ||
| return (ret_stdout, ret_stderr); | ||
| } | ||
|
|
||
| #[test] | ||
| fn test_subprocess_pipe() { | ||
| let process = | ||
| Command::new("/bin/cat") | ||
| .stdin(Stdio::piped()) | ||
| .stdout(Stdio::piped()) | ||
| .stderr(Stdio::piped()) | ||
| .spawn().unwrap(); | ||
| let (ret_stdout, ret_stderr) = subprocess_communicate(process, &TEST_DATA[..]); | ||
| assert_eq!(TEST_DATA.len(), ret_stdout.len()); | ||
| assert_eq!(0usize, ret_stderr.len()); | ||
| let mut i : usize = 0; | ||
| for item in TEST_DATA.iter() { | ||
| assert_eq!(*item, ret_stdout[i]); | ||
| i += 1; | ||
| } | ||
| } |