another-rxrust 0.0.46

A different implementation than `rxRust` for easier use of `ReactiveX` in `Rust`.
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70

use crate::prelude::*;
use std::sync::Arc;

#[derive(Clone)]
pub struct AsyncSubject<'a, Item>
where
  Item: Clone + Send + Sync,
{
  subject: Arc<subject::Subject<'a, Item>>,
}

impl<'a, Item> AsyncSubject<'a, Item>
where
  Item: Clone + Send + Sync,
{
  pub fn new() -> AsyncSubject<'a, Item> {
    AsyncSubject {
      subject: Arc::new(subjects::Subject::new()),
    }
  }

  pub fn next(&self, item: Item) {
    self.subject.next(item);
  }
  pub fn error(&self, err: RxError) {
    self.subject.error(err);
  }
  pub fn complete(&self) {
    self.subject.complete();
  }
  pub fn observable(&self) -> Observable<'a, Item> {
    self.subject.observable().take_last(1).clone()
  }
}

#[cfg(test)]
mod tset {
  use crate::prelude::*;

  #[test]
  fn basic() {
    let sbj = subjects::AsyncSubject::new();

    sbj.observable().subscribe(
      print_next_fmt!("{}"),
      print_error!(),
      print_complete!(),
    );

    sbj.next(1);
    sbj.next(2);
    sbj.next(3);
    sbj.complete();
  }

  #[test]
  fn error() {
    let sbj = subjects::AsyncSubject::new();

    sbj.observable().subscribe(
      print_next_fmt!("{}"),
      print_error_as!(&str),
      print_complete!(),
    );

    sbj.next(1);
    sbj.next(2);
    sbj.error(RxError::from_error("ERR!"));
  }
}