Crate rxrust[−][src]

Expand description

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 Rust implementation of Reactive Extensions

Usage

1.0.x version require rust nightly before GAT stable, 0.15.0` work with stable rust.

Add this to your Cargo.toml:

[dependencies]
rxrust = "1.0.0-alpha.1"

Example

use rxrust:: prelude::*;

let mut numbers = observable::from_iter(0..10);
// create an even stream by filter
let even = numbers.clone().filter(|v| v % 2 == 0);
// create 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")});

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)
  .into_shared()
  .observe_on(pool_scheduler)
  .into_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 does not have yet.

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

Macros

impl_local_shared_both

impl_local_shared_both is a macro helper to implement both the LocalObservable and SharedObservable at once. You can use @ctx to unify the same meaning but different types in local and shared implementation. See (local)[local] and (‘shared’)[shared] mod to learn what type you can use. Define an method macro use to write the method code of the LocalObservable and SharedObservable. This macro should named with method and always accept three arguments are $self $observer and $ctx. $self and $observer are the arguments of the actual_subscribe method. $ctx is either impl_local mod or impl_shared.

of_sequence

Creates an observable producing a multiple values.