rxrust 1.0.0-rc.4

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

use std::ops::Add;

use crate::{
  observable::{CoreObservable, ObservableType},
  ops::{
    filter_map::FilterMap,
    reduce::{Reduce, ReduceInitialFn},
  },
  subscription::Subscription,
};

/// A trait for types that can be averaged.
///
/// This trait extends `Add` to support accumulation.
/// It adds a `div_count` method to calculate the final average by dividing the
/// sum by the count of items.
pub trait Averageable: Add<Output = Self> + Sized {
  /// Divides the accumulated sum by the count of items to produce the average.
  fn div_count(self, count: usize) -> Self;
}

impl Averageable for f32 {
  fn div_count(self, count: usize) -> Self { self / (count as f32) }
}

impl Averageable for f64 {
  fn div_count(self, count: usize) -> Self { self / (count as f64) }
}

/// Implements `Averageable` for integer types.
///
/// This macro provides a blanket implementation of the `Averageable` trait
/// for various integer types, allowing them to be used with the `average`
/// operator.
///
/// The `div_count` method for integers performs integer division.
macro_rules! impl_averageable_int {
  ($($t:ty),*) => {
    $(
      impl Averageable for $t {
        fn div_count(self, count: usize) -> Self {
          self / (count as $t)
        }
      }
    )*
  };
}

impl_averageable_int!(u8, u16, u32, u64, u128, usize, i8, i16, i32, i64, i128, isize);

/// Average operator (type-level wrapper to avoid huge nested types)
///
/// The actual implementation is built as a small pipeline (reduce + filter_map)
/// and delegated to in `subscribe`.
#[derive(Clone)]
pub struct Average<Source> {
  pub(crate) source: Source,
}

impl<Source> Average<Source> {
  #[inline]
  pub fn new(source: Source) -> Self { Self { source } }
}

impl<Source> ObservableType for Average<Source>
where
  Source: ObservableType,
{
  type Item<'a>
    = Source::Item<'a>
  where
    Self: 'a;

  type Err = Source::Err;
}

type AvgAcc<Item> = (Option<Item>, usize);

type AveragePipeline<Source, Item> = FilterMap<
  Reduce<Source, ReduceInitialFn<fn(AvgAcc<Item>, Item) -> AvgAcc<Item>>, AvgAcc<Item>>,
  fn(AvgAcc<Item>) -> Option<Item>,
>;

#[inline]
fn build_pipeline<'a, Source>(source: Source) -> AveragePipeline<Source, Source::Item<'a>>
where
  Source: ObservableType,
  Source::Item<'a>: Averageable,
{
  type Item<'a, S> = <S as ObservableType>::Item<'a>;

  type ReducerFn<'a, S> = fn(AvgAcc<Item<'a, S>>, Item<'a, S>) -> AvgAcc<Item<'a, S>>;

  let reducer: ReducerFn<'a, Source> = (|acc, value| match acc {
    (Some(sum), count) => (Some(sum + value), count + 1usize),
    (None, count) => (Some(value), count + 1usize),
  }) as _;

  let finish: fn(AvgAcc<Item<'a, Source>>) -> Option<Item<'a, Source>> =
    (|(sum, count): AvgAcc<Item<'a, Source>>| sum.map(|s| s.div_count(count))) as _;

  FilterMap {
    source: Reduce { source, strategy: ReduceInitialFn(reducer), initial: Some((None, 0usize)) },
    func: finish,
  }
}

impl<Source, C, Unsub> CoreObservable<C> for Average<Source>
where
  Source: ObservableType,
  for<'a> Source::Item<'a>: Averageable,
  for<'a> AveragePipeline<Source, Source::Item<'a>>: CoreObservable<C, Unsub = Unsub>,
  Unsub: Subscription,
{
  type Unsub = Unsub;

  #[inline]
  fn subscribe(self, observer: C) -> Self::Unsub {
    build_pipeline::<Source>(self.source).subscribe(observer)
  }
}