[−][src]Crate rxrust
Reactive extensions library for Rust: a library for Reactive Programming using Observable, to make it easier to compose asynchronous or callback-based code.
rxRust: a zero cost Rust implementation of Reactive Extensions
Usage
Add this to your Cargo.toml:
[dependencies]
rxrust = "0.8.3"
Example
use rxrust:: prelude::*; let mut numbers = observable::from_iter(0..10); // crate a even stream by filter let even = numbers.clone().filter(|v| v % 2 == 0); // crate an odd stream by filter let odd = numbers.clone().filter(|v| v % 2 != 0); // merge odd and even stream again even.merge(odd).subscribe(|v| print!("{} ", v, )); // "0 1 2 3 4 5 6 7 8 9" will be printed.
Clone Stream
In rxrust
almost all extensions consume the upstream. So when you try to subscribe a stream twice, the compiler will complain.
let o = observable::from_iter(0..10); o.subscribe(|_| { println!("consume in first")} ); o.subscribe(|_| { println!("consume in second")} );
In this case, we must clone the stream.
let o = observable::from_iter(0..10); o.clone().subscribe(|_| {println!("consume in first")}); o.clone().subscribe(|_| {println!("consume in second")});
Scheduler
rxrust
use the runtime of the Future
as the scheduler, LocalPool
and ThreadPool
in futures::executor
can be used as schedulers directly, and tokio::runtime::Runtime
also supported, but need enable the feature futures-scheduler
. Across LocalScheduler
and SharedScheduler
to implement custom Scheduler
.
use rxrust::prelude::*; use futures::executor::ThreadPool; let pool_scheduler = ThreadPool::new().unwrap(); observable::from_iter(0..10) .subscribe_on(pool_scheduler.clone()) .map(|v| v*2) .to_shared() .observe_on(pool_scheduler) .to_shared() .subscribe(|v| {println!("{},", v)});
Converts from a Future
just use observable::from_future
to convert a Future
to an observable sequence.
use rxrust::prelude::*; use futures::{ future, executor::LocalPool }; let mut local_scheduler = LocalPool::new(); observable::from_future(future::ready(1), local_scheduler.spawner()) .subscribe(move |v| println!("subscribed with {}", v)); // Wait `LocalPool` finish. local_scheduler.run();
A from_future_result
function also provided to propagating error from Future
.
Missing Features List
See missing features to know what rxRust not have for now.
All contributions are welcome
We are looking for contributors! Feel free to open issues for asking questions, suggesting features or other things!
Help and contributions can be any of the following:
- use the project and report issues to the project issues page
- documentation and README enhancement (VERY important)
- continuous improvement in a ci Pipeline
- implement any unimplemented operator, remember to create a pull request before you start your code, so other people know you are work on it.
Modules
inner_deref | |
observable | |
observer | |
ops | |
prelude | |
scheduler | |
shared | |
subject | |
subscriber | |
subscription |
Macros
impl_observable | |
impl_throttle_observer | |
of_sequence | Creates an observable producing a multiple values. |