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

use crate::prelude::*;
use crate::subject::{LocalSubject, SharedSubject};

pub struct ConnectableObservable<Source, Subject> {
  source: Source,
  subject: Subject,
}

pub trait Connect {
  type Unsub;
  fn connect(self) -> Self::Unsub;
}

impl<S, O, U, Sub> RawSubscribable<Sub>
  for ConnectableObservable<S, Subject<O, U>>
where
  S: RawSubscribable<Subscriber<O, U>>,
  Subject<O, U>: RawSubscribable<Sub>,
{
  type Unsub = <Subject<O, U> as RawSubscribable<Sub>>::Unsub;

  #[inline(always)]
  fn raw_subscribe(self, subscriber: Sub) -> Self::Unsub {
    self.subject.raw_subscribe(subscriber)
  }
}

impl<'a, Item, Err, S> ConnectableObservable<S, LocalSubject<'a, Item, Err>> {
  pub fn local(observable: S) -> Self {
    Self {
      source: observable,
      subject: Subject::local(),
    }
  }
}

impl<Source, Item, Err> ConnectableObservable<Source, SharedSubject<Item, Err>>
where
  Source: IntoShared,
{
  pub fn shared(
    observable: Source,
  ) -> ConnectableObservable<Source::Shared, SharedSubject<Item, Err>> {
    ConnectableObservable {
      source: observable.to_shared(),
      subject: Subject::shared(),
    }
  }
}

impl<Source, O, U> Connect for ConnectableObservable<Source, Subject<O, U>>
where
  Source: RawSubscribable<Subscriber<O, U>>,
{
  type Unsub = Source::Unsub;
  fn connect(self) -> Self::Unsub {
    self.source.raw_subscribe(Subscriber {
      observer: self.subject.observers,
      subscription: self.subject.subscription,
    })
  }
}

impl<Source, Subject> IntoShared for ConnectableObservable<Source, Subject>
where
  Source: IntoShared,
  Subject: IntoShared,
{
  type Shared = ConnectableObservable<Source::Shared, Subject::Shared>;
  fn to_shared(self) -> Self::Shared {
    ConnectableObservable {
      source: self.source.to_shared(),
      subject: self.subject.to_shared(),
    }
  }
}

impl<Source, Subject> Fork for ConnectableObservable<Source, Subject>
where
  Subject: Fork,
{
  type Output = Subject::Output;
  #[inline(always)]
  fn fork(&self) -> Self::Output { self.subject.fork() }
}

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

  #[test]
  fn smoke() {
    let o = observable::of(100);
    let connected = ConnectableObservable::local(o);
    let mut first = 0;
    let mut second = 0;
    connected.fork().subscribe(|v| first = v);
    connected.fork().subscribe(|v| second = v);

    connected.connect();
    assert_eq!(first, 100);
    assert_eq!(second, 100);
  }

  #[test]
  fn fork_and_shared() {
    let o = observable::of(100);
    let connected = ConnectableObservable::local(o).to_shared();
    connected.fork().subscribe(|_| {});
    connected.fork().subscribe(|_| {});

    connected.connect();
  }
}