WebSocket avec Angular : Communication en temps réel avec RxJS

La communication en temps réel est devenue essentielle dans les applications web modernes. Imaginez une application de messagerie instantanée : quand quelqu'un vous envoie un message, vous voulez le recevoir immédiatement, sans avoir à rafraîchir la page. C'est exactement ce que permettent les WebSockets, et avec RxJS, nous pouvons les gérer de manière encore plus élégante !

Avant de commencer

Vous devez réaliser un serveur qui gère les WebSockets.

Dans cet exemple, nous allons utiliser socket.io.

Comprendre les technologies utilisées

WebSocket : Le protocole natif

WebSocket est un protocole de communication bidirectionnelle standardisé par le W3C qui permet d'établir un canal de communication permanent entre un navigateur et un serveur. Contrairement au HTTP traditionnel où chaque requête nécessite une nouvelle connexion, WebSocket maintient une seule connexion ouverte, ce qui le rend idéal pour :

• Les applications en temps réel (chat, jeux en ligne)
• Les notifications push
• Les mises à jour en direct (cours boursiers, scores sportifs)
• Les applications collaboratives

Exemple de WebSocket natif :

const ws = new WebSocket('ws://monserveur.com');
ws.onmessage = (event) => console.log(event.data);

Socket.IO : La bibliothèque évoluée

Socket.IO est une bibliothèque qui s'appuie sur WebSocket mais offre beaucoup plus de fonctionnalités :

• Reconnexion automatique en cas de perte de connexion
• Support des rooms et des namespaces pour organiser les communications
• Fallback automatique vers d'autres protocoles si WebSocket n'est pas disponible
• Gestion automatique des timeouts et des heartbeats
• Support natif de la sérialisation JSON
• Émission et réception d'événements personnalisés

C'est pourquoi nous l'utilisons dans ce tutoriel plutôt que les WebSockets natifs.

Implémentation avec Socket.IO et RxJS

Commençons par installer les dépendances nécessaires :

Installation

npm install socket.io-client
npm install @types/socket.io-client -D # signifie que c'est une dépendance de développement

1. Création du service WebSocket

Créons d'abord un service qui gérera notre connexion WebSocket avec RxJS. Nous allons voir pourquoi et comment utiliser RxJS pour une meilleure gestion des événements WebSocket.

Pourquoi utiliser RxJS avec les WebSockets ?

Les WebSockets émettent naturellement des événements de manière continue (connexion, messages, déconnexion...). RxJS est parfaitement adapté pour gérer ce type de flux de données car :

• Il permet de traiter les événements comme des flux de données continus
• Il offre des opérateurs puissants pour transformer et combiner ces flux
• Il facilite la gestion des erreurs et des cas limites
• Il simplifie la gestion de la mémoire avec des mécanismes de désabonnement

Conception du service

websocket.service.ts

import { Injectable, signal } from '@angular/core';
import { Socket, io } from 'socket.io-client';
import { Observable, fromEvent } from 'rxjs';
import { User } from './user.interface';

@Injectable({
  providedIn: 'root'
})
export class WebsocketService {
  private socket: Socket;
  
  // Signal pour gérer l'état de la liste des utilisateurs
  private usersSignal = signal<User[]>([]);
  public users = this.usersSignal.asReadonly();

  constructor() {
    // Initialise la connexion WebSocket avec des paramètres de reconnexion automatique
    // pour assurer une connexion robuste en cas de perte de connexion
    this.socket = io('http://localhost:3000', {
      reconnection: true,
      reconnectionDelay: 1000,
      reconnectionAttempts: 5
    });

    // Souscription aux nouveaux utilisateurs directement dans le service
    this.onNewUser().subscribe((user: User) => {
      this.usersSignal.update(users => [...users, user]);
    });
  }

  /**
   * Surveille l'état de connexion au serveur WebSocket
   * Retourne un Observable qui émet lorsque la connexion est établie
   */
  onConnect(): Observable<void> {
    return fromEvent(this.socket, 'connect') as Observable<void>;
  }

  /**
   * Surveille les déconnexions du serveur WebSocket
   * Retourne un Observable qui émet lorsque la connexion est perdue
   */
  onDisconnect(): Observable<void> {
    return fromEvent(this.socket, 'disconnect') as Observable<void>;
  }

  /**
   * Capture les erreurs de communication WebSocket
   * Retourne un Observable qui émet en cas d'erreur de connexion ou de communication
   */
  onError(): Observable<Error> {
    return fromEvent(this.socket, 'error') as Observable<Error>;
  }

  /**
   * Écoute l'arrivée de nouveaux utilisateurs
   * Retourne un Observable qui émet à chaque fois qu'un nouvel utilisateur est ajouté
   */
  onNewUser(): Observable<User> {
    return fromEvent(this.socket, 'new-user') as Observable<User>;
  }

