rx-rust

Reactive Extensions for Rust. rx-rust is a toolkit for composing asynchronous and event-driven programs by chaining observables and operators in a declarative style, inspired by ReactiveX.

• 100+ operators, one per file, grouped the way reactivex.io groups them.
• #![forbid(unsafe_code)], no required runtime dependency: the async runtime (Tokio, async-std, smol, or the futures executors) is selected by a feature flag.
• A built-in stop signal: an observer that has seen enough returns Flow::Stop, and the source stops pushing, even a synchronous one that is still inside subscribe.
• Crosses into async freely: a Future or a Stream becomes an observable, and an observable becomes a Future of its first item or a Stream of all of them.

Installation

Add the crate with the scheduler feature for the runtime you use:

[dependencies]
rx-rust = { version = "1.0", features = ["tokio-scheduler"] }

Pick exactly one of the scheduler features below; tokio-scheduler is the most common choice. The time-based operators (delay, debounce, timeout, Interval, Timer, …) take a scheduler argument, and only compile when a scheduler feature is enabled.

Feature flags

single-threaded (and so local-pool-scheduler) is mutually exclusive with thread-pool-scheduler, tokio-scheduler, async-std-scheduler and smol-scheduler; enabling both is a compile error.

Quick start

Build pipelines by combining operators from ObservableExt, then subscribe with callbacks or with your own Observer:

use rx_rust::{observable::ObservableExt, operators::creating::range::Range};

Range::new(1..=5)
    .map(|value| value * 2)
    .filter(|value| *value % 3 == 0)
    .subscribe_with_callback(
        |value| println!("next: {value}"),
        |termination| println!("done: {termination:?}"),
    );

subscribe returns a Subscription, and dropping it unsubscribes. A synchronous source such as Range delivers everything before subscribe returns, so the example above can ignore it; a source that emits later must have its subscription kept alive for as long as the events are wanted:

#[tokio::main]
async fn main() {
    use rx_rust::{observable::ObservableExt, operators::creating::interval::Interval};
    use std::time::Duration;
    use tokio::time::sleep;

    let scheduler = tokio::runtime::Handle::current();
    let subscription = Interval::new(Duration::from_millis(10), scheduler, None)
        .subscribe_with_callback(|tick| println!("tick {tick}"), |_| {});

    sleep(Duration::from_millis(35)).await;
    drop(subscription); // unsubscribes: no more ticks
}

Core concepts

Observable<'or, T, E> is a source of T items that ends with a Termination<E>: either Completed or Error(E). A source that cannot fail uses E = Infallible, and some conversions (into_future, into_stream) are only available for those. 'or is the lifetime an observer may borrow for; 'static for anything that goes through a scheduler.

An Observer<T, E> has two methods:

use rx_rust::observer::{Flow, Observer, Termination};

struct Printer;

impl Observer<i32, std::convert::Infallible> for Printer {
    fn on_next(&mut self, value: i32) -> Flow {
        println!("{value}");
        Flow::Continue
    }

    fn on_termination(self, termination: Termination<std::convert::Infallible>) {
        println!("{termination:?}");
    }
}

on_termination takes self: a termination is the last event, and the observer is consumed by it.

Stopping from the observer

on_next returns a Flow. Flow::Continue lets the source go on; Flow::Stop is a promise that the observer accepts nothing more, not even a termination. The source stops pushing and drops the observer. This is how a consumer ends an infinite synchronous source, which never leaves subscribe and so has no subscription to drop yet:

use rx_rust::{observable::ObservableExt, observer::Flow, operators::creating::range::Range};

let mut seen = Vec::new();
Range::new(1..).subscribe_with_callback(
    |value| {
        seen.push(value);
        if value == 3 { Flow::Stop } else { Flow::Continue }
    },
    |_| {}, // not called: a stopped observer is not terminated
);
assert_eq!(seen, [1, 2, 3]);

A callback that returns () is Flow::Continue, so the common case needs no trailing value. Operators forward the answer of their downstream to the source, so take, first, take_until and friends stop their source the same way. When the source is not the one calling on_next — when it delivers through a scheduler — the value is queued and the stop reaches the source with the next push, which is why Flow::Continue is only a hint and disposal is still needed.

Schedulers

Every time-based operator takes a value implementing Scheduler, so nothing is global and one program can drive different pipelines on different runtimes:

#[tokio::main]
async fn main() {
    use futures::StreamExt;
    use rx_rust::{observable::ObservableExt, operators::creating::from_iter::FromIter};
    use std::time::Duration;

    let scheduler = tokio::runtime::Handle::current();
    let values = FromIter::new(vec![1, 2, 3])
        .delay(Duration::from_millis(5), scheduler)
        .into_stream()
        .collect::<Vec<_>>()
        .await;

    assert_eq!(values, [1, 2, 3]);
}

subscribe_on and observe_on move the subscription, or the delivery of events, onto a scheduler; the operators without a scheduler argument run synchronously on whichever thread pushes into them.

Futures and streams

A future becomes an observable with FromFuture, or FromTryFuture when its output is a Result whose Err should become the observable's error; both need a scheduler to run the future on. This is how a one-shot call (an HTTP request, a database lookup) goes inside flat_map, concat_map or switch_map:

