Add this to your Cargo.toml:
toml
[dependencies]
rxrust = "1.0.0-beta.0"
```rust 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 2 4 6 8 1 3 5 7 9" will be printed.
```
In rxrust almost all extensions consume the upstream. So when you try to subscribe a stream twice, the compiler will complain.
rust ignore
# 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.
rust
# use rxrust::prelude::*;
let o = observable::from_iter(0..10);
o.clone().subscribe(|_| println!("consume in first"));
o.clone().subscribe(|_| println!("consume in second"));
If you want share the same observable, you can use Subject.
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 Scheduler to implement custom Scheduler.
Some Observable Ops (such as delay, debounce) need the ability to delay, futures-time supports this ability when set with timer feature, but you can also customize it by setting the newtimer function to NEWTIMER_FN variant and removing the timer feature.
```rust
use rxrust::prelude::*;
// FuturesThreadPoolScheduler is the alias of futures::executor::ThreadPool.
let threads_scheduler = FuturesThreadPoolScheduler::new().unwrap();
observable::fromiter(0..10) .subscribeon(threadsscheduler.clone()) .map(|v| v*2) .observeonthreads(threadsscheduler) .subscribe(|v| println!("{},", v)); ```
Also, rxrust supports WebAssembly by enabling the feature wasm-scheduler and using the crate wasm-bindgen. Simple example is here.
Just use observable::from_future to convert a Future to an observable sequence.
```rust use rxrust::prelude::*;
let mut schedulerpool = FuturesLocalSchedulerPool::new(); observable::fromfuture(std::future::ready(1), scheduler_pool.spawner()) .subscribe(move |v| println!("subscribed with {}", v));
// Wait task finish.
scheduler_pool.run();
```
A from_future_result function also provided to propagating error from Future.
See missing features to know what rxRust does not have yet.
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: