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.
Write a class Observable, the core of RxJS. new Observable(producer) stores producer, a function that receives a subscriber object with next(value), error(err), and complete() methods. The producer may return a teardown function that cleans up.
Nothing runs until someone subscribes, and every subscribe runs the producer again.
subscribe(observer) takes an object with optional next, error, and complete methods, or a single function, used as next, or nothing at all. It returns a subscription, { unsubscribe }.
The subscriber passes notifications on to the observer until the subscription closes, which happens on error, complete, or unsubscribe(). After that nothing more gets through. If the producer throws, the error goes to the observer's error.
The teardown runs exactly once, when the subscription closes. If the producer completes before it has returned the teardown, the teardown still runs, by the time subscribe returns. Calling unsubscribe() a second time doesn't run it again.
map(fn) returns a new Observable with each value replaced by fn(value). filter(predicate) returns a new Observable with only the values for which predicate(value) is truthy.
Both pass errors and completion through, and an error thrown by fn or predicate goes to error. Unsubscribing from the Observable they return unsubscribes from the source.
const ticks = new Observable((subscriber) => {
let n = 0;
const id = setInterval(() => subscriber.next(n++), 1000);
return () => clearInterval(id);
});
const subscription = ticks
.filter((n) => n % 2 === 0)
.map((n) => n * 10)
.subscribe((value) => console.log(value)); // 0, 20, 40, …
subscription.unsubscribe(); // clears the interval
const producer = spy();
const source = new Observable(producer);
expect(producer).not.toHaveBeenCalled();
source.subscribe(() => {});
source.subscribe(() => {});
expect(producer).toHaveBeenCalledTimes(2);
const source = new Observable((subscriber) => {
subscriber.next(1);
subscriber.next(2);
subscriber.complete();
});
const log = [];
source.subscribe({ next: (value) => log.push(value), complete: () => log.push("done") });
expect(log).toEqual([1, 2, "done"]);
const source = new Observable((subscriber) => {
subscriber.next("a");
subscriber.complete();
});
const next = spy();
source.subscribe(next);
expect(next).toHaveBeenCalledWith("a");
expect(() => source.subscribe()).not.toThrow();
expect(() => source.subscribe({})).not.toThrow();
const subscription = source.subscribe({ complete() {} });
expect(typeof subscription.unsubscribe).toBe("function");
const source = new Observable((subscriber) => {
subscriber.next(1);
subscriber.complete();
subscriber.next(2);
subscriber.error(new Error("late"));
subscriber.complete();
});
const next = spy();
const error = spy();
const complete = spy();
source.subscribe({ next, error, complete });
expect(next).toHaveBeenCalledTimes(1);
expect(error).not.toHaveBeenCalled();
expect(complete).toHaveBeenCalledTimes(1);
const clock = useFakeTimers();
const teardown = spy();
const ticks = new Observable((subscriber) => {
let n = 0;
const id = setInterval(() => subscriber.next(n++), 100);
return () => {
clearInterval(id);
teardown();
};
});
const next = spy();
const subscription = ticks.subscribe(next);
clock.tick(250);
subscription.unsubscribe();
subscription.unsubscribe();
clock.tick(500);
expect(next.calls.map((call) => call.args[0])).toEqual([0, 1]);
expect(teardown).toHaveBeenCalledTimes(1);
expect(clock.pending()).toBe(0);
const teardown = spy();
const source = new Observable((subscriber) => {
subscriber.next(1);
subscriber.complete();
return teardown;
});
const subscription = source.subscribe(() => {});
expect(teardown).toHaveBeenCalledTimes(1);
subscription.unsubscribe();
expect(teardown).toHaveBeenCalledTimes(1);
const clock = useFakeTimers();
const error = spy();
new Observable(() => {
throw new TypeError("producer");
}).subscribe({ error });
const ticks = new Observable((subscriber) => {
let n = 0;
const id = setInterval(() => subscriber.next(n++), 100);
return () => clearInterval(id);
});
const next = spy();
ticks
.map((n) => {
if (n === 1) throw new RangeError("fn");
return n;
})
.subscribe({ next, error });
clock.tick(300);
expect(next.calls.map((call) => call.args[0])).toEqual([0]);
expect(error).toHaveBeenCalledTimes(2);
expect(error.calls[0].args[0]).toBeInstanceOf(TypeError);
expect(error.calls[1].args[0]).toBeInstanceOf(RangeError);
expect(clock.pending()).toBe(0);
const source = new Observable((subscriber) => {
[1, 2, 3, 4, 5].forEach((n) => subscriber.next(n));
subscriber.complete();
});
const result = source.filter((n) => n % 2 === 1).map((n) => n * 10);
expect(result).toBeInstanceOf(Observable);
expect(result).not.toBe(source);
const log = [];
result.subscribe({ next: (value) => log.push(value), complete: () => log.push("done") });
expect(log).toEqual([10, 30, 50, "done"]);
const clock = useFakeTimers();
const teardown = spy();
const ticks = new Observable((subscriber) => {
let n = 0;
const id = setInterval(() => subscriber.next(n++), 100);
return () => {
clearInterval(id);
teardown();
};
});
const next = spy();
const subscription = ticks.map((n) => n * 2).filter((n) => n > 0).subscribe(next);
clock.tick(300);
subscription.unsubscribe();
clock.tick(300);
expect(next.calls.map((call) => call.args[0])).toEqual([2, 4]);
expect(teardown).toHaveBeenCalledTimes(1);
expect(clock.pending()).toBe(0);
Run tests runs every test on your code, here in your browser. The problem is solved when they all pass.
Ctrl + Enter runs the tests (⌘ + Enter on a Mac).
I'm here to help with Observable with map and filter. I give hints, never the solution, so you get to solve it yourself. Ask for a hint, select code to ask about it, or press Ask Tuto why on a failing test.
Your first 3 answers need no account. for free to keep asking.