src/client/pxar_backup_stream.rs

   1 use std::collections::HashSet;
   2 use std::io::Write;
   3 //use std::os::unix::io::FromRawFd;
   4 use std::path::{Path, PathBuf};
   5 use std::pin::Pin;
   6 use std::sync::{Arc, Mutex};
   7 use std::task::{Context, Poll};
   8 use std::thread;
   9
  10 use failure::*;
  11 use futures::stream::Stream;
  12
  13 use nix::fcntl::OFlag;
  14 use nix::sys::stat::Mode;
  15 use nix::dir::Dir;
  16
  17 use crate::pxar;
  18 use crate::backup::CatalogWriter;
  19
  20 /// Stream implementation to encode and upload .pxar archives.
  21 ///
  22 /// The hyper client needs an async Stream for file upload, so we
  23 /// spawn an extra thread to encode the .pxar data and pipe it to the
  24 /// consumer.
  25 pub struct PxarBackupStream {
  26     rx: Option<std::sync::mpsc::Receiver<Result<Vec<u8>, Error>>>,
  27     child: Option<thread::JoinHandle<()>>,
  28     error: Arc<Mutex<Option<String>>>,
  29 }
  30
  31 impl Drop for PxarBackupStream {
  32
  33     fn drop(&mut self) {
  34         self.rx = None;
  35         self.child.take().unwrap().join().unwrap();
  36     }
  37 }
  38
  39 impl PxarBackupStream {
  40
  41     pub fn new<W: Write + Send + 'static>(
  42         mut dir: Dir,
  43         path: PathBuf,
  44         device_set: Option<HashSet<u64>>,
  45         verbose: bool,
  46         skip_lost_and_found: bool,
  47         catalog: Arc<Mutex<CatalogWriter<W>>>,
  48         exclude_pattern: Vec<pxar::MatchPattern>,
  49         entries_max: usize,
  50     ) -> Result<Self, Error> {
  51
  52         let (tx, rx) = std::sync::mpsc::sync_channel(10);
  53
  54         let buffer_size = 256*1024;
  55
  56         let error = Arc::new(Mutex::new(None));
  57         let error2 = error.clone();
  58
  59         let catalog = catalog.clone();
  60         let child = std::thread::Builder::new().name("PxarBackupStream".to_string()).spawn(move || {
  61             let mut guard = catalog.lock().unwrap();
  62             let mut writer = std::io::BufWriter::with_capacity(buffer_size, crate::tools::StdChannelWriter::new(tx));
  63
  64             if let Err(err) = pxar::Encoder::encode(
  65                 path,
  66                 &mut dir,
  67                 &mut writer,
  68                 Some(&mut *guard),
  69                 device_set,
  70                 verbose,
  71                 skip_lost_and_found,
  72                 pxar::flags::DEFAULT,
  73                 exclude_pattern,
  74                 entries_max,
  75             ) {
  76                 let mut error = error2.lock().unwrap();
  77                 *error = Some(err.to_string());
  78             }
  79         })?;
  80
  81         Ok(Self {
  82             rx: Some(rx),
  83             child: Some(child),
  84             error,
  85         })
  86     }
  87
  88     pub fn open<W: Write + Send + 'static>(
  89         dirname: &Path,
  90         device_set: Option<HashSet<u64>>,
  91         verbose: bool,
  92         skip_lost_and_found: bool,
  93         catalog: Arc<Mutex<CatalogWriter<W>>>,
  94         exclude_pattern: Vec<pxar::MatchPattern>,
  95         entries_max: usize,
  96     ) -> Result<Self, Error> {
  97
  98         let dir = nix::dir::Dir::open(dirname, OFlag::O_DIRECTORY, Mode::empty())?;
  99         let path = std::path::PathBuf::from(dirname);
 100
 101         Self::new(dir, path, device_set, verbose, skip_lost_and_found, catalog, exclude_pattern, entries_max)
 102     }
 103 }
 104
 105 impl Stream for PxarBackupStream {
 106
 107     type Item = Result<Vec<u8>, Error>;
 108
 109     fn poll_next(self: Pin<&mut Self>, _cx: &mut Context) -> Poll<Option<Self::Item>> {
 110         { // limit lock scope
 111             let error = self.error.lock().unwrap();
 112             if let Some(ref msg) = *error {
 113                 return Poll::Ready(Some(Err(format_err!("{}", msg))));
 114             }
 115         }
 116
 117         match crate::tools::runtime::block_in_place(|| self.rx.as_ref().unwrap().recv()) {
 118             Ok(data) => Poll::Ready(Some(data)),
 119             Err(_) => {
 120                 let error = self.error.lock().unwrap();
 121                 if let Some(ref msg) = *error {
 122                     return Poll::Ready(Some(Err(format_err!("{}", msg))));
 123                 }
 124                 Poll::Ready(None) // channel closed, no error
 125             }
 126         }
 127     }
 128 }