gled_core/engine/
outbox.rs

1//! Turns a client's engine states into the messages a remote interface gets:
2//! all assets once, then their changes; the first state whole, then with the
3//! project patched; the timing alone when nothing else changed; preview and
4//! sound only when subscribed. The server's WebSocket sessions and the
5//! desktop's in-process link both use it, so every interface sees the same
6//! messages.
7
8use std::sync::Arc;
9
10use gled_model::{collection::Collections as AssetCollections, timing::BeatState};
11use gled_protocol::{
12    assets::{AssetChanges, all_assets},
13    bytes::SharedBytes,
14    message::{ServerMessage, Subscriptions},
15    project_update::ProjectUpdate,
16};
17
18use super::{ClientState, EngineState, EventLog, RenderStats, SoundSnapshot};
19
20/// What was sent last, per client.
21#[derive(Debug, Default)]
22pub struct Outbox {
23    subscriptions: Subscriptions,
24    preview: Option<Arc<Vec<u8>>>,
25    sound: Option<Arc<SoundSnapshot>>,
26    assets: Option<Arc<AssetCollections>>,
27    state: Option<Arc<EngineState>>,
28    applied_sequence: u64,
29    timing: Option<(BeatState, RenderStats)>,
30    /// The number of the next input event to send; `None` until the first
31    /// state after subscribing, which starts from the events to come.
32    input_events_from: Option<u64>,
33    midi_from: Option<u64>,
34    learn_from: Option<u64>,
35}
36
37impl Outbox {
38    /// Changes which optional streams are sent.
39    pub fn subscribe(&mut self, subscriptions: Subscriptions) {
40        self.subscriptions = subscriptions;
41        if !subscriptions.preview {
42            self.preview = None;
43        }
44        if !subscriptions.sound {
45            self.sound = None;
46        }
47        if !subscriptions.input_events {
48            self.input_events_from = None;
49        }
50        if subscriptions.midi {
51            // The monitor's history matters: ports connect once, at startup.
52            self.midi_from.get_or_insert(0);
53        } else {
54            self.midi_from = None;
55            self.learn_from = None;
56        }
57    }
58
59    /// The messages that bring the client from what it was sent last to
60    /// `state`, in the order to send them.
61    pub fn messages(&mut self, state: ClientState) -> Vec<ServerMessage> {
62        let mut messages = Vec::new();
63        if self.subscriptions.preview
64            && let Some(preview) = &state.preview
65            && !self
66                .preview
67                .as_ref()
68                .is_some_and(|previous| Arc::ptr_eq(previous, preview))
69        {
70            messages.push(ServerMessage::Preview(SharedBytes(preview.clone())));
71            self.preview = Some(preview.clone());
72        }
73        if self.subscriptions.sound
74            && !self
75                .sound
76                .as_ref()
77                .is_some_and(|previous| Arc::ptr_eq(previous, &state.sound))
78        {
79            messages.push(ServerMessage::Sound(state.sound.clone()));
80            self.sound = Some(state.sound.clone());
81        }
82
83        if self.subscriptions.input_events
84            && let Some(events) = new_events(&state.input_log, &mut self.input_events_from)
85        {
86            messages.push(ServerMessage::InputEvents(events));
87        }
88        if self.subscriptions.midi {
89            if let Some(events) = new_events(&state.midi_log, &mut self.midi_from) {
90                messages.push(ServerMessage::MidiMonitor(events));
91            }
92            if let Some(events) = new_events(&state.learn_log, &mut self.learn_from) {
93                messages.push(ServerMessage::MidiLearn(events));
94            }
95        }
96
97        // Assets first: the state refers to them.
98        match &self.assets {
99            None => messages.push(ServerMessage::Assets(all_assets(&state.assets))),
100            Some(previous) if !Arc::ptr_eq(previous, &state.assets) => {
101                let changes = AssetChanges::between(previous, &state.assets);
102                if !changes.is_empty() {
103                    messages.push(ServerMessage::AssetsChanged(changes));
104                }
105            }
106            Some(_) => {}
107        }
108        self.assets = Some(state.assets.clone());
109
110        // The project is compared once, by the update that carries it; the
111        // engine's unchanged project is the same value and costs nothing.
112        let project = self.state.as_ref().map(|previous| {
113            ProjectUpdate::between(previous.project.as_deref(), state.shared.project.as_deref())
114        });
115        let changed = self.applied_sequence != state.applied_sequence
116            || match (&self.state, &project) {
117                (Some(previous), Some(project)) => {
118                    !previous.same_but_timing_and_project(&state.shared)
119                        || !matches!(project, ProjectUpdate::Unchanged)
120                }
121                _ => true,
122            };
123        let timing = (state.shared.beat, state.shared.render);
124        if changed {
125            // The first state goes whole, later ones with the project patched.
126            messages.push(match project {
127                None => ServerMessage::State {
128                    state: state.shared.clone(),
129                    applied_sequence: state.applied_sequence,
130                },
131                Some(project) => ServerMessage::Update {
132                    state: Arc::new(state.shared.without_project()),
133                    project,
134                    applied_sequence: state.applied_sequence,
135                },
136            });
137            self.state = Some(state.shared);
138            self.applied_sequence = state.applied_sequence;
139            self.timing = Some(timing);
140        } else if self.timing != Some(timing) {
141            let (beat, render) = timing;
142            messages.push(ServerMessage::Timing { beat, render });
143            self.timing = Some(timing);
144        }
145        messages
146    }
147}
148
149/// The events of `log` from `from` on, and moves `from` past them. Without a
150/// `from` yet, this only sets it: what happened before is not sent.
151fn new_events<T: Clone>(log: &EventLog<T>, from: &mut Option<u64>) -> Option<Vec<T>> {
152    let events = from.map(|from| log.since(from));
153    *from = Some(log.next());
154    events.filter(|events| !events.is_empty())
155}
156
157#[cfg(test)]
158mod tests {
159    use super::*;
160    use crate::{
161        engine::{InputLog, MidiLog},
162        input::key_code::KeyCode,
163    };
164    use gled_model::input::event::InputEvent;
165    use gled_protocol::midi::MidiMonitorEvent;
166
167    fn client_state(log: &InputLog) -> ClientState {
168        ClientState {
169            shared: Arc::default(),
170            applied_sequence: 0,
171            assets: Arc::default(),
172            preview: None,
173            sound: Arc::default(),
174            input_log: Arc::new(log.clone()),
175            midi_log: Arc::default(),
176            learn_log: Arc::default(),
177        }
178    }
179
180    fn input_events(messages: &[ServerMessage]) -> Vec<InputEvent> {
181        messages
182            .iter()
183            .filter_map(|message| match message {
184                ServerMessage::InputEvents(events) => Some(events.clone()),
185                _ => None,
186            })
187            .flatten()
188            .collect()
189    }
190
191    #[test]
192    fn input_events_start_with_the_subscription_and_none_is_lost() {
193        let mut log = InputLog::default();
194        log.push(InputEvent::Key(KeyCode::A));
195        let mut outbox = Outbox::default();
196        outbox.subscribe(Subscriptions {
197            input_events: true,
198            ..Subscriptions::default()
199        });
200        // What happened before subscribing is not sent.
201        assert!(input_events(&outbox.messages(client_state(&log))).is_empty());
202
203        // Several ticks between two updates: all their events arrive.
204        log.push(InputEvent::Key(KeyCode::B));
205        log.push(InputEvent::Key(KeyCode::C));
206        assert_eq!(
207            input_events(&outbox.messages(client_state(&log))),
208            vec![InputEvent::Key(KeyCode::B), InputEvent::Key(KeyCode::C)]
209        );
210        assert!(input_events(&outbox.messages(client_state(&log))).is_empty());
211    }
212
213    #[test]
214    fn the_midi_monitor_starts_with_the_kept_history() {
215        let mut midi = MidiLog::default();
216        midi.push(MidiMonitorEvent {
217            timestamp_ms: 1,
218            port_name: "APC40".to_owned(),
219            label: "input connected".to_owned(),
220            bytes: Vec::new(),
221        });
222        let mut state = client_state(&InputLog::default());
223        state.midi_log = Arc::new(midi.clone());
224        let mut outbox = Outbox::default();
225        outbox.subscribe(Subscriptions {
226            midi: true,
227            ..Subscriptions::default()
228        });
229        let messages = outbox.messages(state);
230        assert!(messages.iter().any(
231            |message| matches!(message, ServerMessage::MidiMonitor(events) if events.len() == 1)
232        ));
233    }
234}
gled_core/engine/outbox.rs

gled_core/engine/
outbox.rs
