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
use std::collections::{HashMap, VecDeque};
use crate::message_bus::{Envelope, NoOpHook, PublishHook, Subscriber};
pub enum SimulatorEvent {
Envelope(Envelope, std::time::SystemTime),
Tick(std::time::SystemTime),
}
pub struct Simulator<H: PublishHook = NoOpHook> {
subscribers: HashMap<String, Box<dyn Subscriber>>,
events: Vec<VecDeque<SimulatorEvent>>,
time: std::time::SystemTime,
hook: H,
}
impl Simulator<NoOpHook> {
/// Creates a new simulator with the given subscribers, initial time, and initial events.
///
/// The number of queues is determined by the length of the initial_events vector,
/// and this must match the number of queues in a [crate::message_bus::MessageBus] to accurately simulate
/// the message bus.
pub fn new(
subscribers: HashMap<String, Box<dyn Subscriber>>,
initial_time: std::time::SystemTime,
initial_events: Vec<Vec<SimulatorEvent>>,
) -> Self {
Self::with_hook(subscribers, initial_time, initial_events, NoOpHook)
}
}
impl<H: PublishHook> Simulator<H> {
/// Creates a new simulator with the given subscribers, initial time, initial events, and publish hook.
///
/// The number of queues is determined by the length of the initial_events vector,
/// and this must match the number of queues in a [crate::message_bus::MessageBus] to accurately simulate
/// the message bus.
pub fn with_hook(
subscribers: HashMap<String, Box<dyn Subscriber>>,
initial_time: std::time::SystemTime,
initial_events: Vec<Vec<SimulatorEvent>>,
hook: H,
) -> Self {
let mut events: Vec<VecDeque<SimulatorEvent>> = initial_events
.into_iter()
.map(|events| events.into())
.collect();
// Ensure at least one priority queue exists
if events.is_empty() {
events.push(VecDeque::new());
}
Self {
subscribers,
events,
time: initial_time,
hook,
}
}
/// Steps the simluator by some duration, looping through all of the subscribers to
/// process events from the queue, then run their tick.
///
/// Returns the new time after the step.
pub fn step(&mut self, step_by: std::time::Duration) -> std::time::SystemTime {
let subscribers = &mut self.subscribers;
let events = std::mem::take(&mut self.events); // we are replacing this later anyway
let num_queues = events.len();
let mut new_events: Vec<VecDeque<SimulatorEvent>> =
(0..num_queues).map(|_| VecDeque::new()).collect();
// First we increment the time to simulate the passing of time
self.time += step_by;
// Then we process all of the events in the queue, in decreasing priority order (highest first)
for queue in events.into_iter().rev() {
for event in queue {
match event {
SimulatorEvent::Envelope(envelope, at) => {
let subscriber = subscribers.get_mut(&envelope.destination).unwrap();
let envelopes = subscriber.receive(envelope.message, at);
// Add any new envelopes to the appropriate priority queue
for envelope in envelopes {
self.hook.on_publish(&envelope, at);
let priority = envelope.priority.min(num_queues - 1);
new_events[priority].push_back(SimulatorEvent::Envelope(envelope, at));
}
}
SimulatorEvent::Tick(at) => {
for subscriber in subscribers.values_mut() {
let envelopes = subscriber.tick(at);
// Add any new envelopes to the appropriate priority queue
for envelope in envelopes {
self.hook.on_publish(&envelope, at);
let priority = envelope.priority.min(num_queues - 1);
new_events[priority]
.push_back(SimulatorEvent::Envelope(envelope, at));
}
}
}
}
}
}
// Finally we process all of the ticks at the new time
for subscriber in subscribers.values_mut() {
let envelopes = subscriber.tick(self.time);
for envelope in envelopes {
self.hook.on_publish(&envelope, self.time);
let priority = envelope.priority.min(num_queues - 1);
new_events[priority].push_back(SimulatorEvent::Envelope(envelope, self.time));
}
}
// Reset the events queue
self.events = new_events;
self.time
}
pub fn step_to(
&mut self,
time: std::time::SystemTime,
step_by: std::time::Duration,
) -> std::time::SystemTime {
// Keep stepping until we reach the target time
while self.time < time {
// Calculate the remaining time to reach the target
let remaining_time = time
.duration_since(self.time)
.unwrap_or(std::time::Duration::ZERO);
// Step by the minimum of step_by and remaining time
let actual_step = step_by.min(remaining_time);
// If no time remains, break out
if actual_step.is_zero() {
break;
}
self.step(actual_step);
}
self.time
}
}