rxrust/ops/
map_to.rs

1use crate::prelude::*;
2use crate::{complete_proxy_impl, error_proxy_impl, is_stopped_proxy_impl};
3
4#[derive(Clone)]
5pub struct MapToOp<S, B> {
6  pub(crate) source: S,
7  pub(crate) value: B,
8}
9
10#[doc(hidden)]
11macro_rules! observable_impl {
12    ($subscription:ty, $($marker:ident +)* $lf: lifetime) => {
13    fn actual_subscribe<O > (
14      self,
15      subscriber: Subscriber<O, $subscription>,
16    ) -> Self::Unsub
17    where O: Observer<Item=Self::Item,Err= Self::Err> + $($marker +)* $lf {
18      let value = self.value;
19      self.source.actual_subscribe(Subscriber {
20        observer: MapToObserver {
21          observer: subscriber.observer,
22          value,
23          _marker: TypeHint::new(),
24        },
25        subscription: subscriber.subscription,
26      })
27    }
28  }
29}
30
31impl<S, B> Observable for MapToOp<S, B>
32where
33  S: Observable,
34{
35  type Item = B;
36  type Err = S::Err;
37}
38
39impl<'a, B, S> LocalObservable<'a> for MapToOp<S, B>
40where
41  S: LocalObservable<'a>,
42  B: Clone + 'a,
43  S::Item: 'a,
44{
45  type Unsub = S::Unsub;
46  observable_impl!(LocalSubscription,'a);
47}
48
49impl<B, S> SharedObservable for MapToOp<S, B>
50where
51  S: SharedObservable,
52  B: Clone + Send + Sync + 'static,
53  S::Item: 'static,
54{
55  type Unsub = S::Unsub;
56  observable_impl!(SharedSubscription, Send + Sync + 'static);
57}
58
59#[derive(Clone)]
60pub struct MapToObserver<O, B, Item> {
61  observer: O,
62  value: B,
63  _marker: TypeHint<*const Item>,
64}
65
66impl<Item, Err, O, B> Observer for MapToObserver<O, B, Item>
67where
68  O: Observer<Item = B, Err = Err>,
69  B: Clone,
70{
71  type Item = Item;
72  type Err = Err;
73  fn next(&mut self, _: Item) { self.observer.next(self.value.clone()) }
74  error_proxy_impl!(Err, observer);
75  complete_proxy_impl!(observer);
76  is_stopped_proxy_impl!(observer);
77}
78
79#[cfg(test)]
80mod test {
81  use crate::prelude::*;
82
83  #[test]
84  fn primitive_type() {
85    let mut i = 0;
86    observable::from_iter(100..101)
87      .map_to(5)
88      .subscribe(|v| i += v);
89    assert_eq!(i, 5);
90  }
91
92  #[test]
93  fn reference_lifetime_should_work() {
94    let mut i = 0;
95
96    observable::of(100).map_to(5).subscribe(|v| i += v);
97    assert_eq!(i, 5);
98  }
99
100  #[test]
101  fn fork_and_shared() {
102    // type to type can fork
103    let m = observable::from_iter(0..100).map_to(5);
104    m.map_to(6).into_shared().subscribe(|_| {});
105
106    // type mapped to other type can fork
107    let m = observable::from_iter(vec!['a', 'b', 'c']).map_to(1);
108    m.map_to(2.0).into_shared().subscribe(|_| {});
109
110    // ref to ref can fork
111    let m = observable::of(&1).map_to(3);
112    m.map_to(4).into_shared().subscribe(|_| {});
113  }
114
115  #[test]
116  fn map_types_mixed() {
117    let mut i = 0;
118    observable::from_iter(vec!['a', 'b', 'c'])
119      .map_to(1)
120      .subscribe(|v| i += v);
121    assert_eq!(i, 3);
122  }
123
124  #[test]
125  fn bench() { do_bench(); }
126
127  benchmark_group!(do_bench, bench_map_to);
128
129  fn bench_map_to(b: &mut bencher::Bencher) { b.iter(primitive_type); }
130}
rxrust/ops/map_to.rs

rxrust/ops/
map_to.rs
