Code shared from the Rust Playground

playground.rs

trait Observer<T> {
    fn on_next(&mut self, item: Option<T>);
}

impl<T, F> Observer<T> for F where F: FnMut(Option<T>) {
    fn on_next(&mut self, item: Option<T>) {
        self(item)
    }
}

trait Observable: Sized {
    type Item;

    fn subscribe<F>(self, f: F) where F: Observer<Self::Item>;

    fn run(self) {
        self.subscribe(|_| {})
    }
}

impl<B> Observable for B where B: Iterator {
    type Item = B::Item;

    fn subscribe<F>(mut self, mut f: F) where F: Observer<Self::Item> {
        while let Some(item) = self.next() {
            f.on_next(Some(item))
        }
        f.on_next(None)
    }
}

struct Map<O, M> {
    obs: O,
    f: M,
}

impl<O: Observable, M, B> Observable for Map<O, M> where M: FnMut(O::Item) -> B {
    type Item = B;

    fn subscribe<F>(self, mut f: F) where F: Observer<Self::Item> {
        let mut func = self.f;
        self.obs.subscribe(|x: Option<_>| f.on_next(x.map(&mut func)))
    }
}

trait ObservableExt: Observable where Self: Observable + Sized {
    fn rx_map<M, B>(self, f: M) -> Map<Self, M> where M: FnMut(Self::Item) -> B {
        return Map { obs: self, f }
    }
}

impl<O> ObservableExt for O where O: Observable {}

struct Subject<'a, O: Observable> {
    obs: O,
    observers: Vec<Box<Observer<O::Item> + 'a>>,
}

impl<'a, O: Observable> Subject<'a, O> where O::Item: Clone {
    fn new(obs: O) -> Self {
        Subject { obs, observers: Vec::new() }
    }

    pub fn add_subscriber<F>(mut self, f: F) -> Self where F: Observer<O::Item> + 'a {
        self.observers.push(Box::new(f));
        self
    }
}

impl<'a, O: Observable> Observable for Subject<'a, O> where O::Item: Clone {
    type Item = O::Item;

    fn subscribe<F>(self, mut f: F) where F: Observer<Self::Item> {
        let mut observers = self.observers;
        self.obs.subscribe(|x: Option<_>| {
            for observer in &mut observers {
                observer.on_next(x.clone()); // can also use a ref
            }
            f.on_next(x);
        })
    }
}

fn main() {
    Subject::new(
        (0..5)
        .map(|x| x * 2)
        .rx_map(|x| x + 17)
    )
        .add_subscriber(|x| { println!("1: {:?}", x); })
        .add_subscriber(|x| { println!("2: {:?}", x); })
        .run();
}