Skip to content

Instantly share code, notes, and snippets.

Embed
What would you like to do?
Observables
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);
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment