Merge pull request #17 from lucab/to-upstream/journal-matches

journal: add support for match filters

src/journal.rs

@@ -3,6 +3,7 @@ use log::{self, Log, LogRecord, LogLocation, LogLevelFilter, SetLoggerError};
 use std::{fmt, io, ptr, result};
 use std::collections::BTreeMap;
 use std::io::ErrorKind::InvalidData;
+use std::os::raw::c_void;
 use ffi::array_to_iovecs;
 use ffi::id128::sd_id128_t;
 use ffi::journal as ffi;
@@ -152,7 +153,7 @@ impl Journal {
 
     /// Read the next record from the journal. Returns `Ok(None)` if there
     /// are no more records to read.
-    pub fn next_record(&self) -> Result<Option<JournalRecord>> {
+    pub fn next_record(&mut self) -> Result<Option<JournalRecord>> {
         if sd_try!(ffi::sd_journal_next(self.j)) == 0 {
             return Ok(None);
         }
@@ -178,7 +179,7 @@ impl Journal {
 
     /// Seek to a specific position in journal. On success, returns a cursor
     /// to the current entry.
-    pub fn seek(&self, seek: JournalSeek) -> Result<String> {
+    pub fn seek(&mut self, seek: JournalSeek) -> Result<String> {
         match seek {
             JournalSeek::Head => sd_try!(ffi::sd_journal_seek_head(self.j)),
             JournalSeek::Current => 0,
@@ -225,6 +226,38 @@ impl Journal {
         sd_try!(ffi::sd_journal_get_realtime_usec(self.j, &mut timestamp_us));
         Ok(system_time_from_realtime_usec(timestamp_us))
     }
+
+    /// Adds a match by which to filter the entries of the journal.
+    /// If a match is applied, only entries with this field set will be iterated.
+    pub fn match_add<T: Into<Vec<u8>>>(&mut self, key: &str, val: T) -> Result<&mut Journal> {
+        let mut filter = Vec::<u8>::from(key);
+        filter.push('=' as u8);
+        filter.extend(val.into());
+        let data = filter.as_ptr() as *const c_void;
+        let datalen = filter.len() as size_t;
+        sd_try!(ffi::sd_journal_add_match(self.j, data, datalen));
+        Ok(self)
+    }
+
+    /// Inserts a disjunction (i.e. logical OR) in the match list.
+    pub fn match_or(&mut self) -> Result<&mut Journal> {
+        sd_try!(ffi::sd_journal_add_disjunction(self.j));
+        Ok(self)
+    }
+
+    /// Inserts a conjunction (i.e. logical AND) in the match list.
+    pub fn match_and(&mut self) -> Result<&mut Journal> {
+        sd_try!(ffi::sd_journal_add_conjunction(self.j));
+        Ok(self)
+    }
+
+    /// Flushes all matches, disjunction and conjunction terms.
+    /// After this call all filtering is removed and all entries in
+    /// the journal will be iterated again.
+    pub fn match_flush(&mut self) -> Result<&mut Journal> {
+        unsafe { ffi::sd_journal_flush_matches(self.j) };
+        Ok(self)
+    }
 }
 
 impl Drop for Journal {

tests/journal.rs

@@ -35,7 +35,7 @@ fn cursor() {
         return;
     }
 
-    let j = journal::Journal::open(journal::JournalFiles::All, false, false).unwrap();
+    let mut j = journal::Journal::open(journal::JournalFiles::All, false, false).unwrap();
     log!(log::LogLevel::Info, "rust-systemd test_seek entry");
     assert!(j.seek(journal::JournalSeek::Head).is_ok());
     let _s = j.cursor().unwrap();
@@ -47,7 +47,7 @@ fn ts() {
         return;
     }
 
-    let j = journal::Journal::open(journal::JournalFiles::All, false, false).unwrap();
+    let mut j = journal::Journal::open(journal::JournalFiles::All, false, false).unwrap();
     log!(log::LogLevel::Info, "rust-systemd test_seek entry");
     assert!(j.seek(journal::JournalSeek::Head).is_ok());
     let _s = j.timestamp().unwrap();
@@ -56,7 +56,7 @@ fn ts() {
 
 #[test]
 fn test_seek() {
-    let j = journal::Journal::open(journal::JournalFiles::All, false, false).unwrap();
+    let mut j = journal::Journal::open(journal::JournalFiles::All, false, false).unwrap();
     if ! have_journal() {
         return;
     }
@@ -71,3 +71,31 @@ fn test_seek() {
     assert!(j.seek(journal::JournalSeek::Tail).is_ok());
     assert!(j.next_record().is_ok());
 }
+
+#[test]
+fn test_simple_match() {
+    if ! have_journal() {
+        return;
+    }
+    let key = "RUST_TEST_MARKER";
+    let value = "RUST_SYSTEMD_SIMPLE_MATCH";
+    let msg = "MESSAGE=rust-systemd test_match";
+    let filter = format!("{}={}", key, value);
+    let mut j = journal::Journal::open(journal::JournalFiles::All, false, false).unwrap();
+
+    // check for positive matches
+    assert!(j.seek(journal::JournalSeek::Tail).is_ok());
+    journal::send(&[&filter, &msg]);
+    assert!(j.match_flush().unwrap().match_add(key, value).is_ok());
+    let r = j.next_record().unwrap();
+    assert!(r.is_some());
+    let entry = r.unwrap();
+    let entryval = entry.get(key);
+    assert!(entryval.is_some());
+    assert_eq!(entryval.unwrap(), value);
+
+    // check for negative matches
+    assert!(j.seek(journal::JournalSeek::Tail).is_ok());
+    journal::send(&[&msg]);
+    assert!(j.next_record().unwrap().is_none());
+}