diff --git a/src/client/pxar_backup_stream.rs b/src/client/pxar_backup_stream.rs index dffdd589..484e7989 100644 --- a/src/client/pxar_backup_stream.rs +++ b/src/client/pxar_backup_stream.rs @@ -4,7 +4,7 @@ use std::thread; use std::os::unix::io::FromRawFd; use std::path::{Path, PathBuf}; -use futures::{Async, Poll}; +use futures::Poll; use futures::stream::Stream; use nix::fcntl::OFlag; @@ -12,6 +12,7 @@ use nix::sys::stat::Mode; use nix::dir::Dir; use crate::pxar; +use crate::tools::wrapped_reader_stream::WrappedReaderStream; /// Stream implementation to encode and upload .pxar archives. /// @@ -21,15 +22,13 @@ use crate::pxar; /// /// Note: The currect implementation is not fully ansync and can block. pub struct PxarBackupStream { - pipe: Option, - buffer: Vec, + stream: WrappedReaderStream, child: Option>, } impl Drop for PxarBackupStream { fn drop(&mut self) { - drop(self.pipe.take()); self.child.take().unwrap().join().unwrap(); } } @@ -37,12 +36,10 @@ impl Drop for PxarBackupStream { impl PxarBackupStream { pub fn new(mut dir: Dir, path: PathBuf, all_file_systems: bool, verbose: bool) -> Result { - let buffer_size = 1024*1024; - let mut buffer = Vec::with_capacity(buffer_size); - unsafe { buffer.set_len(buffer.capacity()); } let (rx, tx) = nix::unistd::pipe()?; + let buffer_size = 1024*1024; nix::fcntl::fcntl(rx, nix::fcntl::FcntlArg::F_SETPIPE_SZ(buffer_size as i32))?; let child = thread::spawn(move|| { @@ -53,8 +50,9 @@ impl PxarBackupStream { }); let pipe = unsafe { std::fs::File::from_raw_fd(rx) }; + let stream = crate::tools::wrapped_reader_stream::WrappedReaderStream::new(pipe); - Ok(Self { pipe: Some(pipe), buffer, child: Some(child) }) + Ok(Self { stream, child: Some(child) }) } pub fn open(dirname: &Path, all_file_systems: bool, verbose: bool) -> Result { @@ -71,33 +69,7 @@ impl Stream for PxarBackupStream { type Item = Vec; type Error = Error; - // Note: This is not async!! - fn poll(&mut self) -> Poll>, Error> { - - use std::io::Read; - - loop { - let pipe = match self.pipe { - Some(ref mut pipe) => pipe, - None => unreachable!(), - }; - match pipe.read(&mut self.buffer) { - Ok(n) => { - if n == 0 { - return Ok(Async::Ready(None)) - } else { - let data = self.buffer[..n].to_vec(); - return Ok(Async::Ready(Some(data))) - } - } - Err(ref e) if e.kind() == std::io::ErrorKind::Interrupted => { - // try again - } - Err(err) => { - return Err(err.into()) - } - }; - } + self.stream.poll().map_err(Error::from) } }