transducers.rs

use std::collections::HashSet;
use std::hash::Hash;

pub enum StepResult {
    Next,
    Stop,
}

pub trait Reducing<Item, Acc> {
    type Item;

    fn initial(&mut self) -> Acc;

    fn step(&mut self, acc: &mut Acc, item: Item) -> StepResult;

    fn complete(self, acc: Acc) -> Acc;
}

pub trait Transducer<ReducerIn> {
    type ReducerOut;
    fn new(self, reducing_fn: ReducerIn) -> Self::ReducerOut;
}

pub struct ComposedTransducer<AT, BT> {
    a: AT,
    b: BT,
}

impl<ReducerIn, RT, RO, AT, BT> Transducer<ReducerIn> for ComposedTransducer<AT, BT>
where
    AT: Transducer<ReducerIn, ReducerOut = RT>,
    BT: Transducer<RT, ReducerOut = RO>,
{
    type ReducerOut = RO;

    fn new(self, reducing_fn: ReducerIn) -> Self::ReducerOut {
        self.b.new(self.a.new(reducing_fn))
    }
}

pub fn compose<AT, BT>(a: AT, b: BT) -> ComposedTransducer<AT, BT> {
    ComposedTransducer { a, b }
}

pub struct MapTransducer<F> {
    f: F,
}

pub struct MapReducer<R, F> {
    reducer: R,
    transducer: MapTransducer<F>,
}

impl<F, ReducerIn> Transducer<ReducerIn> for MapTransducer<F> {
    type ReducerOut = MapReducer<ReducerIn, F>;

    fn new(self, reducing_fn: ReducerIn) -> Self::ReducerOut {
        MapReducer {
            reducer: reducing_fn,
            transducer: self,
        }
    }
}

impl<R, F, Item, Out, Acc> Reducing<Item, Acc> for MapReducer<R, F>
where
    F: Fn(Item) -> Out,
    R: Reducing<Out, Acc>,
{
    type Item = Out;

    fn initial(&mut self) -> Acc {
        self.reducer.initial()
    }

    fn step(&mut self, acc: &mut Acc, item: Item) -> StepResult {
        self.reducer.step(acc, (self.transducer.f)(item))
    }

    fn complete(self, acc: Acc) -> Acc {
        self.reducer.complete(acc)
    }
}

pub fn map<F, In, Out>(f: F) -> MapTransducer<F>
where
    F: Fn(In) -> Out,
{
    MapTransducer { f }
}

pub struct FilterTransducer<F> {
    f: F,
    inclusive: bool,
}

pub struct FilterReducer<R, F> {
    rf: R,
    t: FilterTransducer<F>,
}

impl<F, ReducerIn> Transducer<ReducerIn> for FilterTransducer<F> {
    type ReducerOut = FilterReducer<ReducerIn, F>;

    fn new(self, reducing_fn: ReducerIn) -> Self::ReducerOut {
        FilterReducer {
            rf: reducing_fn,
            t: self,
        }
    }
}

impl<R, F, Item, Acc> Reducing<Item, Acc> for FilterReducer<R, F>
where
    F: Fn(&Item) -> bool,
    R: Reducing<Item, Acc>,
{
    type Item = Item;

    fn initial(&mut self) -> Acc {
        self.rf.initial()
    }

    #[inline]
    fn step(&mut self, acc: &mut Acc, item: Self::Item) -> StepResult {
        let mut include = (self.t.f)(&item);
        if !self.t.inclusive {
            include = !include;
        }
        if include {
            self.rf.step(acc, item)
        } else {
            StepResult::Next
        }
    }

    fn complete(self, acc: Acc) -> Acc {
        acc
    }
}

pub fn filter<F, T>(f: F) -> FilterTransducer<F>
where
    F: Fn(&T) -> bool,
{
    FilterTransducer { f, inclusive: true }
}

pub struct VecReducer {}

