Observables

Observable.ts

type Callback<A> = (value: A) => void;

class Observable<A> {
    private subscribers: Callback<A>[] = [];

    next(value: A): void {
        this.subscribers.forEach(fn => fn(value));
    }

    map<B>(mapper: (_: A) => B): Observable<B> {
        const result = new Observable<B>();
        this.subscribe(value => result.next(mapper(value)));
        return result;
    }

    merge<B>(b: Observable<B>): Observable<[A, B]> {
        return this.flatMap(x => b.map(y => [x, y]));
    }

    flatMap<B>(mapper: (_: A) => Observable<B>): Observable<B> {
        const ob = new Observable<B>();
        this.subscribe(x => {
            mapper(x).subscribe(y => ob.next(y));
        });
        return ob;
    }

    subscribe(callback: Callback<A>): void {
        this.subscribers.push(callback);
    }

    tap(callback: Callback<A>): Observable<A> {
        this.subscribers.push(callback);
        return this;
    }
}


// Usage:

const observable = new Observable<number>();

let a = observable.map(x => x + 1);
let b = observable.map(y => y + 1);
let c = a.merge(b);
// let d = c.map(xs => xs[0] + xs[1]);
let d = c;

d.subscribe(v => console.log(v));


// Let them run!

observable.next(20);
observable.next(30);