1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143

use std::{
    vec,
    collections::HashMap,
};

use crate::{
    add, col, add_with, error, reduce,

    Add, ColSpec, Flush, Headers, Inspect, Reduce, Row, RowResult, AddWith,
    Del, Group, AdjacentGroup, MockStream, Rename, GroupBuildError, MapRow, MapCol,

    aggregate::Aggregate,
    flush::FlushTarget,
};

/// This trait describes de behaviour of every component in the CSV transformation
/// chain. Functions provided by this trait help construct the chain and can be
/// _chained_.
///
/// Implement this trait to extend `csvsc` with your own processors.
pub trait RowStream: IntoIterator<Item = RowResult> {

    /// Must return headers as they are in this point of the chain. For example
    /// if implementor adds a column, its `headers()` function must return the
    /// new headers including the one just added.
    fn headers(&self) -> &Headers;

    /// Allows adding columns to each row of the stream.
    fn add(self, column: ColSpec) -> Result<Add<Self>, add::BuildError>
    where
        Self: Sized,
    {
        Add::new(self, column)
    }

    /// Deletes the specified columns from each row of the stream
    fn del(self, columns: Vec<&str>) -> Del<Self>
    where
        Self: Sized,
    {
        Del::new(self, columns)
    }

    /// Adds a column to each row of the stream using a closure to compute its
    /// value
    fn add_with<F>(self, colname: &str, f: F) -> Result<AddWith<Self, F>, add_with::BuildError>
    where
        Self: Sized,
        F: FnMut(&Headers, &Row) -> Result<String, col::BuildError>,
    {
        AddWith::new(self, colname, f)
    }

    /// Group by one or more columns, compute aggregates and output the
    /// resulting columns
    fn reduce(
        self,
        columns: Vec<Box<dyn Aggregate>>,
    ) -> Result<Reduce<Self>, reduce::BuildError>
    where
        Self: Sized,
    {
        Reduce::new(self, columns)
    }

    fn adjacent_group<H, F, R>(
        self,
        header_map: H,
        f: F,
        grouping: &[&str],
    ) -> Result<AdjacentGroup<Self, F>, GroupBuildError>
    where
        H: FnMut(Headers) -> Headers,
        F: FnMut(MockStream<vec::IntoIter<RowResult>>) -> R,
        R: RowStream,
        Self: Sized,
    {
        AdjacentGroup::new(self, header_map, f, grouping)
    }

    fn group<H, F, R>(
        self,
        header_map: H,
        f: F,
        grouping: &[&str],
    ) -> Result<Group<Self, F>, GroupBuildError>
    where
        H: FnMut(Headers) -> Headers,
        F: FnMut(MockStream<vec::IntoIter<RowResult>>) -> R,
        R: RowStream,
        Self: Sized,
    {
        Group::new(self, header_map, f, grouping)
    }

    /// When consumed, writes to destination specified by the column given in
    /// the first argument. Other than that this behaves like an `id(x)`
    /// function so you can specify more links in the chain and even more
    /// flushers.
    fn flush(self, target: FlushTarget) -> error::Result<Flush<Self>>
    where
        Self: Sized,
    {
        Ok(Flush::new(self, target)?)
    }

    /// Mostly for debugging, calls a closure on each element. Behaves like the
    /// identity function on the stream returning each row untouched.
    fn inspect<F>(self, f: F) -> Inspect<Self, F>
    where
        Self: Sized,
        F: FnMut(&Headers, &RowResult),
    {
        Inspect::new(self, f)
    }

    /// Renames some columns
    fn rename(self, name_map: &HashMap<&str, &str>) -> Rename<Self>
    where
        Self: Sized,
    {
        Rename::new(self, name_map)
    }

    /// Apply a function to an entire row
    fn map_row<F>(self, f: F) -> MapRow<Self, F>
    where
        Self: Sized,
        F: Fn(&Headers, &Row) -> RowResult,
    {
        MapRow::new(self, f)
    }

    /// Apply a function to a single column of the stream, this function dones't fail
    /// if the column dones't exist.
    fn map_col<F>(self, col: String, f: F) -> MapCol<Self, F>
    where
        Self: Sized,
        F: Fn(&str) -> Result<String, col::BuildError>,
    {
        MapCol::new(self, col, f)
    }
}