rxrust/observable/
trivial.rs

1use crate::prelude::*;
2
3/// Creates an observable that emits no items, just terminates with an error.
4///
5/// # Arguments
6///
7/// * `e` - An error to emit and terminate with
8pub fn throw<Err>(e: Err) -> ObservableBase<ThrowEmitter<Err>> {
9  ObservableBase::new(ThrowEmitter(e))
10}
11
12#[derive(Clone)]
13pub struct ThrowEmitter<Err>(Err);
14
15#[doc(hidden)]
16macro_rules! throw_emitter {
17  ($subscription:ty, $($marker:ident +)* $lf: lifetime) => {
18  #[inline]
19  fn emit<O>(self, mut subscriber: Subscriber<O, $subscription>)
20  where
21    O: Observer<Item=Self::Item,Err= Self::Err> + $($marker +)* $lf
22  {
23    subscriber.error(self.0);
24  }
25}
26}
27impl<Err> Emitter for ThrowEmitter<Err> {
28  type Item = ();
29  type Err = Err;
30}
31
32impl<'a, Err> LocalEmitter<'a> for ThrowEmitter<Err> {
33  throw_emitter!(LocalSubscription, 'a);
34}
35
36impl<Err> SharedEmitter for ThrowEmitter<Err> {
37  throw_emitter!(SharedSubscription, Send + Sync + 'static);
38}
39
40/// Creates an observable that produces no values.
41///
42/// Completes immediately. Never emits an error.
43///
44/// # Examples
45/// ```
46/// use rxrust::prelude::*;
47///
48/// observable::empty()
49///   .subscribe(|v: &i32| {println!("{},", v)});
50///
51/// // Result: no thing printed
52/// ```
53pub fn empty<Item>() -> ObservableBase<EmptyEmitter<Item>> {
54  ObservableBase::new(EmptyEmitter(TypeHint::new()))
55}
56
57#[derive(Clone)]
58pub struct EmptyEmitter<Item>(TypeHint<Item>);
59
60#[doc(hidden)]
61macro_rules! empty_emitter {
62  ($subscription:ty, $($marker:ident +)* $lf: lifetime) => {
63    #[inline]
64    fn emit<O>(self, mut subscriber: Subscriber<O, $subscription>)
65    where
66      O: Observer<Item=Self::Item,Err= Self::Err> + $($marker +)* $lf
67    {
68      subscriber.complete();
69    }
70  }
71}
72
73impl<Item> Emitter for EmptyEmitter<Item> {
74  type Item = Item;
75  type Err = ();
76}
77
78impl<'a, Item> LocalEmitter<'a> for EmptyEmitter<Item> {
79  empty_emitter!(LocalSubscription, 'a);
80}
81
82impl<Item> SharedEmitter for EmptyEmitter<Item> {
83  empty_emitter!(SharedSubscription, Send + Sync + 'static);
84}
85/// Creates an observable that never emits anything.
86///
87/// Neither emits a value, nor completes, nor emits an error.
88pub fn never() -> ObservableBase<NeverEmitter> {
89  ObservableBase::new(NeverEmitter())
90}
91
92#[derive(Clone)]
93pub struct NeverEmitter();
94
95#[doc(hidden)]
96macro_rules! never_emitter {
97  ($subscription:ty, $($marker:ident +)* $lf: lifetime) => {
98  #[inline]
99  fn emit<O>(self, _subscriber: Subscriber<O, $subscription>)
100  where
101    O: Observer<Item=Self::Item,Err= Self::Err> + $($marker +)* $lf
102  {
103  }
104}
105}
106
107impl Emitter for NeverEmitter {
108  type Item = ();
109  type Err = ();
110}
111
112impl<'a> LocalEmitter<'a> for NeverEmitter {
113  never_emitter!(LocalSubscription, 'a);
114}
115
116impl SharedEmitter for NeverEmitter {
117  never_emitter!(SharedSubscription, Send + Sync + 'static);
118}
119
120#[cfg(test)]
121mod test {
122  use crate::prelude::*;
123
124  #[test]
125  fn throw() {
126    let mut value_emitted = false;
127    let mut completed = false;
128    let mut error_emitted = String::new();
129    observable::throw(String::from("error")).subscribe_all(
130      // helping with type inference
131      |_| value_emitted = true,
132      |e| error_emitted = e,
133      || completed = true,
134    );
135    assert!(!value_emitted);
136    assert!(!completed);
137    assert_eq!(error_emitted, "error");
138  }
139
140  #[test]
141  fn empty() {
142    let mut hits = 0;
143    let mut completed = false;
144    observable::empty().subscribe_complete(|()| hits += 1, || completed = true);
145
146    assert_eq!(hits, 0);
147    assert!(completed);
148  }
149}
rxrust/observable/trivial.rs

rxrust/observable/
trivial.rs
