rxrust 0.7.2

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
150
151
152
153
154
155
156
157
158
159
160
161
162

use crate::subscription::Publisher;
use std::cell::RefCell;
use std::rc::Rc;
use std::sync::{Arc, Mutex};

/// An Observer is a consumer of values delivered by an Observable. One for each
/// type of notification delivered by the Observable: `next`, `error`,
/// and `complete`.
///
/// `Item` the type of the elements being emitted.
/// `Err`the type of the error may propagating.
pub trait Observer<Item, Err>:
  ObserverNext<Item> + ObserverError<Err> + ObserverComplete
{
}

/// ObserverNext can consume the items delivered by an Observable.
pub trait ObserverNext<Item> {
  fn next(&mut self, value: Item);
}

/// ObserverNext can consume the error delivered by an Observable.
pub trait ObserverError<Err> {
  fn error(&mut self, err: Err);
}

/// ObserverNext can consume the complete notify delivered by an Observable.
pub trait ObserverComplete {
  fn complete(&mut self);
}

impl<Item, Err, T> Observer<Item, Err> for T where
  T: ObserverNext<Item> + ObserverError<Err> + ObserverComplete
{
}

pub(crate) macro observer_next_proxy_impl(
  $t: ty,  $host_ty: ident, $name:ident, <$($generics: tt),*>) {
  impl<$($generics ,)* Item> ObserverNext<Item> for $t
    where $host_ty: ObserverNext<Item> {
    #[inline(always)]
    fn next(&mut self, value: Item) { self.$name.next(value); }
  }
}

pub(crate) macro observer_error_proxy_impl(
  $t: ty,  $host_ty: ident, $name:ident, <$($generics: tt),*>) {
  impl<$($generics ,)* Err> ObserverError<Err> for $t
    where $host_ty: ObserverError<Err>
  {
    #[inline(always)]
    fn error(&mut self, err: Err) { self.$name.error(err); }
  }
}

pub(crate) macro observer_complete_proxy_impl(
  $t: ty, $host_ty: ident, $name: tt, <$($generics: tt),*>) {
  impl<$($generics),*> ObserverComplete for $t
    where $host_ty: ObserverComplete
  {
    #[inline(always)]
    fn complete(&mut self) { self.$name.complete(); }
  }
}

macro observer_next_pointer_proxy_impl(
  $item: ty,$ty: ty, <$($generics: tt),*>) {
  impl<$($generics),*> ObserverNext<$item> for $ty {
    #[inline(always)]
    fn next(&mut self, value: $item) { (&mut **self).next(value); }
  }
}

macro observer_error_pointer_proxy_impl(
  $err: ty, $ty: ty, <$($generics: tt),*>) {
  impl<$($generics),*> ObserverError<$err> for $ty {
    #[inline(always)]
    fn error(&mut self, err: $err) { (&mut **self).error(err); }
  }
}

macro observer_complete_pointer_proxy_impl(
  $ty: ty, <$($generics: tt),*>) {
  impl<$($generics),*> ObserverComplete for $ty {
    #[inline(always)]
  fn complete(&mut self) { (&mut **self).complete(); }
  }
}

// implement `Observer` for Box<dyn Observer<Item, Err> + 'a>
observer_next_pointer_proxy_impl!(
  Item, Box<dyn Observer<Item, Err> + 'a>, <'a, Item, Err>);
observer_error_pointer_proxy_impl!(
  Err, Box<dyn Observer<Item, Err> + 'a>, <'a, Item, Err>);
observer_complete_pointer_proxy_impl!(
  Box<dyn Observer<Item, Err> + 'a>, <'a, Item, Err>);

// implement `Observer` for Box<dyn Observer<Item, Err> + Send + Sync>
observer_next_pointer_proxy_impl!(
  Item, Box<dyn Observer<Item, Err> + Send + Sync>, <'a, Item, Err>);
observer_error_pointer_proxy_impl!(
  Err, Box<dyn Observer<Item, Err> + Send + Sync>, <'a, Item, Err>);
observer_complete_pointer_proxy_impl!(
  Box<dyn Observer<Item, Err> + Send + Sync>, <'a, Item, Err>);

// implement `Observer` for Box<dyn Publisher<Item, Err> + 'a>
observer_next_pointer_proxy_impl!(
  Item, Box<dyn Publisher<Item, Err> + 'a>, <'a, Item, Err>);
observer_error_pointer_proxy_impl!(
  Err, Box<dyn Publisher<Item, Err> + 'a>, <'a, Item, Err>);
observer_complete_pointer_proxy_impl!(
  Box<dyn Publisher<Item, Err> + 'a>, <'a, Item, Err>);

// implement `Observer` for Box<dyn Publisher<Item, Err> + Send + Sync>
observer_next_pointer_proxy_impl!(
  Item, Box<dyn Publisher<Item, Err> + Send + Sync>, <'a, Item, Err>);
observer_error_pointer_proxy_impl!(
  Err, Box<dyn Publisher<Item, Err> + Send + Sync>, <'a, Item, Err>);
observer_complete_pointer_proxy_impl!(
  Box<dyn Publisher<Item, Err> + Send + Sync>, <'a, Item, Err>);

impl<Item, S> ObserverNext<Item> for Arc<Mutex<S>>
where
  S: ObserverNext<Item>,
{
  fn next(&mut self, value: Item) { self.lock().unwrap().next(value); }
}

impl<Err, S> ObserverError<Err> for Arc<Mutex<S>>
where
  S: ObserverError<Err>,
{
  fn error(&mut self, err: Err) { self.lock().unwrap().error(err); }
}

impl<S> ObserverComplete for Arc<Mutex<S>>
where
  S: ObserverComplete,
{
  fn complete(&mut self) { self.lock().unwrap().complete(); }
}

impl<Item, S> ObserverNext<Item> for Rc<RefCell<S>>
where
  S: ObserverNext<Item>,
{
  fn next(&mut self, value: Item) { self.borrow_mut().next(value); }
}

impl<Err, S> ObserverError<Err> for Rc<RefCell<S>>
where
  S: ObserverError<Err>,
{
  fn error(&mut self, err: Err) { self.borrow_mut().error(err); }
}

impl<S> ObserverComplete for Rc<RefCell<S>>
where
  S: ObserverComplete,
{
  fn complete(&mut self) { self.borrow_mut().complete(); }
}