Add Stream::extract_with() and Stream::par_extract_with() methods which run a function for each Node in an archive

src/stream.rs

@@ -79,6 +79,27 @@ impl<R: Read + Send> Stream<R> {
         Ok(())
     }
 
+    /// Extracts an archive, running the function `f` for each node
+    /// # Errors
+    /// Returns `crate::Error` if io fails or several other error conditions
+    pub fn extract_with<F>(
+        &mut self,
+        prefix: Option<&str>,
+        uid: Option<u32>,
+        gid: Option<u32>,
+        f: F,
+    ) -> Result<(), Error>
+    where
+        F: FnOnce(Node,  Option<u32>, Option<u32>) + Copy,
+    {
+        for node in self {
+            let node = node?;
+            node.extract(prefix, uid, gid)?;
+            f(node, uid, gid);
+        }
+        Ok(())
+    }
+
     #[cfg(feature = "parallel")]
     /// Extracts and archive in parallel
     /// # Errors
@@ -130,4 +151,62 @@ impl<R: Read + Send> Stream<R> {
         sender.send(Message::Eof).map_err(|_| Error::SenderError)?;
         Ok(())
     }
+
+    #[cfg(feature = "parallel")]
+    /// Extracts and archive in parallel and runs the passed in function for
+    /// each `Node`
+    /// # Errors
+    /// Returns `crate::Error` if io fails or several other error conditions
+    pub fn par_extract_with<F>(
+        &mut self,
+        prefix: Option<&str>,
+        uid: Option<u32>,
+        gid: Option<u32>,
+        sender: &Sender<Message>,
+        f: F,
+    ) -> Result<(), Error>
+    where
+        F: FnOnce(Node,  Option<u32>, Option<u32>) + Copy + Send + Sync,
+    {
+        let s = sender.clone();
+        self.into_iter().par_bridge().try_for_each_with(s, |s, n| {
+            let n = n?;
+            n.extract(prefix, uid, gid)?;
+            match n.filetype {
+                FileType::Normal(ref f) => {
+                    s.send(Message::FileExtracted {
+                        name: n.name.clone(),
+                        size: f.len,
+                    })
+                    .map_err(|_| Error::SenderError)?;
+                }
+                FileType::SoftLink(ref t) | FileType::HardLink(ref t) => {
+                    s.send(Message::LinkCreated {
+                        name: n.name.clone(),
+                        target: t.clone(),
+                    })
+                    .map_err(|_| Error::SenderError)?;
+                }
+                FileType::Directory => {
+                    s.send(Message::DirectoryCreated {
+                        name: n.name.clone(),
+                    })
+                    .map_err(|_| Error::SenderError)?;
+                }
+                FileType::Block(_) | FileType::Character(_) | FileType::Fifo => {
+                    s.send(Message::DeviceCreated {
+                        name: n.name.clone(),
+                    })
+                    .map_err(|_| Error::SenderError)?;
+                }
+                FileType::Eof => {
+                    s.send(Message::Eof).map_err(|_| Error::SenderError)?;
+                }
+            }
+            f(n, uid, gid);
+            Ok::<(), Error>(())
+        })?;
+        sender.send(Message::Eof).map_err(|_| Error::SenderError)?;
+        Ok(())
+    }
 }