notmuch: add integration test to attempt to exhaustively walk messages.

notmuch/Cargo.toml

@@ -8,7 +8,9 @@ edition = "2021"
 [dependencies]
 log = "0.4.14"
 serde = { version = "1.0", features = ["derive"] }
-serde_json = "1.0.68"
+serde_json = { version = "1.0", features = ["unbounded_depth"] }
 thiserror = "1.0.30"
+
 [dev-dependencies]
 pretty_assertions = "1"
+rayon = "1.5"

notmuch/src/lib.rs

@@ -208,9 +208,9 @@
 
 use std::{
     ffi::OsStr,
-    io,
+    io::{self, BufRead, BufReader, Lines},
     path::{Path, PathBuf},
-    process::Command,
+    process::{Child, ChildStdout, Command, Stdio},
 };
 
 use log::info;
@@ -356,7 +356,7 @@ pub struct Part {
     #[serde(skip_serializing_if = "Option::is_none")]
     #[serde(rename = "content-id")]
     pub content_id: Option<String>,
-    pub content: Content,
+    pub content: Option<Content>,
 }
 
 /// `encstatus = [{status: "good"|"bad"}]`
@@ -440,6 +440,10 @@ pub enum NotmuchError {
     Notmuch(#[from] io::Error),
     #[error("json decoding error")]
     SerdeJson(#[from] serde_json::Error),
+    #[error("failed to parse bytes as str")]
+    Utf8Error(#[from] std::str::Utf8Error),
+    #[error("failed to parse str as int")]
+    ParseIntError(#[from] std::num::ParseIntError),
 }
 
 #[derive(Default)]
@@ -467,9 +471,32 @@ impl Notmuch {
         Ok(serde_json::from_slice(&res)?)
     }
 
+    pub fn count(&self, query: &str) -> Result<usize, NotmuchError> {
+        let res = self.run_notmuch(["count", query])?;
+        // Strip '\n' from res.
+        let s = std::str::from_utf8(&res[..res.len() - 1])?;
+        Ok(s.parse()?)
+    }
+
     pub fn show(&self, query: &str) -> Result<ThreadSet, NotmuchError> {
-        let res = self.run_notmuch(["show", "--format=json", query])?;
-        Ok(serde_json::from_slice(&res)?)
+        let slice = self.run_notmuch(["show", "--format=json", query])?;
+        //
+        let s = String::from_utf8_lossy(&slice);
+        let mut deserializer = serde_json::Deserializer::from_str(&s);
+        deserializer.disable_recursion_limit();
+        let val = serde::de::Deserialize::deserialize(&mut deserializer)?;
+        deserializer.end()?;
+        Ok(val)
+    }
+
+    pub fn show_original(&self, id: &MessageId) -> Result<Vec<u8>, NotmuchError> {
+        let res = self.run_notmuch(["show", "--part=0", id])?;
+        Ok(res)
+    }
+
+    pub fn message_ids(&self, query: &str) -> Result<Lines<BufReader<ChildStdout>>, NotmuchError> {
+        let mut child = self.run_notmuch_pipe(["search", "--output=messages", query])?;
+        Ok(BufReader::new(child.stdout.take().unwrap()).lines())
     }
 
     fn run_notmuch<I, S>(&self, args: I) -> Result<Vec<u8>, NotmuchError>
@@ -483,10 +510,24 @@ impl Notmuch {
         }
         cmd.args(args);
         info!("{:?}", &cmd);
-        dbg!(&cmd);
         let out = cmd.output()?;
         Ok(out.stdout)
     }
+
+    fn run_notmuch_pipe<I, S>(&self, args: I) -> Result<Child, NotmuchError>
+    where
+        I: IntoIterator<Item = S>,
+        S: AsRef<OsStr>,
+    {
+        let mut cmd = Command::new("notmuch");
+        if let Some(config_path) = &self.config_path {
+            cmd.arg("--config").arg(config_path);
+        }
+        cmd.args(args);
+        info!("{:?}", &cmd);
+        let child = cmd.stdout(Stdio::piped()).spawn()?;
+        Ok(child)
+    }
 }
 
 #[cfg(test)]
@@ -523,6 +564,15 @@ mod tests {
         Ok(())
     }
 
+    #[test]
+    fn count() -> Result<(), NotmuchError> {
+        let nm = Notmuch::with_config("testdata/notmuch.config");
+        nm.new()?;
+        let c = nm.count("*")?;
+        assert_eq!(c, 12);
+        Ok(())
+    }
+
     #[test]
     fn thread_set_serde() {
         let ts = ThreadSet(vec![Thread(vec![ThreadNode(
@@ -537,17 +587,17 @@ mod tests {
                 body: Some(vec![Part {
                     id: 1.into(),
                     content_type: "multipart/mixed".to_string(),
-                    content: Content::Multipart(vec![
+                    content: Some(Content::Multipart(vec![
                         Part {
                             id: 2.into(),
                             content_type: "text/plain".to_string(),
-                            content: Content::String("Spam detection software".to_string()),
+                            content: Some(Content::String("Spam detection software".to_string())),
                             ..Default::default()
                         },
                         Part {
                             id: 3.into(),
                             content_type: "message/rfc822".to_string(),
-                            content: Content::Rfc822(vec![Rfc822 {
+                            content: Some(Content::Rfc822(vec![Rfc822 {
                                 headers: Headers {
                                     subject: "Re: Registration goof".to_string(),
                                     from: "\"Bill Thiede\" <Bill@xinu.tv>".to_string(),
@@ -561,13 +611,13 @@ mod tests {
                                 body: vec![Part {
                                     id: 4.into(),
                                     content_type: "text/plain".to_string(),
-                                    content: Content::String("Hello".to_string()),
+                                    content: Some(Content::String("Hello".to_string())),
                                     ..Default::default()
                                 }],
-                            }]),
+                            }])),
                             ..Default::default()
                         },
-                    ]),
+                    ])),
                     ..Default::default()
                 }]),
                 headers: Headers {

notmuch/tests/allmail.rs

@@ -0,0 +1,53 @@
+use std::{
+    error::Error,
+    io::{stdout, Write},
+    time::{Duration, Instant},
+};
+
+use rayon::iter::{ParallelBridge, ParallelIterator};
+
+use notmuch::{Notmuch, NotmuchError, SearchSummary, ThreadSet};
+
+#[test]
+fn parse() -> Result<(), Box<dyn Error>> {
+    // take_hook() returns the default hook in case when a custom one is not set
+    let orig_hook = std::panic::take_hook();
+    std::panic::set_hook(Box::new(move |panic_info| {
+        // invoke the default handler and exit the process
+        orig_hook(panic_info);
+        std::process::exit(1);
+    }));
+
+    let nm = Notmuch::default();
+    let count = nm.count("*")? as f32;
+    let start = Instant::now();
+    nm.message_ids("*")?
+        .enumerate()
+        .par_bridge()
+        .for_each(|(i, msg)| {
+            let msg = msg.expect("failed to unwrap msg");
+            let ts = nm
+                .show(&msg)
+                .expect(&format!("failed to show msg: {}", msg));
+            //println!("{:?}", ts);
+            if i > 0 && i % 1000 == 0 {
+                let diff = start.elapsed();
+                let percent = (i as f32 * 100.) / count;
+                let eta = diff.mul_f32(count as f32).div_f32(i as f32);
+                print!(
+                    "\nElapsed {}s ETA {}s Percent {}% ",
+                    diff.as_secs_f32(),
+                    eta.as_secs_f32(),
+                    percent
+                );
+                stdout().flush().expect("failed to flush stdout");
+            }
+            if i % 10 == 0 {
+                print!(".");
+                stdout().flush().expect("failed to flush stdout");
+            }
+        });
+    println!("\n");
+    assert!(false);
+    Ok(())
+}