rxrust 0.15.0

A Rust implementation of Reactive Extensions.
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
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149

use crate::prelude::*;

/// Creates an observable that emits no items, just terminates with an error.
///
/// # Arguments
///
/// * `e` - An error to emit and terminate with
pub fn throw<Err>(e: Err) -> ObservableBase<ThrowEmitter<Err>> {
  ObservableBase::new(ThrowEmitter(e))
}

#[derive(Clone)]
pub struct ThrowEmitter<Err>(Err);

#[doc(hidden)]
macro_rules! throw_emitter {
  ($subscription:ty, $($marker:ident +)* $lf: lifetime) => {
  #[inline]
  fn emit<O>(self, mut subscriber: Subscriber<O, $subscription>)
  where
    O: Observer<Item=Self::Item,Err= Self::Err> + $($marker +)* $lf
  {
    subscriber.error(self.0);
  }
}
}
impl<Err> Emitter for ThrowEmitter<Err> {
  type Item = ();
  type Err = Err;
}

impl<'a, Err> LocalEmitter<'a> for ThrowEmitter<Err> {
  throw_emitter!(LocalSubscription, 'a);
}

impl<Err> SharedEmitter for ThrowEmitter<Err> {
  throw_emitter!(SharedSubscription, Send + Sync + 'static);
}

/// Creates an observable that produces no values.
///
/// Completes immediately. Never emits an error.
///
/// # Examples
/// ```
/// use rxrust::prelude::*;
///
/// observable::empty()
///   .subscribe(|v: &i32| {println!("{},", v)});
///
/// // Result: no thing printed
/// ```
pub fn empty<Item>() -> ObservableBase<EmptyEmitter<Item>> {
  ObservableBase::new(EmptyEmitter(TypeHint::new()))
}

#[derive(Clone)]
pub struct EmptyEmitter<Item>(TypeHint<Item>);

#[doc(hidden)]
macro_rules! empty_emitter {
  ($subscription:ty, $($marker:ident +)* $lf: lifetime) => {
    #[inline]
    fn emit<O>(self, mut subscriber: Subscriber<O, $subscription>)
    where
      O: Observer<Item=Self::Item,Err= Self::Err> + $($marker +)* $lf
    {
      subscriber.complete();
    }
  }
}

impl<Item> Emitter for EmptyEmitter<Item> {
  type Item = Item;
  type Err = ();
}

impl<'a, Item> LocalEmitter<'a> for EmptyEmitter<Item> {
  empty_emitter!(LocalSubscription, 'a);
}

impl<Item> SharedEmitter for EmptyEmitter<Item> {
  empty_emitter!(SharedSubscription, Send + Sync + 'static);
}
/// Creates an observable that never emits anything.
///
/// Neither emits a value, nor completes, nor emits an error.
pub fn never() -> ObservableBase<NeverEmitter> {
  ObservableBase::new(NeverEmitter())
}

#[derive(Clone)]
pub struct NeverEmitter();

#[doc(hidden)]
macro_rules! never_emitter {
  ($subscription:ty, $($marker:ident +)* $lf: lifetime) => {
  #[inline]
  fn emit<O>(self, _subscriber: Subscriber<O, $subscription>)
  where
    O: Observer<Item=Self::Item,Err= Self::Err> + $($marker +)* $lf
  {
  }
}
}

impl Emitter for NeverEmitter {
  type Item = ();
  type Err = ();
}

impl<'a> LocalEmitter<'a> for NeverEmitter {
  never_emitter!(LocalSubscription, 'a);
}

impl SharedEmitter for NeverEmitter {
  never_emitter!(SharedSubscription, Send + Sync + 'static);
}

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

  #[test]
  fn throw() {
    let mut value_emitted = false;
    let mut completed = false;
    let mut error_emitted = String::new();
    observable::throw(String::from("error")).subscribe_all(
      // helping with type inference
      |_| value_emitted = true,
      |e| error_emitted = e,
      || completed = true,
    );
    assert!(!value_emitted);
    assert!(!completed);
    assert_eq!(error_emitted, "error");
  }

  #[test]
  fn empty() {
    let mut hits = 0;
    let mut completed = false;
    observable::empty().subscribe_complete(|()| hits += 1, || completed = true);

    assert_eq!(hits, 0);
    assert!(completed);
  }
}