ferroui_reactive/concurrency/
virtual_time_scheduler.rs1use super::{IScheduler, IStopwatch, IStopwatchProvider, ScheduledAction, ScheduledItem};
5use crate::internal::PriorityQueue;
6use ferroui_system::IDisposable;
7use std::cell::{Cell, RefCell};
8use std::rc::{Rc, Weak};
9use std::time::{Duration, SystemTime};
10
11pub trait VirtualTimeClock: 'static {
15 type Absolute: Ord + Copy + 'static;
17 type Relative: Copy + 'static;
19
20 fn add(absolute: Self::Absolute, relative: Self::Relative) -> Self::Absolute;
22
23 fn to_system_time(absolute: Self::Absolute) -> SystemTime;
25
26 fn to_relative(duration: Duration) -> Self::Relative;
28
29 fn subtract(absolute: Self::Absolute, _duration: Duration) -> Self::Absolute {
32 absolute
33 }
34
35 fn adjust_due_time(due_time: Self::Absolute, _clock: Self::Absolute) -> Self::Absolute {
39 due_time
40 }
41}
42
43pub struct VirtualTimeScheduler<C: VirtualTimeClock> {
47 this: Weak<VirtualTimeScheduler<C>>,
48 clock: Cell<C::Absolute>,
49 is_enabled: Cell<bool>,
50 queue: RefCell<PriorityQueue<C::Absolute, Rc<ScheduledItem<C::Absolute>>>>,
51}
52
53impl<C: VirtualTimeClock> VirtualTimeScheduler<C> {
54 pub fn new(initial_clock: C::Absolute) -> Rc<Self> {
56 Rc::new_cyclic(|this| Self {
57 this: this.clone(),
58 clock: Cell::new(initial_clock),
59 is_enabled: Cell::new(false),
60 queue: RefCell::new(PriorityQueue::new()),
61 })
62 }
63
64 fn this(&self) -> Rc<dyn IScheduler> {
65 self.this.upgrade().expect("the scheduler is alive while it runs work")
66 }
67
68 pub fn clock(&self) -> C::Absolute {
70 self.clock.get()
71 }
72
73 pub fn set_clock(&self, clock: C::Absolute) {
75 self.clock.set(clock);
76 }
77
78 pub fn is_enabled(&self) -> bool {
80 self.is_enabled.get()
81 }
82
83 pub fn schedule_absolute(&self, due_time: C::Absolute, action: ScheduledAction) -> Rc<dyn IDisposable> {
85 let due_time = C::adjust_due_time(due_time, self.clock.get());
86 let item = ScheduledItem::new(due_time, action);
87 self.queue.borrow_mut().push(due_time, item.clone());
88 item
89 }
90
91 pub fn schedule_relative(&self, due_time: C::Relative, action: ScheduledAction) -> Rc<dyn IDisposable> {
93 self.schedule_absolute(C::add(self.clock.get(), due_time), action)
94 }
95
96 fn peek_next(&self) -> Option<(C::Absolute, Rc<ScheduledItem<C::Absolute>>)> {
98 let mut queue = self.queue.borrow_mut();
99 loop {
100 let (due, item) = queue.peek().map(|(due, item)| (*due, item.clone()))?;
101 if item.is_canceled() {
102 queue.pop();
103 continue;
104 }
105 return Some((due, item));
106 }
107 }
108
109 fn run_next(&self, due: C::Absolute, item: Rc<ScheduledItem<C::Absolute>>) {
110 self.queue.borrow_mut().remove_where(|queued| Rc::ptr_eq(queued, &item));
113 if due > self.clock.get() {
114 self.clock.set(due);
115 }
116 item.invoke(&self.this());
117 }
118
119 pub fn start(&self) {
122 if self.is_enabled.replace(true) {
123 return;
124 }
125 while self.is_enabled.get() {
126 match self.peek_next() {
127 Some((due, item)) => self.run_next(due, item),
128 None => self.is_enabled.set(false),
129 }
130 }
131 }
132
133 pub fn stop(&self) {
135 self.is_enabled.set(false);
136 }
137
138 pub fn advance_to(&self, time: C::Absolute) {
141 if time < self.clock.get() {
142 panic!("Specified argument was out of the range of valid values. (Parameter 'time')");
143 }
144 if time == self.clock.get() {
145 return;
146 }
147 if self.is_enabled.replace(true) {
148 panic!("Cannot advance while running: AdvanceTo.");
149 }
150 while self.is_enabled.get() {
151 match self.peek_next() {
152 Some((due, item)) if due <= time => self.run_next(due, item),
153 _ => self.is_enabled.set(false),
154 }
155 }
156 self.clock.set(time);
157 }
158
159 pub fn advance_by(&self, time: C::Relative) {
161 let target = C::add(self.clock.get(), time);
162 if target < self.clock.get() {
163 panic!("Specified argument was out of the range of valid values. (Parameter 'time')");
164 }
165 if target == self.clock.get() {
166 return;
167 }
168 if self.is_enabled.get() {
169 panic!("Cannot advance while running: AdvanceBy.");
170 }
171 self.advance_to(target);
172 }
173
174 pub fn sleep(&self, time: C::Relative) {
176 let target = C::add(self.clock.get(), time);
177 if target < self.clock.get() {
178 panic!("Specified argument was out of the range of valid values. (Parameter 'time')");
179 }
180 self.clock.set(target);
181 }
182}
183
184impl<C: VirtualTimeClock> IScheduler for VirtualTimeScheduler<C> {
185 fn now(&self) -> SystemTime {
186 C::to_system_time(self.clock.get())
187 }
188
189 fn schedule(&self, action: ScheduledAction) -> Rc<dyn IDisposable> {
190 self.schedule_absolute(self.clock.get(), action)
191 }
192
193 fn schedule_after(&self, due_time: Duration, action: ScheduledAction) -> Rc<dyn IDisposable> {
194 self.schedule_relative(C::to_relative(due_time), action)
195 }
196
197 fn schedule_at(&self, due_time: SystemTime, action: ScheduledAction) -> Rc<dyn IDisposable> {
198 match due_time.duration_since(self.now()) {
199 Ok(relative) => self.schedule_relative(C::to_relative(relative), action),
200 Err(past) => self.schedule_absolute(C::subtract(self.clock.get(), past.duration()), action),
203 }
204 }
205
206 fn as_stopwatch_provider(&self) -> Option<&dyn IStopwatchProvider> {
207 Some(self)
208 }
209}
210
211impl<C: VirtualTimeClock> IStopwatchProvider for VirtualTimeScheduler<C> {
212 fn start_stopwatch(&self) -> Box<dyn IStopwatch> {
213 Box::new(VirtualTimeStopwatch::<C> { scheduler: self.this.clone(), start: self.now() })
214 }
215}
216
217struct VirtualTimeStopwatch<C: VirtualTimeClock> {
219 scheduler: Weak<VirtualTimeScheduler<C>>,
220 start: SystemTime,
221}
222
223impl<C: VirtualTimeClock> IStopwatch for VirtualTimeStopwatch<C> {
224 fn elapsed(&self) -> Duration {
225 match self.scheduler.upgrade() {
226 Some(scheduler) => scheduler.now().duration_since(self.start).unwrap_or(Duration::ZERO),
227 None => Duration::ZERO,
228 }
229 }
230}