README.md

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.

 # use rxrust::prelude::*;
 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.

 # use rxrust::prelude::*;
 let o = observable::from_iter(0..10);
 o.clone().subscribe(|_| {println!("consume in first")});
 o.clone().subscribe(|_| {println!("consume in second")});

Scheduler

use rxrust::prelude::*;

observable::from_iter(0..10)
  .subscribe_on(Schedulers::NewThread)
  .map(|v| v*2)
  .to_shared()
  .observe_on(Schedulers::NewThread)
  .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;

observable::from_future(future::ready(1))
  .to_shared()
  .subscribe(move |v| println!("subscribed with {}", v));

// because all future in rxrust are execute async, so we wait a second to see
// the print, no need to use this line in your code.
std::thread::sleep(std::time::Duration::new(1, 0));

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.