another-rxrust

Crates.ioanother-rxrust
lib.rsanother-rxrust
version0.0.46
sourcesrc
created_at2023-03-28 03:15:11.441239
updated_at2024-03-26 10:47:42.311412
descriptionA different implementation than `rxRust` for easier use of `ReactiveX` in `Rust`.
homepagehttps://github.com/CODIANZ/another-rxrust
repositoryhttps://github.com/CODIANZ/another-rxrust
max_upload_size
id822701
size232,980
Terukazu Inoue (terukazu-inoue)

documentation

https://docs.rs/another-rxrust/

README

another-rxrust

why new implementation?

rxRust is a Rust language implementation of ReactiveX, but it is not suitable for complex combinations of observable. For those who use ReactiveX in other languages such as rxjs(TypeScript) or rxcpp, rxRust is a very difficult library.

Therefore, I created another-rxrust, thinking that I needed a library that allows observable to be connected in the same way as other platforms, and that ReactiveX can be enjoyed in Rust.

In addition, ReactiveX may not be the best solution if the purpose is to parallelize heavy processing and speed it up. However, ReactiveX is one answer for complex combinations of non-blocking I/O and error handling.

Implementation policy

  • It is assumed that the values and functions that can be emitted may be shared between threads.
  • Value to emit should be Clone + Send + Sync only.
  • Use move to emit values ​​as much as possible.
  • Functions should be Fn() + Send + Sync only.
  • Errors are type-erased using std::any.
    • Thus requiring a 'static lifetime.
    • You can use anyhow.
  • Prioritize flexibility over memory efficiency and execution speed.

Implementation status

See implementation status.

Installation

cargo add  another-rxrust

or

[dependencies]
another-rxrust = {}

If you want to use it on the web, set it as follows.

[dependencies]
another-rxrust = {features=["web"]}

Samples

from_iter, map, zip

use another_rxrust::prelude::*;
use another_rxrust::{print_next_fmt, print_error, print_complete};

fn main() {
  let ob = observables::from_iter(0..10);
  ob.zip(&[ob.map(|x| x + 10), ob.map(|x| x + 20)]).subscribe(
    print_next_fmt!("{:?}"),
    print_error!(),
    print_complete!(),
  );
}
// [console-results]
// next - [0, 10, 20]
// next - [1, 11, 21]
// next - [2, 12, 22]
// next - [3, 13, 23]
// next - [4, 14, 24]
// next - [5, 15, 25]
// next - [6, 16, 26]
// next - [7, 17, 27]
// next - [8, 18, 28]
// next - [9, 19, 29]
// complete

subject, scheduler, take, sample

use another_rxrust::prelude::*;
use std::{thread, time};
use another_rxrust::{print_next_fmt, print_error, print_complete};

fn main() {
  let sbj = subjects::Subject::new();
  observables::interval(
    time::Duration::from_millis(100),
    schedulers::new_thread_scheduler(),
  )
  .sample(sbj.observable())
  .take(3)
  .subscribe(print_next_fmt!("{}"), print_error!(), print_complete!());

  (0..3).for_each(|_| {
    thread::sleep(time::Duration::from_millis(500));
    sbj.next(());
  });
  sbj.complete();
  thread::sleep(time::Duration::from_millis(500));
}
// [console-results (Depends on execution environment)]
// next - 3
// next - 8
// next - 13
// complete

just, error, emptry, never, flat_map...

use another_rxrust::prelude::*;
use std::{thread, time};

fn main() {
  // observable creator function
  fn ob() -> Observable<'static, i32> {
    Observable::create(|s| {
      s.next(100);
      s.next(200);
      s.complete();
    })
  }

  observables::from_iter(1..10)
    .observe_on(schedulers::new_thread_scheduler())
    .flat_map(|x| match x {
      1 => observables::empty(),
      2 => observables::just(x),
      3 => ob().map(move |y| (y + x)),
      4 => observables::error(RxError::from_error("some error")),
      _ => observables::never(),
    })
    .map(|x| format!("{}", x))
    .on_error_resume_next(|e| {
      ob().map(move |x| {
        format!(
          "resume {:?} {}",
          e.downcast_ref::<&str>(),
          x
        )
      })
    })
    .subscribe(
      |x| println!("{}", x),
      |e| println!("{:?}", e.downcast_ref::<&str>()),
      || println!("complete"),
    );

  thread::sleep(time::Duration::from_millis(500));
}
// [console-results]
// next 2
// next 103
// next 203
// next resume "some error" 100
// next resume "some error" 200
// complete
Commit count: 241

cargo fmt