src/client/pxar_backup_stream.rs: use WrappedStreamReader

to make it fully async ...

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<std::fs::File>,
+    stream: WrappedReaderStream<std::fs::File>,
-    buffer: Vec<u8>,
     child: Option<thread::JoinHandle<()>>,
 }
 
 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<Self, Error> {
-        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<Self, Error> {
@@ -71,33 +69,7 @@ impl Stream for PxarBackupStream {
     type Item = Vec<u8>;
     type Error = Error;
 
-    // Note: This is not async!!
-
     fn poll(&mut self) -> Poll<Option<Vec<u8>>, Error> {
-
+        self.stream.poll().map_err(Error::from)
-        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())
-                }
-            };
-        }
     }
 }