1use std::cmp::Ordering;
34use std::sync::atomic::{AtomicU64, Ordering as AtomicOrdering};
35use std::time::{SystemTime, UNIX_EPOCH};
36
37#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash)]
46pub struct Timestamp(i64);
47
48impl Timestamp {
49 pub fn from_micros(us: i64) -> Self {
51 Self(us)
52 }
53
54 pub fn from_secs(secs: i64) -> Self {
56 Self(secs * 1_000_000)
57 }
58
59 pub fn as_micros(&self) -> i64 {
61 self.0
62 }
63
64 pub fn as_secs(&self) -> i64 {
66 self.0 / 1_000_000
67 }
68
69 pub fn now() -> Self {
71 let d = SystemTime::now()
72 .duration_since(UNIX_EPOCH)
73 .expect("system clock before UNIX epoch");
74 Self(d.as_micros() as i64)
75 }
76}
77
78pub trait Clock: Send + Sync + 'static {
88 fn now(&self) -> Timestamp;
90}
91
92pub struct SystemClock;
94
95impl Clock for SystemClock {
96 fn now(&self) -> Timestamp {
97 Timestamp::now()
98 }
99}
100
101pub struct TestClock {
105 now: Timestamp,
106}
107
108impl TestClock {
109 pub fn new(now: Timestamp) -> Self {
111 Self { now }
112 }
113
114 pub fn set(&mut self, now: Timestamp) {
116 self.now = now;
117 }
118
119 pub fn advance_secs(&mut self, secs: i64) {
121 self.now = Timestamp::from_micros(self.now.as_micros() + secs * 1_000_000);
122 }
123
124 pub fn advance_micros(&mut self, us: i64) {
126 self.now = Timestamp::from_micros(self.now.as_micros() + us);
127 }
128}
129
130impl Clock for TestClock {
131 fn now(&self) -> Timestamp {
132 self.now
133 }
134}
135
136static NEXT_DEADLINE_ID: AtomicU64 = AtomicU64::new(1);
141
142#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
144pub struct DeadlineId(u64);
145
146impl DeadlineId {
147 fn next() -> Self {
148 Self(NEXT_DEADLINE_ID.fetch_add(1, AtomicOrdering::Relaxed))
149 }
150}
151
152struct DeadlineEntry<T> {
158 id: DeadlineId,
159 deadline: Timestamp,
160 event: T,
161}
162
163pub struct Deadlines<T: 'static> {
171 entries: Vec<DeadlineEntry<T>>,
172}
173
174impl<T: 'static> Deadlines<T> {
175 pub fn new() -> Self {
177 Self {
178 entries: Vec::new(),
179 }
180 }
181
182 pub fn schedule(&mut self, deadline: Timestamp, event: T) -> DeadlineId {
185 let id = DeadlineId::next();
186 let pos = self
187 .entries
188 .binary_search_by(|e| e.deadline.cmp(&deadline).then(Ordering::Less))
189 .unwrap_or_else(|i| i);
190 self.entries.insert(
191 pos,
192 DeadlineEntry {
193 id,
194 deadline,
195 event,
196 },
197 );
198 id
199 }
200
201 pub fn cancel(&mut self, id: DeadlineId) -> bool {
203 if let Some(pos) = self.entries.iter().position(|e| e.id == id) {
204 self.entries.remove(pos);
205 true
206 } else {
207 false
208 }
209 }
210
211 pub fn drain_overdue(&mut self, now: Timestamp) -> Vec<T> {
216 let split = self.entries.partition_point(|e| e.deadline <= now);
218 if split == 0 {
219 return Vec::new();
220 }
221 self.entries.drain(..split).map(|e| e.event).collect()
222 }
223
224 pub fn len(&self) -> usize {
226 self.entries.len()
227 }
228
229 pub fn is_empty(&self) -> bool {
231 self.entries.is_empty()
232 }
233
234 pub fn next_deadline(&self) -> Option<Timestamp> {
236 self.entries.first().map(|e| e.deadline)
237 }
238}
239
240impl<T: 'static> Default for Deadlines<T> {
241 fn default() -> Self {
242 Self::new()
243 }
244}
245
246#[cfg(test)]
251mod tests {
252 use super::*;
253
254 #[test]
257 fn timestamp_from_secs_roundtrip() {
258 let ts = Timestamp::from_secs(1000);
259 assert_eq!(ts.as_secs(), 1000);
260 assert_eq!(ts.as_micros(), 1_000_000_000);
261 }
262
263 #[test]
264 fn timestamp_from_micros() {
265 let ts = Timestamp::from_micros(123_456_789);
266 assert_eq!(ts.as_micros(), 123_456_789);
267 assert_eq!(ts.as_secs(), 123); }
269
270 #[test]
271 fn timestamp_ordering() {
272 let a = Timestamp::from_secs(100);
273 let b = Timestamp::from_secs(200);
274 assert!(a < b);
275 assert!(b > a);
276 assert_eq!(a, Timestamp::from_secs(100));
277 }
278
279 #[test]
280 fn timestamp_now_is_positive() {
281 let ts = Timestamp::now();
282 assert!(ts.as_micros() > 0);
283 }
284
285 #[test]
288 fn system_clock_returns_positive() {
289 let clock = SystemClock;
290 assert!(clock.now().as_micros() > 0);
291 }
292
293 #[test]
294 fn test_clock_manual_control() {
295 let mut clock = TestClock::new(Timestamp::from_secs(1000));
296 assert_eq!(clock.now().as_secs(), 1000);
297
298 clock.advance_secs(60);
299 assert_eq!(clock.now().as_secs(), 1060);
300
301 clock.set(Timestamp::from_secs(2000));
302 assert_eq!(clock.now().as_secs(), 2000);
303 }
304
305 #[test]
306 fn test_clock_advance_micros() {
307 let mut clock = TestClock::new(Timestamp::from_micros(0));
308 clock.advance_micros(500_000);
309 assert_eq!(clock.now().as_micros(), 500_000);
310 }
311
312 #[test]
315 fn deadline_ids_are_unique() {
316 let a = DeadlineId::next();
317 let b = DeadlineId::next();
318 assert_ne!(a, b);
319 }
320
321 #[derive(Debug, PartialEq)]
324 struct TestEvent(u32);
325
326 #[test]
327 fn schedule_and_drain() {
328 let mut deadlines = Deadlines::new();
329 deadlines.schedule(Timestamp::from_secs(100), TestEvent(1));
330 deadlines.schedule(Timestamp::from_secs(200), TestEvent(2));
331 deadlines.schedule(Timestamp::from_secs(300), TestEvent(3));
332
333 assert_eq!(deadlines.len(), 3);
334
335 let fired = deadlines.drain_overdue(Timestamp::from_secs(150));
337 assert_eq!(fired, vec![TestEvent(1)]);
338 assert_eq!(deadlines.len(), 2);
339 }
340
341 #[test]
342 fn drain_all_overdue_batch() {
343 let mut deadlines = Deadlines::new();
344 deadlines.schedule(Timestamp::from_secs(100), TestEvent(1));
345 deadlines.schedule(Timestamp::from_secs(200), TestEvent(2));
346 deadlines.schedule(Timestamp::from_secs(300), TestEvent(3));
347
348 let fired = deadlines.drain_overdue(Timestamp::from_secs(300));
350 assert_eq!(fired, vec![TestEvent(1), TestEvent(2), TestEvent(3)]);
351 assert!(deadlines.is_empty());
352 }
353
354 #[test]
355 fn drain_none_overdue() {
356 let mut deadlines = Deadlines::new();
357 deadlines.schedule(Timestamp::from_secs(200), TestEvent(1));
358
359 let fired = deadlines.drain_overdue(Timestamp::from_secs(100));
360 assert!(fired.is_empty());
361 assert_eq!(deadlines.len(), 1);
362 }
363
364 #[test]
365 fn drain_empty_queue() {
366 let mut deadlines: Deadlines<TestEvent> = Deadlines::new();
367 let fired = deadlines.drain_overdue(Timestamp::from_secs(100));
368 assert!(fired.is_empty());
369 }
370
371 #[test]
372 fn cancel_removes_entry() {
373 let mut deadlines = Deadlines::new();
374 let id = deadlines.schedule(Timestamp::from_secs(100), TestEvent(1));
375 deadlines.schedule(Timestamp::from_secs(200), TestEvent(2));
376
377 assert!(deadlines.cancel(id));
378 assert_eq!(deadlines.len(), 1);
379
380 let fired = deadlines.drain_overdue(Timestamp::from_secs(300));
381 assert_eq!(fired, vec![TestEvent(2)]);
382 }
383
384 #[test]
385 fn cancel_nonexistent_returns_false() {
386 let mut deadlines: Deadlines<TestEvent> = Deadlines::new();
387 let id = DeadlineId::next();
388 assert!(!deadlines.cancel(id));
389 }
390
391 #[test]
392 fn cancel_already_cancelled_returns_false() {
393 let mut deadlines = Deadlines::new();
394 let id = deadlines.schedule(Timestamp::from_secs(100), TestEvent(1));
395 assert!(deadlines.cancel(id));
396 assert!(!deadlines.cancel(id));
397 }
398
399 #[test]
400 fn sorted_insertion_order() {
401 let mut deadlines = Deadlines::new();
402 deadlines.schedule(Timestamp::from_secs(300), TestEvent(3));
404 deadlines.schedule(Timestamp::from_secs(100), TestEvent(1));
405 deadlines.schedule(Timestamp::from_secs(200), TestEvent(2));
406
407 let fired = deadlines.drain_overdue(Timestamp::from_secs(400));
409 assert_eq!(fired, vec![TestEvent(1), TestEvent(2), TestEvent(3)]);
410 }
411
412 #[test]
413 fn same_deadline_fires_all() {
414 let mut deadlines = Deadlines::new();
415 deadlines.schedule(Timestamp::from_secs(100), TestEvent(1));
416 deadlines.schedule(Timestamp::from_secs(100), TestEvent(2));
417
418 let fired = deadlines.drain_overdue(Timestamp::from_secs(100));
419 assert_eq!(fired.len(), 2);
420 }
421
422 #[test]
423 fn next_deadline_returns_earliest() {
424 let mut deadlines = Deadlines::new();
425 assert!(deadlines.next_deadline().is_none());
426
427 deadlines.schedule(Timestamp::from_secs(300), TestEvent(3));
428 deadlines.schedule(Timestamp::from_secs(100), TestEvent(1));
429
430 assert_eq!(deadlines.next_deadline(), Some(Timestamp::from_secs(100)));
431 }
432
433 #[test]
436 fn world_add_deadline_type_is_idempotent() {
437 let mut world = crate::world::World::new();
438 world.add_deadline_type::<TestEvent>();
439 world.add_deadline_type::<TestEvent>(); assert!(world.try_resource::<Deadlines<TestEvent>>().is_some());
441 }
442
443 #[test]
444 fn world_schedule_and_drain() {
445 let mut world = crate::world::World::new();
446 world.add_deadline_type::<TestEvent>();
447
448 world.schedule_deadline(Timestamp::from_secs(100), TestEvent(1));
449 world.schedule_deadline(Timestamp::from_secs(200), TestEvent(2));
450
451 world.drain_deadlines::<TestEvent>(Timestamp::from_secs(150));
453
454 world.update_events();
456
457 let events = world.resource::<crate::event::Events<TestEvent>>();
458 assert_eq!(events.len(), 1);
459 let fired: Vec<_> = events.read().collect();
460 assert_eq!(fired[0].0, 1);
461
462 assert_eq!(world.resource::<Deadlines<TestEvent>>().len(), 1);
464 }
465
466 #[test]
467 fn world_cancel_deadline() {
468 let mut world = crate::world::World::new();
469 world.add_deadline_type::<TestEvent>();
470
471 let id = world.schedule_deadline(Timestamp::from_secs(100), TestEvent(1));
472 assert!(world.cancel_deadline::<TestEvent>(id));
473 assert!(world.resource::<Deadlines<TestEvent>>().is_empty());
474 }
475
476 #[test]
477 fn world_batch_reconciliation() {
478 let mut world = crate::world::World::new();
479 world.add_deadline_type::<TestEvent>();
480
481 world.schedule_deadline(Timestamp::from_secs(100), TestEvent(1));
483 world.schedule_deadline(Timestamp::from_secs(200), TestEvent(2));
484 world.schedule_deadline(Timestamp::from_secs(300), TestEvent(3));
485
486 world.drain_deadlines::<TestEvent>(Timestamp::from_secs(1000));
488 world.update_events();
489
490 let events = world.resource::<crate::event::Events<TestEvent>>();
491 assert_eq!(events.len(), 3);
492 let values: Vec<u32> = events.read().map(|e| e.0).collect();
493 assert_eq!(values, vec![1, 2, 3]);
494 }
495
496 #[test]
497 fn world_drain_with_test_clock() {
498 let mut world = crate::world::World::new();
499 world.add_deadline_type::<TestEvent>();
500
501 let mut clock = TestClock::new(Timestamp::from_secs(0));
502
503 world.schedule_deadline(Timestamp::from_secs(60), TestEvent(1));
504 world.schedule_deadline(Timestamp::from_secs(120), TestEvent(2));
505
506 world.drain_deadlines::<TestEvent>(clock.now());
508 world.update_events();
509 assert!(
510 world
511 .resource::<crate::event::Events<TestEvent>>()
512 .is_empty()
513 );
514
515 clock.advance_secs(60);
517 world.drain_deadlines::<TestEvent>(clock.now());
518 world.update_events();
519 assert_eq!(world.resource::<crate::event::Events<TestEvent>>().len(), 1);
520
521 clock.advance_secs(60);
523 world.drain_deadlines::<TestEvent>(clock.now());
524 world.update_events();
525 assert_eq!(world.resource::<crate::event::Events<TestEvent>>().len(), 1);
526
527 assert!(world.resource::<Deadlines<TestEvent>>().is_empty());
529 }
530
531 #[test]
534 fn drain_all_deadlines_with_clock_resource() {
535 let mut world = crate::world::World::new();
536 world.add_deadline_type::<TestEvent>();
537 world
538 .insert_resource(Box::new(TestClock::new(Timestamp::from_secs(150))) as Box<dyn Clock>);
539
540 world.schedule_deadline(Timestamp::from_secs(100), TestEvent(1));
541 world.schedule_deadline(Timestamp::from_secs(200), TestEvent(2));
542
543 world.drain_all_deadlines();
545
546 world.update_events();
548
549 let events = world.resource::<crate::event::Events<TestEvent>>();
550 assert_eq!(events.len(), 1);
551 assert_eq!(events.read().next().unwrap().0, 1);
552 }
553
554 #[test]
555 fn drain_all_deadlines_no_clock_is_noop() {
556 let mut world = crate::world::World::new();
557 world.add_deadline_type::<TestEvent>();
558 world.schedule_deadline(Timestamp::from_secs(100), TestEvent(1));
559
560 world.drain_all_deadlines();
562 assert_eq!(world.resource::<Deadlines<TestEvent>>().len(), 1);
563 }
564
565 #[test]
568 fn schedule_run_fires_deadlines_readable_same_tick() {
569 use crate::schedule::Schedule;
570
571 let mut world = crate::world::World::new();
572 world.add_deadline_type::<TestEvent>();
573 world
574 .insert_resource(Box::new(TestClock::new(Timestamp::from_secs(200))) as Box<dyn Clock>);
575
576 world.schedule_deadline(Timestamp::from_secs(100), TestEvent(42));
578
579 world.insert_resource(0_u32); fn count_fired(
582 reader: crate::event::EventReader<'_, TestEvent>,
583 mut counter: crate::system_param::ResMut<'_, u32>,
584 ) {
585 for _ in reader.read() {
586 *counter += 1;
587 }
588 }
589
590 let mut schedule = Schedule::new();
591 schedule.add_system::<(
592 crate::event::EventReader<'_, TestEvent>,
593 crate::system_param::ResMut<'_, u32>,
594 )>("update", "count_fired", count_fired);
595
596 schedule.run(&mut world);
598
599 assert_eq!(*world.resource::<u32>(), 1);
601 assert!(world.resource::<Deadlines<TestEvent>>().is_empty());
603 }
604
605 #[test]
606 fn schedule_run_multiple_deadline_types() {
607 use crate::schedule::Schedule;
608
609 #[derive(Debug, PartialEq)]
610 struct OtherEvent(u32);
611
612 let mut world = crate::world::World::new();
613 world.add_deadline_type::<TestEvent>();
614 world.add_deadline_type::<OtherEvent>();
615 world
616 .insert_resource(Box::new(TestClock::new(Timestamp::from_secs(500))) as Box<dyn Clock>);
617
618 world.schedule_deadline(Timestamp::from_secs(100), TestEvent(1));
619 world.schedule_deadline(Timestamp::from_secs(200), OtherEvent(2));
620
621 let mut schedule = Schedule::new();
623 schedule.run(&mut world);
624
625 assert!(world.resource::<Deadlines<TestEvent>>().is_empty());
627 assert!(world.resource::<Deadlines<OtherEvent>>().is_empty());
628
629 assert_eq!(world.resource::<crate::event::Events<TestEvent>>().len(), 1);
631 assert_eq!(
632 world.resource::<crate::event::Events<OtherEvent>>().len(),
633 1
634 );
635 }
636}