#[tokio::main]
async fn main() {
    use futures::StreamExt;
    use rx_rust::{
        observable::ObservableExt,
        operators::creating::{from_future::FromFuture, range::Range},
    };

    async fn fetch_name(id: u32) -> String {
        format!("user-{id}")
    }

    let handle = tokio::runtime::Handle::current();
    let names = Range::new(1..=3)
        .concat_map(move |id| FromFuture::new(fetch_name(id), handle.clone()))
        .into_stream()
        .collect::<Vec<_>>()
        .await;

    assert_eq!(names, ["user-1", "user-2", "user-3"]);
}

In the other direction, into_future() — or into_try_future() for a source that can fail — turns an observable into a future of its first item, stopping the source as soon as that item is in. It resolves with an Option<T>, or a Result<Option<T>, E>. Put last(), collect() or reduce() in front of it to pick another item or to make the source always emit:

#[tokio::main]
async fn main() {
    use rx_rust::{
        observable::ObservableExt,
        operators::creating::{from_result::FromResult, range::Range},
    };

    let first = Range::new(1..=5).into_future().await;
    assert_eq!(first, Some(1));

    let sum = Range::new(1..=5).sum().into_future().await;
    assert_eq!(sum, Some(15));

    let failed = FromResult::new(Err::<i32, _>("boom")).into_try_future().await;
    assert_eq!(failed, Err("boom"));
}

A whole sequence, rather than one item of it, is consumed through into_stream() / into_try_stream(), which turn an observable into a Stream of its items; FromStream / FromTryStream go the other way. These four need the futures feature, which every scheduler feature enables.

Backpressure

rx-rust has no request(n): an observable pushes, and the only word its observer has is Flow::Stop. Demand-driven consumption is what Stream is for, so the pull side lives there rather than in a second observable type — the Flowable of RxJava — and the two meet at the conversions. FromStream polls its stream only after on_next returns, so the stream's pace is kept through every synchronous operator, up to the first asynchronous one. In the other direction into_stream() keeps every item that arrives between two polls, so a source faster than the consumer grows its buffer without bound; into_stream_with(buffer) bounds it with a StreamBuffer that decides what survives — Latest keeps the newest item, Bounded a fixed number of them, dropping the oldest or the newest — or folds the items into one, in an implementation of your own:

use futures::{FutureExt, StreamExt};
use rx_rust::{
    observable::ObservableExt, observer::Observer,
    operators::others::observable_try_stream::Latest,
    subject::publish_subject::PublishSubject,
};
use std::convert::Infallible;

let mut subject = PublishSubject::<_, Infallible>::new();
let mut stream = subject.clone().into_stream_with(Latest::new());
assert_eq!(stream.next().now_or_never(), None); // the first poll subscribes

subject.on_next(1);
subject.on_next(2);
subject.on_next(3);
assert_eq!(stream.next().now_or_never(), Some(Some(3)));

Single, Maybe and Completable

rx-rust deliberately has no Single, Maybe or Completable types. ReactiveX added them for languages that lack a native way to express "one asynchronous result"; Rust has one, and async/await composes it better than any operator set could:

The operators of Single map onto plain async code — map/flatMap is async { f(future.await?) }, zip is futures::try_join!, amb is futures::future::select, timeout/delay are the runtime's timeout/sleep, retry is a loop around .await, cache is FutureExt::shared, and dropping the future is its Disposable. into_try_future() is the bridge: its output is exactly a Maybe, and with collect() in front, a Single. Keeping one-shot results as futures also keeps this crate's operator set a single one: every operator works on Observable, and none needs a second implementation for Single.

Subjects

A subject is both an Observer and an Observable: push into one end, subscribe to the other. They are the hot sources of the crate, and what publish / share / replay multicast through.

Subjects are Clone; clone one to keep a sending end after subscribing the other:

use rx_rust::{
    observable::ObservableExt,
    observer::{Observer, Termination},
    subject::publish_subject::PublishSubject,
};

let mut seen = Vec::new();

let subject = PublishSubject::<i32, std::convert::Infallible>::new();
let mut sender = subject.clone();
let subscription = subject.subscribe_with_callback(|value| seen.push(value), |_| {});

let _ = sender.on_next(1);
let _ = sender.on_next(2);
sender.on_termination(Termination::Completed);

drop(subscription);
assert_eq!(seen, [1, 2]);

A cold observable is turned into a shared, hot one with share() (publish().ref_count()): the source is subscribed when the first observer arrives and unsubscribed when the last leaves. publish() / replay() / multicast() give the ConnectableController back for manual connect() / disconnect().

Operators

The full list, with a doc example for each, is on docs.rs. Sources are types in rx_rust::operators::creating; everything else is a method of ObservableExt.

Project layout

• src/observable – The Observable trait, ObservableExt (every operator as a method), and the boxed observables.
• src/observer – The Observer trait, Flow, Termination, and the callback observer.
• src/disposable – Disposable and the Subscription that disposes on drop.
• src/operators – One operator per file, grouped by category to mirror ReactiveX terminology.
• src/subject – The subjects.
• src/scheduler – The Scheduler trait and its adapters for the supported runtimes.
• src/utils – Shared machinery: the lock wrapper, the Rc/Arc and Send/Sync abstraction behind the single-threaded build, serialized delivery.
• tests/ – One integration test file per operator, using the same checklist of cases for each; great as executable documentation.

Contributing

See DEVELOPMENT.md for the review checklist and the test case names every operator is expected to cover, and AGENTS.md for the build and test commands. Requires Rust 1.88 or later.

License

MIT. See LICENSE.