nami 0.11.0

A powerful, lightweight reactive framework.
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

use alloc::{boxed::Box, rc::Rc};
use core::{
    cell::{Cell, RefCell},
    fmt::Debug,
    time::Duration,
};
use executor_core::{DefaultExecutor, LocalExecutor, Task};
use nami_core::watcher::Context;

use crate::{
    Signal,
    utils::sleep,
    watcher::{WatcherManager, WatcherManagerGuard},
};
use nami_core::{SignalIdentity, observe::Origin};

/// A throttle wrapper that limits the rate of signal updates to at most once per duration.
///
/// Unlike debounce, throttle emits the first update immediately and then limits subsequent
/// updates until the throttle period expires.
pub struct Throttle<S, E>
where
    S: Signal,
{
    signal: S,
    duration: Duration,
    watchers: WatcherManager<S::Output>,
    executor: E,
    timer: Rc<RefCell<Option<Box<dyn Task<()>>>>>,
    guard: Rc<RefCell<Option<S::Guard>>>,
    throttled: Rc<Cell<bool>>,
}

impl<S, E> Debug for Throttle<S, E>
where
    S: Signal + Debug,
    E: Debug,
{
    fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
        f.debug_struct("Throttle")
            .field("signal", &self.signal)
            .field("duration", &self.duration)
            .field("watchers", &"<...>")
            .field("executor", &self.executor)
            .finish_non_exhaustive()
    }
}

impl<S, E> Clone for Throttle<S, E>
where
    S: Signal,
    E: Clone,
{
    fn clone(&self) -> Self {
        Self {
            signal: self.signal.clone(),
            duration: self.duration,
            watchers: self.watchers.clone(),
            executor: self.executor.clone(),
            timer: self.timer.clone(),
            guard: self.guard.clone(),
            throttled: self.throttled.clone(),
        }
    }
}

impl<S, E> Throttle<S, E>
where
    E: LocalExecutor + Clone + 'static,
    S: Signal,
{
    /// Creates a new throttle wrapper with a custom executor.
    #[track_caller]
    pub fn with_executor(signal: S, duration: Duration, executor: E) -> Self {
        let timer = Rc::default();
        let origin = Origin::capture::<Self>(SignalIdentity::from_rc(&timer));
        Self {
            signal,
            watchers: WatcherManager::with_origin(origin),
            duration,
            executor,
            timer,
            guard: Rc::default(),
            throttled: Rc::default(),
        }
    }
}

impl<S> Throttle<S, DefaultExecutor>
where
    S: Signal,
{
    /// Creates a new throttle wrapper with the default executor.
    pub fn new(signal: S, duration: Duration) -> Self {
        Self::with_executor(signal, duration, DefaultExecutor)
    }
}

impl<S, E> Signal for Throttle<S, E>
where
    S: Signal,
    S::Output: Clone + 'static,
    E: LocalExecutor + Clone + 'static,
{
    type Output = S::Output;
    type Guard = WatcherManagerGuard<S::Output>;

    fn get(&self) -> Self::Output {
        self.signal.get()
    }

    fn watch(&self, watcher: impl Fn(Context<Self::Output>) + 'static) -> Self::Guard {
        let signal = self.signal.clone();
        let watchers = self.watchers.clone();
        let executor = self.executor.clone();
        let timer = self.timer.clone();
        let throttled = self.throttled.clone();
        let duration = self.duration;

        // Ensure we only set up the upstream watcher once
        let _signal_guard = self.guard.borrow_mut().get_or_insert_with(|| {
            signal.watch(move |ctx| {
                // If we're currently throttled, ignore this update
                if throttled.get() {
                    return;
                }

                if watchers.is_empty() {
                    return;
                }

                // Immediately emit the update
                watchers.notify(&ctx);

                // Set throttled state and start timer
                throttled.set(true);

                let throttled = throttled.clone();
                let task = executor.spawn_local(async move {
                    sleep(duration).await;
                    // Reset throttled state after the duration
                    throttled.set(false);
                });

                *timer.borrow_mut() = Some(Box::new(task));
            })
        });

        self.watchers.register_as_guard(watcher)
    }
}