Reactive Programming in Rust: Libraries and Examples |…

Reactive Programming in Rust: From Observable Objects to Asynchronous Streams

Reactive programming is a paradigm focused on processing asynchronous data streams and propagating changes. In the Rust ecosystem, it is implemented through a combination of a powerful type system, safe memory management, and modern asynchronous primitives. In this article, we will explore key libraries, patterns, and practical examples of reactive programming in Rust.

Core Libraries for Reactive Programming in Rust

Rust does not have a single "standard" for reactive programming, but there are several mature libraries covering different levels of abstraction:

RxRust — a port of the popular ReactiveX (Rx) for Rust. Provides Observable, Observer, Scheduler, and a wide set of operators (map, filter, flat_map, merge, etc.).
futures-rs — a fundamental library for working with asynchronous computations. The Stream trait is the foundation of reactive streams.
async-std and tokio — asynchronous runtimes providing channels, timers, and other primitives for reactive chains.
frunk — a library for functional programming, including reactive patterns like Lens and State.

Example 1: Simple Observable with RxRust

RxRust implements the "observer" pattern: an Observable generates events, and an Observer subscribes to them.

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

In this example, a stream of numbers 1..5 is created, even numbers are filtered, multiplied by 10, and printed. The operator chain is executed lazily — only after calling subscribe.

Example 2: Asynchronous Streams with futures-rs

The futures-rs library provides the Stream trait, which is an asynchronous counterpart to Iterator. Combinators from StreamExt allow building reactive chains.

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

Here, filter accepts an asynchronous function, allowing calls to external APIs or databases to be embedded within the chain. take(3) limits the number of elements — this is an example of backpressure.

Example 3: Reactive Channels with async-std

Channels are a classic way to implement reactive interactions between components. async-std provides multi-producer, multi-consumer channels.

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

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

    // Consumer (reactive)    while let Some(msg) = receiver.recv().await {        println!("Received: {}", msg);        if msg == 4 {            break;        }    }}

Channels are ideal for implementing an Event Bus, event bus, or simple reactive systems with multiple data sources.