Add this to your Cargo.toml:
toml
[dependencies]
rxrust = "0.10.0"
```rust 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.
```
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")});
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.
```rust use rxrust::prelude::*; use futures::executor::ThreadPool;
let poolscheduler = ThreadPool::new().unwrap(); observable::fromiter(0..10) .subscribeon(poolscheduler.clone()) .map(|v| v*2) .intoshared() .observeon(poolscheduler) .intoshared() .subscribe(|v| {println!("{},", v)}); ```
just use observable::from_future to convert a Future to an observable sequence.
```rust use rxrust::prelude::*; use futures::{ future, executor::LocalPool };
let mut localscheduler = LocalPool::new(); observable::fromfuture(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.
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: