Write a minimal Observable class with subscribe, unsubscribe, teardown, and a map method.
Before you start
What's an Observable?
An Observable is a source of values that arrive over time, such as clicks, messages from a server, or the ticks of a timer, that you can subscribe to. It's the central idea of RxJS, the Reactive Extensions library for JavaScript. Two sides meet when you subscribe. The producer is the function you give the Observable when you create it: it makes the values and pushes them out. The observer is what you give subscribe: an object with up to three callbacks, next for each value, error if something goes wrong, and complete when there will be no more values. A single function also works, as next.
A plain Observable is lazy: creating one runs nothing. The producer runs when someone subscribes, and runs again, separately, for each subscriber, so two subscribers to a timer get two timers. It may call next any number of times, then at most one error or complete, and after that nothing more reaches the observer. You can also stop early by calling unsubscribe() on the subscription that subscribe returns. However it ends, the teardown runs: a cleanup function the producer returns, which stops a timer, removes an event listener, or closes a connection. Here's a timer made with RxJS, whose Observable you import from rxjs:
const ticks = new Observable((subscriber) => {
console.log("started");
let n = 0;
const id = setInterval(() => subscriber.next(n++), 1000);
return () => {
clearInterval(id);
console.log("cleaned up");
};
});
const subscription = ticks.subscribe({
next: (n) => console.log(n),
complete: () => console.log("done"),
});
setTimeout(() => subscription.unsubscribe(), 3500);
| When | What happens | Logged |
|---|---|---|
ticks is created | Nothing runs: nobody has subscribed. | Nothing |
subscribe, at 0 ms | The producer runs and starts a timer. | started |
| 1000 ms | The timer calls subscriber.next(0), which reaches the observer's next. | 0 |
| 2000 ms | next(1) | 1 |
| 3000 ms | next(2) | 2 |
| 3500 ms | unsubscribe() runs the teardown, which stops the timer. complete isn't called, because the Observable didn't finish: you stopped listening. | cleaned up |
Operators make new Observables out of existing ones, the way map and filter on an array make new arrays. In RxJS you pass them to pipe, importing filter and map from rxjs too: ticks.pipe(filter((n) => n % 2 === 0), map((n) => n * 10)) is an Observable of 0, 20, 40, and so on. The source is left as it was, nothing runs until someone subscribes to the new Observable, and unsubscribing from it unsubscribes from the source too, so the timer stops.
A promise is like a parcel you've ordered: one delivery, already on its way. An Observable is more like a newsletter: nothing arrives until you sign up, and issues keep coming until it ends or you unsubscribe.
| Question | Promise | Observable |
|---|---|---|
| How many values? | One value, or one error. | Any number of values, then possibly one error or one completion. |
| When does the work start? | When the promise is created, whether or not anyone calls then. | When you subscribe, and again for each subscriber. |
| Can you cancel it? | No: a promise has no cancel method. | Yes: unsubscribe() stops it and runs the teardown. |
You meet Observables most in Angular, whose HttpClient returns one for each request and only sends the request when something subscribes, and wherever RxJS combines streams of events, such as keystrokes in a search box. RxJS has dozens of operators, such as debounceTime and switchMap, and they all follow this same contract.
new Observable(producer) stores producer, a function that takes a subscriber and may return a teardown function: new Observable((subscriber) => { …; return teardown; }). The producer calls subscriber.next(value), subscriber.error(err), and subscriber.complete().
Nothing runs until subscribe(observer). observer is an object with optional next, error, and complete methods, a next function on its own, or nothing at all. Each subscribe runs the producer again and returns { unsubscribe }.
After error, complete, or unsubscribe, no further notifications get through, and the teardown runs exactly once, even when the producer completes before it has returned the teardown. An error thrown by the producer goes to error.
map(fn) returns a new Observable whose values are the source's values passed through fn.
class Observable {
constructor(producer) {
// your code
}
subscribe(observer) {}
map(fn) {}
}
// const numbers = new Observable((subscriber) => {
// subscriber.next(1);
// subscriber.next(2);
// subscriber.complete();
// subscriber.next(3); // never gets through: the observable has completed
// });
// numbers.map((n) => n * 10).subscribe({
// next: (value) => console.log(value),
// complete: () => console.log("done"),
// });
// // logs 10, 20, then "done"
//
// const ticks = new Observable((subscriber) => {
// let n = 0;
// const id = setInterval(() => subscriber.next(n++), 1000);
// return () => clearInterval(id);
// });
// const subscription = ticks.subscribe((value) => console.log(value)); // logs 0, 1, 2, … once a second
// subscription.unsubscribe(); // runs the teardown, which stops the interval
Solve it in the editor, with tests
The answer
Write it yourself first. Then compare it with a reference solution and the checks it passes.
Checking your sign-in…