Basic and silly implementation of conduit in Rust

conduit.rs

#![feature(never_type)]

pub enum Conduit<I, O, R> {
    Send(O, Box<dyn FnOnce() -> Conduit<I, O, R>>),
    Recv(Box<dyn FnOnce(Option<I>) -> Conduit<I, O, R>>),
    Done(R),
    Leftover(I, Box<dyn FnOnce() -> Conduit<I, O, R>>),
}

use Conduit::*;

impl <I, O> Conduit<I, O, ()> {
    pub fn send(o: O) -> Self {
        Send(o, Box::new(|| Done(())))
    }
}

impl<I, O> Conduit<I, O, Option<I>> {
    pub fn recv() -> Self {
        Recv(Box::new(|oi| Done(oi)))
    }
}

impl<I: 'static, O: 'static, R1: 'static> Conduit<I, O, R1> {
    pub fn and_then<R2: 'static>(self, next: impl FnOnce(R1) -> Conduit<I, O, R2> + 'static) -> Conduit<I, O, R2> {
        match self {
            Send(_, _) => todo!(),
            Recv(f) => Recv(Box::new(move |oi| {
                f(oi).and_then(next)
            })),
            Done(r1) => next(r1),
            Leftover(_, _) => todo!(),
        }
    }
}

impl<R> Conduit<(), (), R> {
    pub fn run(self) -> R {
        use Conduit::*;
        let mut conduit = self;
        loop {
            match conduit {
                Recv(f) => {
                    conduit = f(None);
                }
                Done(r) => {
                    return r;
                }
                Send(_, _) => unreachable!(),
                Leftover(_, _) => unreachable!(),
            }
        }
    }
}

impl<O: Clone + 'static> Conduit<(), O, ()> {
    pub fn repeat(o: O) -> Self {
        Conduit::Send(o.clone(), Box::new(move || Conduit::repeat(o)))
    }
}

impl<A> Conduit<A, A, ()> {
    pub fn take(count: usize) -> Self {
        if count == 0 {
            Conduit::Done(())
        } else {
            Conduit::Recv(Box::new(move |oa| {
                match oa {
                    None => Conduit::Done(()),
                    Some(a) => Conduit::Send(a, Box::new(move || Conduit::take(count - 1))),
                }
            }))
        }
    }
}

impl<A: 'static> Conduit<A, (), Vec<A>> {
    pub fn sink_vec() -> Self {
        Conduit::sink_vec_helper(Vec::new())
    }

    fn sink_vec_helper(mut v: Vec<A>) -> Self {
        Conduit::Recv(Box::new(move |oa| {
            match oa {
                None => Conduit::Done(v),
                Some(a) => {
                    v.push(a);
                    Conduit::sink_vec_helper(v)
                }
            }
        }))
    }
}

impl<I: 'static, O: 'static> Conduit<I, O, ()> {
    pub fn map<F: FnMut(I) -> O + 'static>(mut f: F) -> Self {
        Conduit::Recv(Box::new(move |oa| {
            match oa {
                None => Done(()),
                Some(i) => {
                    let o = f(i);
                    Send(o, Box::new(move || Conduit::map(f)))
                }
            }
        }))
    }
}

impl<A: 'static, B: 'static> Conduit<A, B, ()> {
    pub fn fuse<C: 'static, R: 'static>(self, c: Conduit<B, C, R>) -> Conduit<A, C, R> {
        match c {
            Done(r) => Done(r),
            Send(b, right) => {
                let rest = move || self.fuse(right());
                Send(b, Box::new(rest))
            }
            Recv(f) => match self {
                Send(a, left) => {
                    left().fuse(f(Some(a)))
                }
                Recv(_) => todo!(),
                Done(()) => Done(()).fuse(f(None)),
                Leftover(_, _) => todo!(),
            }
            Leftover(_, _) => todo!(),
        }
    }
}

#[cfg(test)]
mod tests {
    use super::Conduit;

    #[test]
    fn simple_vec() {
        let expected = vec![42, 42, 42];
        let actual = Conduit::repeat(42).fuse(Conduit::take(3)).fuse(Conduit::sink_vec()).run();
        assert_eq!(expected, actual);
    }

    #[test]
    fn with_map() {
        let expected = vec![44, 44, 44];
        let actual = Conduit::repeat(42).fuse(Conduit::take(3)).fuse(Conduit::map(|x| x + 2)).fuse(Conduit::sink_vec()).run();
        assert_eq!(expected, actual);
    }

    #[test]
    fn with_two_maps() {
        let expected = vec![47, 47, 47];
        let actual = Conduit::repeat(42)
            .fuse(Conduit::take(3))
            .fuse(Conduit::map(|x| x + 2))
            .fuse(Conduit::map(|x| x + 3))
            .fuse(Conduit::sink_vec()).run();
        assert_eq!(expected, actual);
    }

    #[test]
    fn use_and_then() {
        let expected = vec![Some("Alice")];
        let actual = Conduit::repeat("Alice")
            .fuse(Conduit::recv().and_then(|x| Conduit::send(x)))
            .fuse(Conduit::sink_vec())
            .run();
        assert_eq!(expected, actual);
    }
}