add a filter to trim whitespace

Author: cammeresi

src/filter/mod.rs

@@ -1,6 +1,6 @@
 //! Optional ADIF data transformations
 
-use crate::{Error, Record};
+use crate::{Datum, Error, Record};
 use chrono::{Days, NaiveDateTime};
 use futures::stream::Stream;
 use std::collections::HashSet;
@@ -99,6 +99,48 @@ pub trait FilterExt: Stream {
 
 impl<S> FilterExt for S where S: Stream {}
 
+/// Stream adapter that transforms each record by consuming and rebuilding it.
+pub struct Map<S, F> {
+    stream: S,
+    f: F,
+}
+
+impl<S, F> Stream for Map<S, F>
+where
+    S: Stream<Item = Result<Record, Error>> + Unpin,
+    F: FnMut(Record) -> Result<Record, Error> + Unpin,
+{
+    type Item = Result<Record, Error>;
+
+    fn poll_next(
+        self: Pin<&mut Self>, cx: &mut Context<'_>,
+    ) -> Poll<Option<Self::Item>> {
+        let this = self.get_mut();
+        match Pin::new(&mut this.stream).poll_next(cx) {
+            Poll::Ready(Some(Ok(record))) => {
+                Poll::Ready(Some((this.f)(record)))
+            }
+            Poll::Ready(Some(Err(e))) => Poll::Ready(Some(Err(e))),
+            Poll::Ready(None) => Poll::Ready(None),
+            Poll::Pending => Poll::Pending,
+        }
+    }
+}
+
+/// Extension trait providing the `map` method on streams.
+pub trait MapExt: Stream {
+    /// Transform each record by consuming and rebuilding it.
+    fn map<F>(self, f: F) -> Map<Self, F>
+    where
+        Self: Sized,
+        F: FnMut(Record) -> Result<Record, Error>,
+    {
+        Map { stream: self, f }
+    }
+}
+
+impl<S> MapExt for S where S: Stream {}
+
 /// Normalize date and time fields from multiple possible source fields into
 /// combined datetime values.
 ///
@@ -282,3 +324,35 @@ where
 {
     stream.filter(|record| !record.is_header())
 }
+
+/// Trim leading and trailing whitespace from string field values.
+///
+/// Non-string datum variants pass through unchanged.
+///
+/// ```
+/// use difa::{
+///     Record, RecordStreamExt, TagDecoder, filter::trim_whitespace,
+/// };
+/// use futures::StreamExt;
+///
+/// # tokio_test::block_on(async {
+/// let data = b"<call:6> W1AW <eor>";
+/// let stream = TagDecoder::new_stream(&data[..], true).records();
+/// let mut stream = trim_whitespace(stream);
+/// let record = stream.next().await.unwrap().unwrap();
+/// assert_eq!(record.get("call").unwrap().as_str(), "W1AW");
+/// # });
+/// ```
+pub fn trim_whitespace<S>(
+    stream: S,
+) -> Map<S, impl FnMut(Record) -> Result<Record, Error>>
+where
+    S: Stream<Item = Result<Record, Error>>,
+{
+    stream.map(|r| {
+        Ok(r.map_fields(|_, v| match v {
+            Datum::String(s) => Datum::String(s.trim().to_string()),
+            other => other,
+        }))
+    })
+}

src/filter/test.rs

@@ -456,3 +456,65 @@ async fn filter_end_of_stream() {
     assert_eq!(rec.get("call").unwrap().as_str(), "W1AW");
     no_record(&mut s).await;
 }
+
+#[tokio::test]
+async fn trim_whitespace_trims_strings() {
+    let rec = parse_one("<call:6> W1AW <eor>", trim_whitespace).await;
+    assert_eq!(rec.get("call").unwrap().as_str(), "W1AW");
+}
+
+#[tokio::test]
+async fn trim_whitespace_no_change() {
+    let rec = parse_one("<call:4>W1AW<eor>", trim_whitespace).await;
+    assert_eq!(rec.get("call").unwrap().as_str(), "W1AW");
+}
+
+#[tokio::test]
+async fn trim_whitespace_preserves_typed_values() {
+    let rec =
+        parse_one("<n:5:n> 1.0 <b:3:b> Y <call:6> W1AW <eor>", trim_whitespace)
+            .await;
+    assert_eq!(
+        rec.get("n").unwrap().as_number().unwrap().to_string(),
+        "1.0"
+    );
+    assert!(rec.get("b").unwrap().as_bool().unwrap());
+    assert_eq!(rec.get("call").unwrap().as_str(), "W1AW");
+}
+
+#[tokio::test]
+async fn trim_whitespace_preserves_header() {
+    let mut s = parse_many(
+        "<adifver:7> 3.1.4 <eoh><call:6> W1AW <eor>",
+        trim_whitespace,
+    );
+    let rec = next(&mut s).await;
+    assert!(rec.is_header());
+    assert_eq!(rec.get("adifver").unwrap().as_str(), "3.1.4");
+    let rec = next(&mut s).await;
+    assert!(!rec.is_header());
+    assert_eq!(rec.get("call").unwrap().as_str(), "W1AW");
+    no_record(&mut s).await;
+}
+
+#[tokio::test]
+async fn trim_whitespace_multiple_fields() {
+    let rec = parse_one(
+        "<call:6> W1AW <band:5> 20m <mode:5> SSB <eor>",
+        trim_whitespace,
+    )
+    .await;
+    assert_eq!(rec.get("call").unwrap().as_str(), "W1AW");
+    assert_eq!(rec.get("band").unwrap().as_str(), "20m");
+    assert_eq!(rec.get("mode").unwrap().as_str(), "SSB");
+}
+
+#[tokio::test]
+async fn trim_whitespace_error_passthrough() {
+    let mut s =
+        parse_many_ignore("<call:6> W1AW <eor><bad", false, trim_whitespace);
+    let rec = next(&mut s).await;
+    assert_eq!(rec.get("call").unwrap().as_str(), "W1AW");
+    next_err(&mut s, partial_data(1, 20, 19)).await;
+    no_record(&mut s).await;
+}

src/lib.rs

@@ -28,7 +28,7 @@ mod test;
 
 pub use cabrillo::CabrilloSink;
 pub use cistring::{CiStr, CiString};
-pub use filter::{FilterExt, NormalizeExt};
+pub use filter::{FilterExt, MapExt, NormalizeExt};
 pub use parse::{RecordStream, RecordStreamExt, TagDecoder, TagStream};
 pub use write::{OutputTypes, RecordSink, TagEncoder, TagSink, TagSinkExt};
 
@@ -509,6 +509,25 @@ impl Record {
         self.fields.into_iter().map(|(k, v)| (k.into_string(), v))
     }
 
+    /// Rebuild a record by applying a transformation to each field.
+    pub fn map_fields<F>(self, mut f: F) -> Self
+    where
+        F: FnMut(&str, Datum) -> Datum,
+    {
+        let fields = self
+            .fields
+            .into_iter()
+            .map(|(k, v)| {
+                let v = f(k.as_str(), v);
+                (k, v)
+            })
+            .collect();
+        Self {
+            header: self.header,
+            fields,
+        }
+    }
+
     /// Return an iterator over all fields in this record.
     pub fn fields(&self) -> impl Iterator<Item = (&str, &Datum)> {
         self.fields.iter().map(|(k, v)| (k.as_str(), v))

src/test.rs

@@ -194,6 +194,36 @@ fn into_fields() {
     assert_eq!(fields[1].1.as_number().unwrap(), Decimal::from(14));
 }
 
+#[test]
+fn map_fields() {
+    let mut r = Record::new();
+    r.insert("call", " W1AW ").unwrap();
+    r.insert("freq", Decimal::from(14)).unwrap();
+
+    let r = r.map_fields(|_, v| match v {
+        Datum::String(s) => Datum::String(s.trim().to_string()),
+        other => other,
+    });
+    assert_eq!(r.get("call").unwrap().as_str(), "W1AW");
+    assert_eq!(
+        r.get("freq").unwrap().as_number().unwrap(),
+        Decimal::from(14)
+    );
+}
+
+#[test]
+fn map_fields_header() {
+    let mut r = Record::new_header();
+    r.insert("adifver", " 3.1.4 ").unwrap();
+    r.insert("foo", true).unwrap();
+    let r = r.map_fields(|_, v| match v {
+        Datum::String(s) => Datum::String(s.trim().to_string()),
+        other => other,
+    });
+    assert!(r.is_header());
+    assert_eq!(r.get("adifver").unwrap().as_str(), "3.1.4");
+}
+
 #[test]
 fn to_cabrillo() {
     let s = Datum::String("test".to_string());