gistfile1.rs

use std::num::{One};

trait Observer<A> {
    fn next(&mut self, val: A);
}

trait Observable<A> {
    fn subscribe<'r, N: Observer<A>>(&mut self, observer: N);

    #[inline]
    fn map<'r, B>(self, f: |A|: 'r -> B) -> MapObservable<'r, A, B, Self> {
        MapObservable {f: Some(f), source: self}
    }

    #[inline]
    fn flat_map<'r, B, C: Observable<B>>(self, f: |A|: 'r -> C) -> FlatMapObservable<'r, A, B, C, Self> {
        FlatMapObservable {f: Some(f), source: self}
    }

    #[inline]
    fn filter<'r>(self, f: |&A|: 'r -> bool) -> FilterObservable<'r, A, Self> {
        FilterObservable {f: Some(f), source: self}
    }
}

struct RangeObservable<'r, A> {
    state: A,
    stop: A,
    one: A
}

impl<'r, A: Add<A, A> + PartialOrd + Clone + ToPrimitive> Observable<A> for RangeObservable<'r, A> {
    #[inline]
    fn subscribe<'r, N: Observer<A>>(&mut self, mut observer: N) {
        loop {
            observer.next(self.state.clone());
            self.state = self.state + self.one;
            if self.state >= self.stop {
                break;
            }
        }
    }
}

#[inline]
fn range<'r, A: Add<A, A> + PartialOrd + Clone + One>(start: A, stop: A) -> RangeObservable<'r, A> {
    RangeObservable {state: start, stop: stop, one: One::one()}
}

struct MapObservable<'r, A, B, S> {
    f: Option<|A|: 'r -> B>,
    source: S
}

impl<'r, A, B, S: Observable<A>> Observable<B> for MapObservable<'r, A, B, S> {
    #[inline]
    fn subscribe<'r, N: Observer<B>>(&mut self, observer: N) {
        self.source.subscribe(MapObserver {f: self.f.take().unwrap(), observer: observer});
    }
}

struct MapObserver<'r, A, B, N> {
    f: |A|: 'r -> B,
    observer: N
}

impl<'r, A, B, N: Observer<B>> Observer<A> for MapObserver<'r, A, B, N> {
    #[inline]
    fn next(&mut self, val: A) {
        self.observer.next((self.f)(val));
    }
}

struct FlatMapObservable<'r, A, B, C, S> {
    f: Option<|A|: 'r -> C>,
    source: S
}

impl<'r, A, B, C: Observable<B>, S: Observable<A>> Observable<B> for FlatMapObservable<'r, A, B, C, S> {
    #[inline]
    fn subscribe<'r, N: Observer<B>>(&mut self, observer: N) {
        self.source.subscribe(FlatMapObserver {f: self.f.take().unwrap(), observer: observer});
    }
}

struct FlatMapObserver<'r, A, B, C, N> {
    f: |A|: 'r -> C,
    observer: N
}

impl<'r, A, B, C: Observable<B>, N: Observer<B>> Observer<A> for FlatMapObserver<'r, A, B, C, N> {
    #[inline]
    fn next(&mut self, val: A) {
        let mut observable = (self.f)(val);

        observable.subscribe(AnonymousObserver {
            next: |a| self.observer.next(a)
        })
    }
}

struct FilterObservable<'r, A, S> {
    f: Option<|&A|: 'r -> bool>,
    source: S
}

impl<'r, A, S: Observable<A>> Observable<A> for FilterObservable<'r, A, S> {
    #[inline]
    fn subscribe<'r, N: Observer<A>>(&mut self, observer: N) {
        self.source.subscribe(FilterObserver {f: self.f.take().unwrap(), observer: observer});
    }
}

struct FilterObserver<'r, A, N> {
    f: |&A|: 'r -> bool,
    observer: N
}

impl<'r, A, N: Observer<A>> Observer<A> for FilterObserver<'r, A, N> {
    #[inline]
    fn next(&mut self, val: A) {
        if (self.f)(&val) {
            self.observer.next(val);
        }
    }
}

struct AnonymousObserver<'r, A> {
    next: |A|: 'r
}

impl<'r, A> Observer<A> for AnonymousObserver<'r, A> {
    #[inline]
    fn next(&mut self, val: A) {
        (self.next)(val);
    }
}

fn main() {
    range(0, 5u).
        map(|a| a + 5).
        filter(|a| a % 2 != 0).
        flat_map(|a| range(a, a + 5)).
        subscribe(AnonymousObserver {
            next: |a| println!("Got {}", a)
        });
}