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        let changed = self.applied_sequence != state.applied_sequence
111            || self
112                .state
113                .as_ref()
114                .is_none_or(|previous| !previous.same_but_timing(&state.shared));
115        let timing = (state.shared.beat, state.shared.render);
116        if changed {
117            // The first state goes whole, later ones with the project patched.
118            messages.push(match &self.state {
119                None => ServerMessage::State {
120                    state: state.shared.clone(),
121                    applied_sequence: state.applied_sequence,
122                },
123                Some(previous) => ServerMessage::Update {
124                    state: Arc::new(state.shared.without_project()),
125                    project: ProjectUpdate::between(
126                        previous.project.as_ref(),
127                        state.shared.project.as_ref(),
128                    ),
129                    applied_sequence: state.applied_sequence,
130                },
131            });
132            self.state = Some(state.shared);
133            self.applied_sequence = state.applied_sequence;
134            self.timing = Some(timing);
135        } else if self.timing != Some(timing) {
136            let (beat, render) = timing;
137            messages.push(ServerMessage::Timing { beat, render });
138            self.timing = Some(timing);
139        }
140        messages
141    }
142}
143
144/// The events of `log` from `from` on, and moves `from` past them. Without a
145/// `from` yet, this only sets it: what happened before is not sent.
146fn new_events<T: Clone>(log: &EventLog<T>, from: &mut Option<u64>) -> Option<Vec<T>> {
147    let events = from.map(|from| log.since(from));
148    *from = Some(log.next());
149    events.filter(|events| !events.is_empty())
150}
151
152#[cfg(test)]
153mod tests {
154    use super::*;
155    use crate::{
156        engine::{InputLog, MidiLog},
157        input::key_code::KeyCode,
158    };
159    use gled_model::input::event::InputEvent;
160    use gled_protocol::midi::MidiMonitorEvent;
161
162    fn client_state(log: &InputLog) -> ClientState {
163        ClientState {
164            shared: Arc::default(),
165            applied_sequence: 0,
166            assets: Arc::default(),
167            preview: None,
168            sound: Arc::default(),
169            input_log: Arc::new(log.clone()),
170            midi_log: Arc::default(),
171            learn_log: Arc::default(),
172        }
173    }
174
175    fn input_events(messages: &[ServerMessage]) -> Vec<InputEvent> {
176        messages
177            .iter()
178            .filter_map(|message| match message {
179                ServerMessage::InputEvents(events) => Some(events.clone()),
180                _ => None,
181            })
182            .flatten()
183            .collect()
184    }
185
186    #[test]
187    fn input_events_start_with_the_subscription_and_none_is_lost() {
188        let mut log = InputLog::default();
189        log.push(InputEvent::Key(KeyCode::A));
190        let mut outbox = Outbox::default();
191        outbox.subscribe(Subscriptions {
192            input_events: true,
193            ..Subscriptions::default()
194        });
195        // What happened before subscribing is not sent.
196        assert!(input_events(&outbox.messages(client_state(&log))).is_empty());
197
198        // Several ticks between two updates: all their events arrive.
199        log.push(InputEvent::Key(KeyCode::B));
200        log.push(InputEvent::Key(KeyCode::C));
201        assert_eq!(
202            input_events(&outbox.messages(client_state(&log))),
203            vec![InputEvent::Key(KeyCode::B), InputEvent::Key(KeyCode::C)]
204        );
205        assert!(input_events(&outbox.messages(client_state(&log))).is_empty());
206    }
207
208    #[test]
209    fn the_midi_monitor_starts_with_the_kept_history() {
210        let mut midi = MidiLog::default();
211        midi.push(MidiMonitorEvent {
212            timestamp_ms: 1,
213            port_name: "APC40".to_owned(),
214            label: "input connected".to_owned(),
215            bytes: Vec::new(),
216        });
217        let mut state = client_state(&InputLog::default());
218        state.midi_log = Arc::new(midi.clone());
219        let mut outbox = Outbox::default();
220        outbox.subscribe(Subscriptions {
221            midi: true,
222            ..Subscriptions::default()
223        });
224        let messages = outbox.messages(state);
225        assert!(messages.iter().any(
226            |message| matches!(message, ServerMessage::MidiMonitor(events) if events.len() == 1)
227        ));
228    }
229}
gled_core/engine/outbox.rs

gled_core/engine/
outbox.rs
