reactor_rs 0.0.5

reactor-rust is an implementation of the Reactive-Streams.
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

use std::marker::PhantomData;

pub const REQUEST_MAX: usize = 0x7fff_ffff;

pub trait Publisher {
  type Item;
  type Error;
  fn subscribe(
    self,
    subscriber: impl Subscriber<Item = Self::Item, Error = Self::Error> + 'static + Send,
  ) where
    Self: Sized;
}

pub trait Subscription: Sized {
  fn request(&self, n: usize);
  fn cancel(&self);
}

pub trait Subscriber {
  type Item;
  type Error;
  fn on_complete(&self);
  fn on_next(&self, t: Self::Item)
  where
    Self: Sized;

  fn on_subscribe(&self, subscription: impl Subscription)
  where
    Self: Sized;

  fn on_error(&self, e: Self::Error)
  where
    Self: Sized;
}

pub struct Subscribers;

impl Subscribers {
  pub fn next<T, E, F>(action: F) -> impl Subscriber<Item = T, Error = E>
  where
    F: Fn(T),
  {
    NextSubscriber::new(action)
  }

  pub fn noop<T, E>() -> impl Subscriber<Item = T, Error = E> {
    NoopSubscriber::new()
  }
}

pub(crate) struct NextSubscriber<T, E, F>
where
  F: Fn(T),
{
  f: F,
  _t: PhantomData<T>,
  _e: PhantomData<E>,
}

impl<T, E, F> NextSubscriber<T, E, F>
where
  F: Fn(T),
{
  pub(crate) fn new(f: F) -> NextSubscriber<T, E, F> {
    NextSubscriber {
      f,
      _t: PhantomData,
      _e: PhantomData,
    }
  }
}

impl<T, E, F> Subscriber for NextSubscriber<T, E, F>
where
  F: Fn(T),
{
  type Item = T;
  type Error = E;

  fn on_subscribe(&self, subscription: impl Subscription) {
    subscription.request(REQUEST_MAX);
  }

  fn on_complete(&self) {}
  fn on_next(&self, t: T) {
    (self.f)(t)
  }
  fn on_error(&self, _e: E) {}
}

pub(crate) struct NoopSubscriber<T, E> {
  _t: PhantomData<T>,
  _e: PhantomData<E>,
}

impl<T, E> NoopSubscriber<T, E> {
  pub(crate) fn new() -> NoopSubscriber<T, E> {
    NoopSubscriber {
      _t: PhantomData,
      _e: PhantomData,
    }
  }
}

impl<T, E> Subscriber for NoopSubscriber<T, E> {
  type Item = T;
  type Error = E;
  fn on_subscribe(&self, subscription: impl Subscription) {
    subscription.request(REQUEST_MAX);
  }
  fn on_complete(&self) {}
  fn on_next(&self, _t: T) {}
  fn on_error(&self, _e: E) {}
}