Programmation réactive en Rust : bibliothèques, exemples et pratique

Programmation réactive en Rust : des observables aux flux asynchrones

La programmation réactive est un paradigme axé sur le traitement de flux de données asynchrones et la propagation des changements. Dans l'écosystème Rust, elle est mise en œuvre grâce à une combinaison d'un système de types puissant, d'une gestion sécurisée de la mémoire et de primitives asynchrones modernes. Dans cet article, nous explorerons les bibliothèques clés, les motifs de conception et des exemples pratiques de programmation réactive en Rust.

Bibliothèques essentielles pour la programmation réactive en Rust

Rust ne possède pas de « standard » unique pour la programmation réactive, mais plusieurs bibliothèques matures couvrent différents niveaux d'abstraction :

• RxRust — le port de la populaire ReactiveX (Rx) pour Rust. Fournit Observable, Observer, Scheduler et un large éventail d'opérateurs (map, filter, flat_map, merge, etc.).
• futures-rs — la bibliothèque fondamentale pour les calculs asynchrones. Le trait Stream est la base des flux réactifs.
• async-std et tokio — des runtimes asynchrones offrant des canaux (channels), des minuteurs et d'autres primitives pour les chaînes réactives.
• frunk — une bibliothèque de programmation fonctionnelle incluant des motifs réactifs comme Lens et State.

Exemple 1 : Observable simple avec RxRust

RxRust implémente le motif « observateur » : un Observable génère des événements, un Observer s'y abonne.

// Cargo.toml: rxrust = "1.0"
use rxrust::prelude::*;

fn main() {
    let observable = observable::from_iter(vec![1, 2, 3, 4, 5]);

    observable
        .filter(|x| x % 2 == 0)
        .map(|x| x * 10)
        .subscribe(|x| println!("Reçu : {}", x));
}

Dans cet exemple, un flux de nombres 1 à 5 est créé, les nombres pairs sont filtrés, multipliés par 10, puis affichés. La chaîne d'opérateurs est exécutée paresseusement — seulement après l'appel à subscribe.

Exemple 2 : Flux asynchrones avec futures-rs

La bibliothèque futures-rs fournit le trait Stream, équivalent asynchrone de Iterator. Les combinateurs de StreamExt permettent de construire des chaînes réactives.

// Cargo.toml: futures = "0.3"
use futures::stream::{self, StreamExt};

#[tokio::main]
async fn main() {
    let stream = stream::iter(0..10)
        .filter(|x| futures::future::ready(x % 2 == 0))
        .map(|x| x * 10)
        .take(3);

    let result: Vec<i32> = stream.collect().await;
    println!("Résultat : {:?}", result); // [0, 20, 40]
}

Ici, filter accepte une fonction asynchrone, ce qui permet d'intégrer des appels à des API externes ou des bases de données dans la chaîne. take(3) limite le nombre d'éléments — un exemple de « contre-pression » (backpressure).

Exemple 3 : Canaux réactifs avec async-std

Les canaux sont un moyen classique de mettre en œuvre des interactions réactives entre composants. async-std fournit des canaux multi-producteurs et multi-consommateurs.

// Cargo.toml: async-std = "1.12"
use async_std::channel;
use async_std::task;

#[async_std::main]
async fn main() {
    let (sender, receiver) = channel::unbounded();

    // Producteur
    task::spawn(async move {
        for i in 0..5 {
            sender.send(i).await.unwrap();
            task::sleep(std::time::Duration::from_millis(100)).await;
        }
    });

    // Consommateur (réactif)
    while let Some(msg) = receiver.recv().await {
        println!("Reçu : {}", msg);
        if msg == 4 {
            break;
        }
    }
}

Les canaux sont parfaits pour implémenter un Event Bus, un bus d'événements ou des systèmes réactifs simples avec plusieurs sources de données.

Motifs avancés : combinaison de flux

La programmation réactive prend tout son sens lors de la combinaison de plusieurs sources de données.