diff --git a/src/chain.rs b/src/chain.rs new file mode 100644 index 0000000..1d6e0cb --- /dev/null +++ b/src/chain.rs @@ -0,0 +1,72 @@ +use crate::RowResult; +pub type BoxedIterator = Box>; + +/// A struct that wraps a RowResult iterator for convenience +pub struct Chain { + iterator: BoxedIterator, +} +impl Chain { + pub fn new(iterator: BoxedIterator) -> Self { + Self { + iterator: Box::new(iterator), + } + } + pub fn with_state(self, state: S) -> StatefulChainBuilder { + StatefulChainBuilder::new(Box::new(self), state) + } +} +impl Iterator for Chain { + type Item = RowResult; + + fn next(&mut self) -> Option { + self.iterator.next() + } +} + +pub struct StatefulChainBuilder { + iterator: BoxedIterator, + state: S, +} +impl StatefulChainBuilder { + pub fn new(iterator: BoxedIterator, state: S) -> Self { + Self { state, iterator } + } + pub fn map(self, f: F) -> Chain + where + F: FnMut(RowResult, &mut S) -> RowResult, + { + let x = StatefulChain { + iterator: self.iterator, + state: self.state, + f, + }; + x.into_chain() + } +} + +pub struct StatefulChain RowResult> { + iterator: BoxedIterator, + state: S, + f: F, +} +impl StatefulChain +where + F: FnMut(RowResult, &mut S) -> RowResult, +{ + pub fn into_chain(self) -> Chain { + Chain::new(Box::new(self)) + } +} +impl Iterator for StatefulChain +where + F: FnMut(RowResult, &mut S) -> RowResult, +{ + type Item = RowResult; + + fn next(&mut self) -> Option { + match self.iterator.next() { + Some(item) => Some((self.f)(item, &mut self.state)), + None => None, + } + } +} diff --git a/src/headers.rs b/src/headers.rs index 0e744aa..2251187 100644 --- a/src/headers.rs +++ b/src/headers.rs @@ -1,19 +1,20 @@ use crate::Row; +use csv::StringRecordIter; use std::collections::HashMap; #[derive(Debug, Clone, PartialEq)] pub struct Headers { indexes: HashMap, - names: Row, + row: Row, } impl Headers { - pub fn add(&mut self, name: &str) -> bool { + pub fn push_field(&mut self, name: &str) -> bool { if self.indexes.contains_key(name) { return false; } - self.names.push_field(name); - self.indexes.insert(name.to_string(), self.names.len() - 1); + self.row.push_field(name); + self.indexes.insert(name.to_string(), self.row.len() - 1); true } @@ -21,6 +22,10 @@ impl Headers { pub fn contains(&self, name: &str) -> bool { self.indexes.contains_key(name) } + + pub fn get_row(&self) -> &Row { + &self.row + } } impl From for Headers { fn from(row: Row) -> Headers { @@ -30,7 +35,20 @@ impl From for Headers { .enumerate() .map(|(index, entry)| (entry.to_string(), index)) .collect(), - names: row, + row, } } } +impl<'a> IntoIterator for &'a Headers { + type Item = &'a str; + type IntoIter = StringRecordIter<'a>; + + fn into_iter(self) -> StringRecordIter<'a> { + self.row.into_iter() + } +} +impl From for Row { + fn from(headers: Headers) -> Row { + headers.row + } +} diff --git a/src/lib.rs b/src/lib.rs index c979b44..51b29e6 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -1,10 +1,9 @@ -pub mod headers; -pub mod pipe; -pub mod pipeline; +mod chain; +mod headers; +mod pipeline; pub use headers::Headers; -pub use pipe::Pipe; -pub use pipeline::Pipeline; +pub use pipeline::{Pipeline, PipelineBuilder}; #[derive(Debug, Clone, PartialEq)] pub enum Error { diff --git a/src/pipe.rs b/src/pipe.rs deleted file mode 100644 index 19281df..0000000 --- a/src/pipe.rs +++ /dev/null @@ -1,63 +0,0 @@ -use crate::RowResult; - -pub type PipeIterator = Box>; - -pub struct Pipe { - iterator: PipeIterator, -} -impl Pipe { - pub fn new(iterator: PipeIterator) -> Self { - Self { - iterator: Box::new(iterator.into_iter()), - } - } - pub fn with_state(self, state: S) -> StatefulPipeBuilder { - StatefulPipeBuilder::new(self.iterator, state) - } -} -impl Iterator for Pipe { - type Item = RowResult; - - fn next(&mut self) -> Option { - self.iterator.next() - } -} - -pub struct StatefulPipeBuilder { - iterator: PipeIterator, - state: S, -} -impl StatefulPipeBuilder { - pub fn new(iterator: PipeIterator, state: S) -> Self { - Self { state, iterator } - } - pub fn map(self, f: F) -> StatefulPipe - where - F: FnMut(RowResult, &mut S) -> RowResult, - { - StatefulPipe { - iterator: self.iterator, - state: self.state, - f, - } - } -} - -pub struct StatefulPipe RowResult> { - pub(crate) iterator: PipeIterator, - state: S, - f: F, -} -impl Iterator for StatefulPipe -where - F: FnMut(RowResult, &mut S) -> RowResult, -{ - type Item = RowResult; - - fn next(&mut self) -> Option { - match self.iterator.next() { - Some(item) => Some((self.f)(item, &mut self.state)), - None => None, - } - } -} diff --git a/src/pipeline.rs b/src/pipeline.rs index a50c140..51a5d79 100644 --- a/src/pipeline.rs +++ b/src/pipeline.rs @@ -1,16 +1,16 @@ +use super::chain::{BoxedIterator, Chain}; use super::headers::Headers; -use super::pipe::{Pipe, PipeIterator}; use crate::{Error, Row, RowResult}; use csv::{Reader, ReaderBuilder}; use std::fs::File; use std::path::Path; -pub struct Pipeline { +pub struct PipelineBuilder { pub headers: Headers, - pipe: PipeIterator, + chain: Chain, } -impl Pipeline { +impl PipelineBuilder { pub fn from_reader(mut reader: Reader) -> Self { let headers_row = reader.headers().unwrap().clone(); let records = reader.into_records().map(|r| { @@ -22,7 +22,7 @@ impl Pipeline { }); Self { headers: Headers::from(headers_row), - pipe: Box::new(records), + chain: Chain::new(Box::new(records)), } } @@ -46,9 +46,9 @@ impl Pipeline { /// ## Example /// /// ``` - /// use csv_pipeline::Pipeline; + /// use csv_pipeline::PipelineBuilder; /// - /// Pipeline::from_path("test/Countries.csv") + /// PipelineBuilder::from_path("test/Countries.csv") /// .add_col("Language", |headers, row| { /// Ok("".to_string()) /// }); @@ -57,36 +57,67 @@ impl Pipeline { where F: FnMut(&Headers, &Row) -> Result, { - self.headers.add(name); + self.headers.push_field(name); struct State { get_value: F, headers: Headers, } - let pipe = Pipe::new(self.pipe).with_state(State { + let stateful_chain = self.chain.with_state(State { get_value, headers: self.headers.clone(), }); - let newpipe = pipe.map(|row_result, state| { + let new_chain = stateful_chain.map(|row_result, state| { + println!("ADDCOL-map"); let mut row = row_result?; let value = (state.get_value)(&state.headers, &row)?; row.push_field(&value); Ok(row) }); - self.pipe = Box::new(newpipe.iterator); + self.chain = new_chain; self } + + pub fn build(self) -> Pipeline { + Pipeline { + headers: self.headers, + iterator: Box::new(self.chain), + } + } +} + +pub struct Pipeline { + pub headers: Headers, + pub iterator: BoxedIterator, +} +impl Iterator for Pipeline { + type Item = RowResult; + + fn next(&mut self) -> Option { + self.iterator.next() + } } #[cfg(test)] mod tests { - use crate::Pipeline; + use crate::PipelineBuilder; #[test] fn add_col() { - let mut pipeline = Pipeline::from_path("test/Countries.csv") - .add_col("Language", |_headers, _row| Ok("".to_string())); + let mut pipeline = PipelineBuilder::from_path("test/Countries.csv") + .add_col("Language", |_headers, _row| Ok("".to_string())) + .build(); + + let mut writer = csv::Writer::from_writer(vec![]); + writer.write_record(&pipeline.headers).unwrap(); + println!("{:?}", pipeline.headers.get_row()); + while let Some(item) = pipeline.next() { + println!("{:?}", item.clone().unwrap()); + writer.write_record(&item.unwrap()).unwrap(); + } + let s = String::from_utf8(writer.into_inner().unwrap()).unwrap(); + print!("{}", s); } }