use failure::*; use std::thread; use std::os::unix::io::FromRawFd; use std::path::{Path, PathBuf}; use futures::Poll; use futures::stream::Stream; use nix::fcntl::OFlag; 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. /// /// The hyper client needs an async Stream for file upload, so we /// spawn an extra thread to encode the .pxar data and pipe it to the /// consumer. pub struct PxarBackupStream { stream: Option>, child: Option>, } impl Drop for PxarBackupStream { fn drop(&mut self) { self.stream = None; self.child.take().unwrap().join().unwrap(); } } impl PxarBackupStream { pub fn new(mut dir: Dir, path: PathBuf, all_file_systems: bool, verbose: bool) -> Result { 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|| { let mut writer = unsafe { std::fs::File::from_raw_fd(tx) }; let no_xattrs = false; let no_fcaps = false; if let Err(err) = pxar::Encoder::encode(path, &mut dir, &mut writer, all_file_systems, verbose, no_xattrs, no_fcaps) { eprintln!("pxar encode failed - {}", err); } }); let pipe = unsafe { std::fs::File::from_raw_fd(rx) }; let stream = crate::tools::wrapped_reader_stream::WrappedReaderStream::new(pipe); Ok(Self { stream: Some(stream), child: Some(child) }) } pub fn open(dirname: &Path, all_file_systems: bool, verbose: bool) -> Result { let dir = nix::dir::Dir::open(dirname, OFlag::O_DIRECTORY, Mode::empty())?; let path = std::path::PathBuf::from(dirname); Self::new(dir, path, all_file_systems, verbose) } } impl Stream for PxarBackupStream { type Item = Vec; type Error = Error; fn poll(&mut self) -> Poll>, Error> { self.stream.as_mut().unwrap().poll().map_err(Error::from) } }