Angular Observables
This is not an Angular library, but a library called RxJS that is used in Angular to handle asynchronous operations.
What is an Observable?
Produces or emits stream of data.
Naming conventions
For observables properties, we use the $ sign at the end of the variable name.
BehaviorSubject
BehaviorSuibject is a way to emit data and to subscribe to it to get the last emitted value. It's one way to create and observable.
import { BehaviorSubject } from 'rxjs';
import { Injectable } from '@angular/core';
@Injectable({
providedIn: 'root'
})
export class DataService {
messages = new BehaviorSubject<string[]>([]);
private message = string[] = [];
get allMessages() {
return [...this.messages];
}
addMessage(message: string) {
this.messages= [...this.messages, message];
}
}Interval function
Create an observable that emits a sequence of numbers in a specified interval of time.
import { interval } from 'rxjs';
@Component({
selector: 'app-root',
templateUrl: './app.component.html',
styleUrls: ['./app.component.css']
})
export class AppComponent implements OnInit {
ngOnInit() {
const interval$ = interval(1000);
interval$.subscribe({
// here we are in an observer
next: (value) => console.log(value), // will be triggered when the observable emits a value
complete: () => console.log('Complete') // will be triggered when the observable as no more values to emit
error: (error) => console.log(error) // will be triggered when an error occurs
});
}
}If you only care to next function you can also pass a function to the subscribe method.
interval$.subscribe((value) => console.log(value));Unsuscribe
When you use Oservable you really need to care about unsubscribing to avoid memory leaks. You an do it like this
import { inject, DestroyRef } from "@angular/core";
import { interval } from "rxjs";
@Component({
selector: "app-root",
templateUrl: "./app.component.html",
styleUrls: ["./app.component.css"],
})
export class AppComponent implements OnInit {
private destroyRef = inject(DestroyRef);
ngOnInit() {
const interval$ = interval(1000);
interval$.subscribe((value) => console.log(value));
this.destroyRef.subscribe(() => {
interval$.unsubscribe();
});
}
}Operators
Operators are functions that are used to manipulate the data emitted by the observable. Keep in mind you always need at least one subscription per observable to trigger the observable.
import { inject, DestroyRef } from "@angular/core";
import { interval } from "rxjs";
@Component({
selector: "app-root",
templateUrl: "./app.component.html",
styleUrls: ["./app.component.css"],
})
export class AppComponent implements OnInit {
private destroyRef = inject(DestroyRef);
ngOnInit() {
const interval$ = interval(1000);
interval$
.pipe(
// rxjs operator here, to put it in our pipeline
map(
// operator used to convert the value of the observable
(value) => value * 2
).subscribe((value) => {
next: (
value // value here is the new of the map operator
) => console.log(value);
})
)
.subscribe((value) => console.log(value));
this.destroyRef.subscribe(() => {
interval$.unsubscribe();
});
}
}Subjects
Take care to also emitting the values manually You just don't need to subscribe to the observable to emit values So, subjects vs observables : observables emit values automatically, subjects emit values manually.
Subjects vs Signals
To basic emitting values, you can use signals. Signals are very new in Angular, in the past when you wanted to emit values you had to use subjects/observables.
Observables are a pipeline of values over times Signals are a values container.
Observables are great for managing events & streamed data over time Signals are great for managing application state
Observables can have NO initial value That's why when we convert an observable to a signal, we need to provide an initial value. Subjects, can have an initial value. Signals have always an initial value. So when we convert an observable to a signal, Angular as a workaround, set the initial value to undefined when we use the toSignal function.
toObservable and toSignal
export class AppComponent implements OnInit {
private destroyRef = inject(DestroyRef);
clickCount = signal(0);
clicCount$ = toObservable(this.clickCount);
interval$ = interval(1000);
intervalSignal = toSignal(this.interval$, {initialValue: 0});
/*
As mentioned before, the initial value of the signal is undefined
To avoid this we cas use the initial value to override the default undefined value.
*/
/**
* now we can use intervalSignal as a signal iwht computed, effect, etc
*/
/***
* When you use toSignal on an observables, angular will destroy when you components gets removed
* you can disable this behavior with the manualCleanup option but you doh't have tto do that.
*/
constructor () {
toObservable(this.clicCount)
const subscription = this.clickCount$.subscribe({
next: (value) => console.log(value)
});
this .destroyRef.onDestry(() => {
subscription.unsubscribe();
});
}
ngOnInit() {
this.clickCount$.subscribe({
next: (value) => console.log(value)
});
});
}
onClick() {
this.clickCount.update(value => value + 1);
}
}Create a custom observable
import { Observable } from "rxjs";
export class AppComponent implements OnInit {
customInterval$ = new Observable(
(
subscriber // we got the subscriber object here when we called the subscribe method, that's here we can call next, complete, error etc.
) => {
let count = 0;
const intervalID = setInterval(() => {
if (count > 10) {
clearInterval(intervalID);
subscriber.complete();
return;
}
// you can also defined the error method
// subscriber.error(new Error('An error occured'));
subscriber.next({
message: "Hello",
count: count,
});
count++;
}, 1000);
}
);
ngOnInit() {
this.customInterval$.subscribe({
next: (value) => console.log(value),
complete: () => console.log("Complete"),
error: (error) => console.log(error),
});
}
}