  /**
   * Envoie les informations d'un nouvel utilisateur au serveur
   * Cette méthode est void car elle ne fait qu'émettre un événement sans attendre de réponse
   */
  sendMessage(user: User): void {
    this.socket.emit('new-user', user);
  }
}

Bonne pratique

Dans cet exemple, nous utilisons :

• Un Signal pour gérer l'état de la liste des utilisateurs car c'est une donnée qui change au fil du temps et qui doit être réactive dans l'interface
• Des Observables pour gérer les flux d'événements (connexion, nouveaux utilisateurs, erreurs) car ce sont des événements qui surviennent de manière asynchrone

Cette séparation permet de :

• Avoir un état prévisible et facile à déboguer avec les signaux
• Garder la puissance des opérateurs RxJS pour la gestion des flux d'événements

2. Utilisation dans un composant

Le composant devient plus simple car il n'a plus besoin de gérer la liste des utilisateurs :

user.component.ts

import { Component, inject } from '@angular/core';
import { WebsocketService } from '../../core/services/websocket.service';
import { User } from '../../core/interfaces/user.interface';
import { toSignal } from '@angular/core/rxjs-interop';

@Component({
  selector: 'app-user',
  standalone: true,
  template: `
    @if (isConnected()) {
      <div class="connected">Connecté au serveur</div>
    } @else {
      <div class="disconnected">Déconnecté du serveur</div>
    }

    @if (error()) {
      <div class="error">Erreur de connexion au serveur</div>
    }

    @for (user of websocketService.users(); track user.id) {
      <div class="user-card">
        <h3>{{ user.name }}</h3>
        <p>{{ user.email }}</p>
      </div>
    }
  `
})
export class UserComponent {
    websocketService = inject(WebsocketService);
    isConnected = toSignal(this.websocketService.onConnect());
    error = toSignal(this.websocketService.onError());

    addUser(user: User) {
        this.websocketService.sendMessage(user);
    }
}

toSignal est une fonction utilitaire d'Angular qui permet de convertir un Observable en Signal. C'est particulièrement utile quand :

• Vous voulez utiliser la syntaxe des signaux (()) dans vos templates plutôt que le pipe async
• Vous souhaitez une meilleure intégration avec le système de change detection d'Angular
• Vous voulez combiner des données venant d'Observables avec d'autres Signals

Dans notre composant :

isConnected = toSignal(this.websocketService.onConnect());
error = toSignal(this.websocketService.onError());

Cette conversion nous permet de :

• Avoir une syntaxe plus concise dans le template : @if (isConnected()) au lieu de @if (isConnected$ | async)
• Bénéficier des optimisations de performance des Signals
• Gérer automatiquement la désinscription des Observables (pas besoin de takeUntilDestroyed)

Sécurité

Assurez-vous de gérer correctement l'authentification et la sécurité de vos connexions WebSocket en production.

Transformation des données avec RxJS

L'un des grands avantages de RxJS est la possibilité de transformer les flux de données. Voici un exemple détaillé :

export class UserComponent {
  // Filtrer et transformer les utilisateurs en temps réel
  latestActiveUsers$ = this.websocketService.onNewUser().pipe(
    // Ne garde que les utilisateurs actifs
    filter(user => user.status === 'active'),
    
    // Ajoute un timestamp à chaque utilisateur
    map(user => ({
      ...user,
      lastSeen: new Date()
    })),
    
    // Maintient une liste glissante des 5 derniers utilisateurs
    scan((acc, user) => {
      const users = [...acc, user];
      return users.slice(-5);
    }, [] as User[]),
    
    // Évite les émissions inutiles
    distinctUntilChanged()
  );
}

Comprendre les opérateurs RxJS

Chaque opérateur a un rôle spécifique dans la transformation du flux de données :

1. filter : Agit comme un filtre dans un tuyau

   • Ne laisse passer que les éléments qui correspondent à une condition
   • Exemple : ne garder que les utilisateurs actifs

   filter(user => user.status === 'active')

2. map : Transforme chaque élément du flux

   • Comme une usine qui modifie chaque produit qui passe
   • Utile pour enrichir ou modifier les données

   map(user => ({
     ...user,
     lastSeen: new Date()
   }))

3. scan : Accumule les valeurs comme un "reducer"

   • Garde un état entre les émissions
   • Parfait pour maintenir des listes ou des compteurs

   scan((acc, user) => {
     const users = [...acc, user];
     return users.slice(-5); // Garde les 5 derniers
   }, [] as User[])

4. distinctUntilChanged : Évite les doublons consécutifs

   • Ne laisse passer une valeur que si elle est différente de la précédente
   • Optimise les performances en évitant les rendus inutiles

Performance

Attention à ne pas surcharger vos pipes RxJS avec trop d'opérateurs. Chaque opérateur ajoute une étape de traitement. Utilisez-les judicieusement en fonction de vos besoins réels.