[READ-ONLY] Mirror of https://github.com/probablykasper/csv-pipeline.
Fix FnMut lifetime issue
+104
-64
+22
-24
src/chain.rs
+22
-24
src/chain.rs
···
2
2
pub type BoxedIterator = Box<dyn Iterator<Item = RowResult>>;
3
3
4
4
/// A struct that wraps a RowResult iterator for convenience
5
-
pub struct Chain {
6
-
iterator: BoxedIterator,
5
+
pub struct Chain<C: Iterator<Item = RowResult>> {
6
+
iterator: C,
7
7
}
8
-
impl Chain {
9
-
pub fn new(iterator: BoxedIterator) -> Self {
10
-
Self {
11
-
iterator: Box::new(iterator),
12
-
}
8
+
impl<C: Iterator<Item = RowResult>> Chain<C> {
9
+
pub fn new(iterator: C) -> Self {
10
+
Self { iterator: iterator }
13
11
}
14
-
pub fn with_state<S>(self, state: S) -> StatefulChainBuilder<S> {
15
-
StatefulChainBuilder::new(Box::new(self), state)
12
+
pub fn with_state<S>(self, state: S) -> StatefulChainBuilder<C, S> {
13
+
StatefulChainBuilder::new(self.iterator, state)
16
14
}
17
15
}
18
-
impl Iterator for Chain {
16
+
impl<C: Iterator<Item = RowResult>> Iterator for Chain<C> {
19
17
type Item = RowResult;
20
18
21
19
fn next(&mut self) -> Option<Self::Item> {
···
23
21
}
24
22
}
25
23
26
-
pub struct StatefulChainBuilder<S> {
27
-
iterator: BoxedIterator,
24
+
pub struct StatefulChainBuilder<C: Iterator<Item = RowResult>, S> {
25
+
iterator: C,
28
26
state: S,
29
27
}
30
-
impl<S> StatefulChainBuilder<S> {
31
-
pub fn new(iterator: BoxedIterator, state: S) -> Self {
28
+
impl<C: Iterator<Item = RowResult>, S> StatefulChainBuilder<C, S> {
29
+
pub fn new(iterator: C, state: S) -> Self {
32
30
Self { state, iterator }
33
31
}
34
-
pub fn map<F>(self, f: F) -> Chain
32
+
pub fn map<F>(self, f: F) -> StatefulChain<C, S, F>
35
33
where
36
34
F: FnMut(RowResult, &mut S) -> RowResult,
37
35
{
38
-
let x = StatefulChain {
36
+
StatefulChain {
39
37
iterator: self.iterator,
40
38
state: self.state,
41
39
f,
42
-
};
43
-
x.into_chain()
40
+
}
44
41
}
45
42
}
46
43
47
-
pub struct StatefulChain<S, F: FnMut(RowResult, &mut S) -> RowResult> {
48
-
iterator: BoxedIterator,
44
+
pub struct StatefulChain<C: Iterator<Item = RowResult>, S, F: FnMut(RowResult, &mut S) -> RowResult>
45
+
{
46
+
iterator: C,
49
47
state: S,
50
48
f: F,
51
49
}
52
-
impl<S, F> StatefulChain<S, F>
50
+
impl<C: Iterator<Item = RowResult>, S, F> StatefulChain<C, S, F>
53
51
where
54
52
F: FnMut(RowResult, &mut S) -> RowResult,
55
53
{
56
-
pub fn into_chain(self) -> Chain {
57
-
Chain::new(Box::new(self))
54
+
pub fn into_chain(self) -> Chain<StatefulChain<C, S, F>> {
55
+
Chain::new(self)
58
56
}
59
57
}
60
-
impl<S, F> Iterator for StatefulChain<S, F>
58
+
impl<C: Iterator<Item = RowResult>, S, F> Iterator for StatefulChain<C, S, F>
61
59
where
62
60
F: FnMut(RowResult, &mut S) -> RowResult,
63
61
{
+23
src/lib.rs
+23
src/lib.rs
···
2
2
mod headers;
3
3
mod pipeline;
4
4
5
+
use std::io;
6
+
7
+
use csv::StringRecordsIntoIter;
5
8
pub use headers::Headers;
6
9
pub use pipeline::{Pipeline, PipelineBuilder};
7
10
···
13
16
14
17
pub type Row = csv::StringRecord;
15
18
pub type RowResult = Result<Row, Error>;
19
+
20
+
pub struct RowIter<R: io::Read> {
21
+
inner: StringRecordsIntoIter<R>,
22
+
}
23
+
impl<R: io::Read> RowIter<R> {
24
+
pub fn from_records(records: StringRecordsIntoIter<R>) -> Self {
25
+
RowIter { inner: records }
26
+
}
27
+
}
28
+
impl<R: io::Read> Iterator for RowIter<R> {
29
+
type Item = RowResult;
30
+
31
+
fn next(&mut self) -> Option<Self::Item> {
32
+
self.inner.next().map(|result| {
33
+
result.map_err(|err| {
34
+
return Error::Csv;
35
+
})
36
+
})
37
+
}
38
+
}
+59
-40
src/pipeline.rs
+59
-40
src/pipeline.rs
···
1
-
use super::chain::{BoxedIterator, Chain};
2
1
use super::headers::Headers;
3
-
use crate::{Error, Row, RowResult};
2
+
use crate::{Error, Row, RowIter, RowResult};
4
3
use csv::{Reader, ReaderBuilder};
5
4
use std::fs::File;
6
5
use std::path::Path;
7
6
8
-
pub struct PipelineBuilder {
7
+
pub struct PipelineBuilder<'a> {
9
8
pub headers: Headers,
10
-
chain: Chain,
9
+
iterator: Box<dyn Iterator<Item = RowResult> + 'a>,
11
10
}
12
11
13
-
impl PipelineBuilder {
12
+
impl<'a> PipelineBuilder<'a> {
14
13
pub fn from_reader(mut reader: Reader<File>) -> Self {
15
14
let headers_row = reader.headers().unwrap().clone();
16
-
let records = reader.into_records().map(|r| {
17
-
let row_result: RowResult = match r {
18
-
Ok(row) => Ok(row),
19
-
Err(err) => Err(Error::Csv),
20
-
};
21
-
row_result
22
-
});
23
-
Self {
15
+
let row_iterator = RowIter::from_records(reader.into_records());
16
+
PipelineBuilder {
24
17
headers: Headers::from(headers_row),
25
-
chain: Chain::new(Box::new(records)),
18
+
iterator: Box::new(row_iterator),
26
19
}
27
20
}
28
21
···
55
48
/// ```
56
49
pub fn add_col<F>(mut self, name: &str, get_value: F) -> Self
57
50
where
58
-
F: FnMut(&Headers, &Row) -> Result<String, Error>,
51
+
F: FnMut(&Headers, &Row) -> Result<String, Error> + 'a,
59
52
{
60
53
self.headers.push_field(name);
61
54
62
-
struct State<F> {
63
-
get_value: F,
64
-
headers: Headers,
65
-
}
66
-
let stateful_chain = self.chain.with_state(State {
67
-
get_value,
55
+
let add_col = AddCol {
56
+
iterator: self.iterator,
57
+
f: get_value,
68
58
headers: self.headers.clone(),
69
-
});
70
-
let new_chain = stateful_chain.map(|row_result, state| {
71
-
println!("ADDCOL-map");
72
-
let mut row = row_result?;
73
-
let value = (state.get_value)(&state.headers, &row)?;
74
-
row.push_field(&value);
75
-
Ok(row)
76
-
});
59
+
};
77
60
78
-
self.chain = new_chain;
61
+
self.iterator = Box::new(add_col);
79
62
80
63
self
81
64
}
82
65
83
-
pub fn build(self) -> Pipeline {
66
+
pub fn build(self) -> Pipeline<'a> {
84
67
Pipeline {
85
68
headers: self.headers,
86
-
iterator: Box::new(self.chain),
69
+
iterator: Box::new(self.iterator),
70
+
}
71
+
}
72
+
}
73
+
74
+
struct AddCol<I, F: FnMut(&Headers, &Row) -> Result<String, Error>> {
75
+
iterator: I,
76
+
f: F,
77
+
headers: Headers,
78
+
}
79
+
impl<I, F> Iterator for AddCol<I, F>
80
+
where
81
+
I: Iterator<Item = RowResult>,
82
+
F: FnMut(&Headers, &Row) -> Result<String, Error>,
83
+
{
84
+
type Item = RowResult;
85
+
86
+
fn next(&mut self) -> Option<Self::Item> {
87
+
let row = match self.iterator.next() {
88
+
Some(Ok(row)) => row,
89
+
Some(Err(e)) => return Some(Err(e)),
90
+
None => return None,
91
+
};
92
+
match (self.f)(&self.headers, &row) {
93
+
Ok(value) => {
94
+
let mut row = row;
95
+
row.push_field(&value);
96
+
Some(Ok(row))
97
+
}
98
+
Err(e) => Some(Err(e)),
87
99
}
88
100
}
89
101
}
90
102
91
-
pub struct Pipeline {
103
+
pub struct Pipeline<'a> {
92
104
pub headers: Headers,
93
-
pub iterator: BoxedIterator,
105
+
pub iterator: Box<dyn Iterator<Item = RowResult> + 'a>,
94
106
}
95
-
impl Iterator for Pipeline {
107
+
impl<'a> Iterator for Pipeline<'a> {
96
108
type Item = RowResult;
97
109
98
110
fn next(&mut self) -> Option<Self::Item> {
···
107
119
#[test]
108
120
fn add_col() {
109
121
let mut pipeline = PipelineBuilder::from_path("test/Countries.csv")
110
-
.add_col("Language", |_headers, _row| Ok("".to_string()))
122
+
.add_col("Language", |_headers, row| match row.get(1) {
123
+
Some("Norway") => Ok("Norwegian".to_string()),
124
+
_ => Ok("Unknown".to_string()),
125
+
})
111
126
.build();
112
127
113
128
let mut writer = csv::Writer::from_writer(vec![]);
114
129
writer.write_record(&pipeline.headers).unwrap();
115
-
println!("{:?}", pipeline.headers.get_row());
116
130
while let Some(item) = pipeline.next() {
117
-
println!("{:?}", item.clone().unwrap());
118
131
writer.write_record(&item.unwrap()).unwrap();
119
132
}
120
-
let s = String::from_utf8(writer.into_inner().unwrap()).unwrap();
121
-
print!("{}", s);
133
+
let csv_str = String::from_utf8(writer.into_inner().unwrap()).unwrap();
134
+
135
+
assert_eq!(
136
+
csv_str,
137
+
"ID,Country,Language\n\
138
+
1,Norway,Norwegian\n\
139
+
2,Tuvalu,Unknown\n"
140
+
);
122
141
}
123
142
}