Programación reactiva en Rust: bibliotecas, ejemplos y práctica

Programación reactiva en Rust: de objetos observables a flujos asíncronos

La programación reactiva es un paradigma orientado al procesamiento de flujos de datos asíncronos y la propagación de cambios. En el ecosistema de Rust, se implementa mediante una combinación de un potente sistema de tipos, gestión segura de memoria y primitivas asíncronas modernas. En este artículo, exploraremos las bibliotecas clave, los patrones y ejemplos prácticos de programación reactiva en Rust.

Bibliotecas principales para programación reactiva en Rust

En Rust no existe un único «estándar» de programación reactiva, pero hay varias bibliotecas maduras que cubren diferentes niveles de abstracción:

• RxRust — un puerto de la popular ReactiveX (Rx) para Rust. Proporciona Observable, Observer, Scheduler y un amplio conjunto de operadores (map, filter, flat_map, merge, etc.).
• futures-rs — biblioteca fundamental para trabajar con cómputos asíncronos. El trait Stream es la base de los flujos reactivos.
• async-std y tokio — runtimes asíncronos que proporcionan canales (channels), temporizadores y otras primitivas para cadenas reactivas.
• frunk — biblioteca para programación funcional que incluye patrones reactivos como Lens y State.

Ejemplo 1: Observable simple con RxRust

RxRust implementa el patrón «observador»: un Observable genera eventos y un Observer se suscribe a ellos.

// 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!("Recibido: {}", x));}

En este ejemplo, se crea un flujo con los números del 1 al 5, se filtran los pares, se multiplican por 10 y se muestran. La cadena de operadores se ejecuta de forma perezosa — solo después de llamar a subscribe.

Ejemplo 2: Flujos asíncronos con futures-rs

La biblioteca futures-rs proporciona el trait Stream, que es el equivalente asíncrono de Iterator. Los combinadores de StreamExt permiten construir cadenas reactivas.

// 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!("Resultado: {:?}", result); // [0, 20, 40]}

Aquí, filter acepta una función asíncrona, lo que permite incorporar llamadas a APIs externas o bases de datos dentro de la cadena. take(3) limita la cantidad de elementos — esto es un ejemplo de «contrapresión» (backpressure).

Ejemplo 3: Canales reactivos con async-std

Los canales son una forma clásica de implementar interacciones reactivas entre componentes. async-std proporciona canales multiproductor y multiconsumidor.

// 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();

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

    // Consumidor (reactivo)    while let Some(msg) = receiver.recv().await {        println!("Recibido: {}", msg);        if msg == 4 {            break;        }    }}

Los canales son ideales para implementar un Bus de Eventos, un bus de eventos o sistemas reactivos simples con múltiples fuentes de datos.

Patrones avanzados: combinación de flujos