RxJS - Observer and Subscription

Full Stack Developer and JavaScript and TypeScript enthusiastic.
Search for a command to run...

Full Stack Developer and JavaScript and TypeScript enthusiastic.
No comments yet. Be the first to comment.
Today, I wanna show you how to set up your devices with Antigravity using only the agentic mode, without touching any code. At the last CodeMotion Rome, I got my special badge. A CM Node, a special ha

As a developer, you have to take control of your projects every day. Whether it is a company repository, an open source project you maintain or collaborate on, or a simple pet project. Gaining control over your projects often depends on the platform ...

My Antigravity first experience

Write clear and insightful commit messages using AI and GitLens.

We are in the AI era. New models emerge daily, and many applications have already integrated AI into their workflows. Gemini, OpenAI, Copilot, Deepseek, Llama, and many others enable AI in your applications or help you write code, but they all need a...

Welcome back guys, today the topics of the article are Observers and Subscriptions.
An observer is a consumer of values. An observer is something that is interested to the emitted values by the observable. In RxJS an Observer is simply a set of callbacks (next, error, complete). Here's a simple example of an observer
import { Observer } from "rxjs";
const observer: Observer<string> = {
next: (value: string) =>
console.log(`[observer] next`, value),
error: (error: Error) =>
console.error(`[observer] error`, error),
complete: () =>
console.log(`[observer] complete!`),
};
As you can see an observer has three callbacks, one for each type of notification that an Observable may emit. Every callback can react in the base of the observable event notification. An observer could have not all the three callbacks defined because maybe it doesn't want to react to all the notifications. In these cases, you can create an observer with the only callbacks you need.
import {
NextObserver,
ErrorObserver,
CompletionObserver,
} from "rxjs";
const nextObserver: NextObserver<string> = {
next: (value: string) =>
console.log(`[nextObserver] next`, value),
};
const errorObserver: ErrorObserver<string> = {
error: (error: Error) =>
console.error(`[errorObserver] error`, error),
};
const completeObserver: CompletionObserver<string> = {
complete: () =>
console.log(`[completeObserver] complete!`),
};
It's possible to define a NextObserver without the next property but indicating only the body of the next method, in this case, the observer by default is of the type NextObserver.
const defaultNextObserver: (value: string) => void = (value: string) =>
console.log(`${new Date().toISOString()} - [defaultNextObserver] next`, value)
A Subscription is an actor that decides when an Observable must be listened and when we can stop listening to it. Until an observable is not subscribed in your code nothing happens, but as soon as you create a subscription, the magic starts. A Subscription in RxJS is an object created using the method "subscribe" and it has one main method: "unsubscribe"; this method allows you to stop listening the observable event. In your code is important to call the "unsubscribe" when you no longer need the subscription, this prevent problems as the memory leaks. An example of a Subscription
import { Subscription } from "rxjs";
const observer = (value: string) => console.log(`[unsubscribe method] next`, value)
const subscription: Subscription = observable.subscribe(observer);
subscription.unsubscribe();
Another pretty feature in RxJS Subscription is the "add" method; this method allows you to collect more subscriptions inside of one Subscription instance and after that, you can unsubscribe all the subscriptions at once time.
import { Subscription } from "rxjs";
const subscription: Subscription = observable.subscribe((value: string) =>
console.log(
`[unsubscribe 1 method] next`,
value
)
);
subscription.add(
observable.subscribe((value: string) =>
console.log(
`[unsubscribe 2 method] next`,
value
)
)
);
subscription.unsubscribe();
In some cases when you call the unsubscribe method you need to run some special code in your observable: RxJS let us do this using a special syntax inside of the observable declaration. When you create the observable you can return a function that the library invokes in the future during the unsubscription. Below a simple example to understand better the problem and the solution:
import {
Observable,
Subscriber,
Subscription,
} from "rxjs";
const observableWithCallback = new Observable<string>(
(subscriber: Subscriber<string>) => {
let count = 0;
const id = setInterval(() => {
subscriber.next(`Count: ${++count}`);
}, 300);
return () => {
console.log("On UnSubscription");
clearInterval(id);
};
}
);
const subscriptionObservableWithCallback: Subscription = observableWithCallback.subscribe({
next: (value: string) =>
console.log(`[observableWithCallback] Next: ${value}`),
});
setTimeout(() => {
subscriptionObservableWithCallback.unsubscribe();
}, 3000);
[observableWithCallback] Next: Count: 1
[observableWithCallback] Next: Count: 2
[observableWithCallback] Next: Count: 3
[observableWithCallback] Next: Count: 4
[observableWithCallback] Next: Count: 5
[observableWithCallback] Next: Count: 6
[observableWithCallback] Next: Count: 7
[observableWithCallback] Next: Count: 8
[observableWithCallback] Next: Count: 9
On UnSubscription
Ok, guys, that's all for today. In the next article we'll see the last fundamental of RxJS: the Operator and later, we can move to see the features of this Library.
You can find the examples of this article here.
See you soon guys!