impl VecReducer {
    pub fn new() -> Self {
        VecReducer {}
    }
}

impl<Item> Reducing<Item, Vec<Item>> for VecReducer {
    type Item = Item;

    fn initial(&mut self) -> Vec<Self::Item> {
        Vec::new()
    }

    fn step(&mut self, acc: &mut Vec<Self::Item>, item: Self::Item) -> StepResult {
        acc.push(item);
        StepResult::Next
    }

    fn complete(self, acc: Vec<Self::Item>) -> Vec<Self::Item> {
        acc
    }
}

//Joins
#[derive(Debug, Copy, Clone, Eq, PartialEq)]
enum CrossState {
    Start,
    LeftRight,
    Left,
    RightCache,
}

pub struct JoinItem<I: Hash + Eq> {
    l: I,
    r: I,
}

pub struct JoinTransducer<AT, BT, I> {
    a: AT,
    b: BT,
    pos: usize,
    pos_a: usize,
    pos_b: usize,
    state: CrossState,
    left: HashSet<I>,
}

impl<ReducerIn, RT, RO, AT, BT, I> Transducer<ReducerIn> for JoinTransducer<AT, BT, I>
where
    AT: Transducer<ReducerIn, ReducerOut = RT>,
    BT: Transducer<RT, ReducerOut = RO>,
{
    type ReducerOut = RO;

    fn new(self, reducing_fn: ReducerIn) -> Self::ReducerOut {
        self.b.new(self.a.new(reducing_fn))
    }
}
// Incomplete: How????
// impl<R, F, Item, Acc> Reducing<JoinItem<Item>, Acc> for JoinTransducer<R, F, Item>
// where
//     F: Fn(JoinItem<Item>) -> Vec<Item>,
//     R: Reducing<JoinItem<Item>, Acc>,
//     Item: Eq + Hash + Clone,
// {
//     type Item = JoinItem<Item>;
//
//     fn initial(&mut self) -> Acc {
//         unimplemented!()
//     }
//
//     fn step(&mut self, acc: &mut Acc, item: Self::Item) -> StepResult {
//         dbg!(self.pos_a, self.pos_b, self.pos, self.state);
//         match self.state {
//             CrossState::Start => {
//                 if let StepResult::Stop = self.a.step(acc, item) {
//                     StepResult::Stop
//                 } else {
//                     self.left.insert(item.l.clone());
//                     StepResult::Next
//                 }
//             }
//             CrossState::LeftRight => {
//                 if let StepResult::Stop = self.b.step(acc, item) {
//                     dbg!("B ITER");
//                     StepResult::Stop
//                 } else {
//                     self.left.insert(item.l.clone());
//                     StepResult::Next
//                 }
//             }
//             CrossState::Left => {}
//             CrossState::RightCache => {}
//         };
//     }
//
//     fn complete(self, acc: Acc) -> Acc {
//         acc
//     }
// }

pub fn transduce<T, Collect, Reducer, Item, Acc>(
    transducer: T,
    reducer: Collect,
    iter: impl Iterator<Item = Item>,
) -> Acc
where
    T: Transducer<Collect, ReducerOut = Reducer>,
    Reducer: Reducing<Item, Acc>,
{
    let mut r = transducer.new(reducer);

    let mut accumulator = r.initial();
    for item in iter {
        if let StepResult::Stop = r.step(&mut accumulator, item) {
            break;
        }
    }
    r.complete(accumulator)
}

fn main() {
    let nums = 1..1000;
    let double = map(|x: i32| x * 2);
    let plus = map(|x: i32| x + 10);
    let select = filter(|x: &i32| x % 2 == 0);

    let fused = compose(double, plus);
    let fused = compose(fused, select);

    let r: VecReducer = VecReducer::new();

    let t = transduce(fused, r, nums.into_iter());
    dbg!(t);

    let a = 1..10;
    let b = 21..30;
    let tuple = map(|(l, r): (i32, i32)| vec![l, r]);
}