id_effect 0.3.0

Effect<A, E, R> (sync + async), capability DI, pipe — interpreter-style, no bundled executor
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

//! FSM-style [`state_scan`] — step function returns optional outputs per element.
//!
//! Unlike [`Stream::scan`], which emits on every element, `state_scan` only yields when the
//! step function returns `Some(output)`.

use crate::streaming::stream::Stream;

/// Step function for [`state_scan`]: `(state, item) -> (next_state, optional_output)`.
pub type StateStep<S, A, B> = Box<dyn FnMut(S, A) -> (S, Option<B>) + Send>;

/// Stateful scan with optional emissions — useful for simple FSM transitions.
#[inline]
pub fn state_scan<S, A, B, E, R>(
  stream: Stream<A, E, R>,
  init: S,
  mut step: impl FnMut(S, A) -> (S, Option<B>) + Send + 'static,
) -> Stream<B, E, R>
where
  S: 'static,
  A: Send + 'static,
  B: Send + 'static,
  E: Send + 'static,
  R: 'static,
{
  Stream::new(move |r: &mut R| {
    Box::pin(async move {
      let mut upstream = stream;
      let mut state = init;
      let mut out = Vec::new();
      loop {
        let Some(chunk) = upstream.poll_next_chunk(r).await? else {
          break;
        };
        for item in chunk.into_vec() {
          let (next, maybe) = step(state, item);
          state = next;
          if let Some(b) = maybe {
            out.push(b);
          }
        }
      }
      Ok(out)
    })
  })
}

impl<A, E, R> Stream<A, E, R>
where
  A: Send + 'static,
  E: Send + 'static,
  R: 'static,
{
  /// Convenience wrapper around [`state_scan`] with the receiver as upstream.
  #[inline]
  pub fn state_scan<S, B, F>(self, init: S, step: F) -> Stream<B, E, R>
  where
    S: 'static,
    B: Send + 'static,
    F: FnMut(S, A) -> (S, Option<B>) + Send + 'static,
  {
    state_scan(self, init, step)
  }
}

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

  fn block_on<F: core::future::Future>(fut: F) -> F::Output {
    pollster::block_on(fut)
  }

  #[derive(Clone, Copy, Debug, PartialEq, Eq)]
  enum St {
    Idle,
    Active,
  }

  #[test]
  fn emits_only_on_transition_to_active() {
    let stream = Stream::from_iterable([0u8, 1, 1, 0, 1]).state_scan(St::Idle, |st, x| {
      let next = if x > 0 { St::Active } else { St::Idle };
      let emit = matches!((st, next), (St::Idle, St::Active));
      (next, emit.then_some(x))
    });
    let out = block_on(stream.run_collect().run(&mut ())).expect("collect");
    assert_eq!(out, vec![1, 1]);
  }

  #[test]
  fn state_scan_no_transition_skips_emit() {
    let stream = Stream::from_iterable([1u8, 1, 1])
      .state_scan(0u8, |st, x| if x == st { (st, None) } else { (x, Some(x)) });
    let out = block_on(stream.run_collect().run(&mut ())).expect("collect");
    assert_eq!(out, vec![1]);
  }

  #[test]
  fn state_scan_emits_each_change() {
    let stream = Stream::from_iterable([1u8, 2, 3]).state_scan(0u8, |_st, x| (x, Some(x)));
    let out = block_on(stream.run_collect().run(&mut ())).expect("collect");
    assert_eq!(out, vec![1, 2, 3]);
  }

  #[test]
  fn free_state_scan_function() {
    let stream = state_scan(Stream::from_iterable([1u8, 2]), 0u32, |s, x| {
      (s + x as u32, Some(s))
    });
    let out = block_on(stream.run_collect().run(&mut ())).expect("collect");
    assert_eq!(out, vec![0, 1]);
  }

  #[test]
  fn empty_upstream_yields_empty() {
    let stream = Stream::<u8, (), ()>::from_iterable(core::iter::empty())
      .state_scan(0u32, |s, x| (s + x as u32, Some(s)));
    let out = block_on(stream.run_collect().run(&mut ())).expect("collect");
    assert!(out.is_empty());
  }
}