gled-core 2.31.2

//! Turns a client's engine states into the messages a remote interface gets:
//! all assets once, then their changes; the first state whole, then with the
//! project patched; the timing alone when nothing else changed; preview and
//! sound only when subscribed. The server's WebSocket sessions and the
//! desktop's in-process link both use it, so every interface sees the same
//! messages.

use std::sync::Arc;

use gled_model::{collection::Collections as AssetCollections, timing::BeatState};
use gled_protocol::{
    assets::{AssetChanges, all_assets},
    bytes::SharedBytes,
    message::{ServerMessage, Subscriptions},
    project_update::ProjectUpdate,
};

use super::{ClientState, EngineState, EventLog, RenderStats, SoundSnapshot};

/// What was sent last, per client.
#[derive(Debug, Default)]
pub struct Outbox {
    subscriptions: Subscriptions,
    preview: Option<Arc<Vec<u8>>>,
    sound: Option<Arc<SoundSnapshot>>,
    assets: Option<Arc<AssetCollections>>,
    state: Option<Arc<EngineState>>,
    applied_sequence: u64,
    timing: Option<(BeatState, RenderStats)>,
    /// The number of the next input event to send; `None` until the first
    /// state after subscribing, which starts from the events to come.
    input_events_from: Option<u64>,
    midi_from: Option<u64>,
    learn_from: Option<u64>,
}

impl Outbox {
    /// Changes which optional streams are sent.
    pub fn subscribe(&mut self, subscriptions: Subscriptions) {
        self.subscriptions = subscriptions;
        if !subscriptions.preview {
            self.preview = None;
        }
        if !subscriptions.sound {
            self.sound = None;
        }
        if !subscriptions.input_events {
            self.input_events_from = None;
        }
        if subscriptions.midi {
            // The monitor's history matters: ports connect once, at startup.
            self.midi_from.get_or_insert(0);
        } else {
            self.midi_from = None;
            self.learn_from = None;
        }
    }

    /// The messages that bring the client from what it was sent last to
    /// `state`, in the order to send them.
    pub fn messages(&mut self, state: ClientState) -> Vec<ServerMessage> {
        let mut messages = Vec::new();
        if self.subscriptions.preview
            && let Some(preview) = &state.preview
            && !self
                .preview
                .as_ref()
                .is_some_and(|previous| Arc::ptr_eq(previous, preview))
        {
            messages.push(ServerMessage::Preview(SharedBytes(preview.clone())));
            self.preview = Some(preview.clone());
        }
        if self.subscriptions.sound
            && !self
                .sound
                .as_ref()
                .is_some_and(|previous| Arc::ptr_eq(previous, &state.sound))
        {
            messages.push(ServerMessage::Sound(state.sound.clone()));
            self.sound = Some(state.sound.clone());
        }

        if self.subscriptions.input_events
            && let Some(events) = new_events(&state.input_log, &mut self.input_events_from)
        {
            messages.push(ServerMessage::InputEvents(events));
        }
        if self.subscriptions.midi {
            if let Some(events) = new_events(&state.midi_log, &mut self.midi_from) {
                messages.push(ServerMessage::MidiMonitor(events));
            }
            if let Some(events) = new_events(&state.learn_log, &mut self.learn_from) {
                messages.push(ServerMessage::MidiLearn(events));
            }
        }

        // Assets first: the state refers to them.
        match &self.assets {
            None => messages.push(ServerMessage::Assets(all_assets(&state.assets))),
            Some(previous) if !Arc::ptr_eq(previous, &state.assets) => {
                let changes = AssetChanges::between(previous, &state.assets);
                if !changes.is_empty() {
                    messages.push(ServerMessage::AssetsChanged(changes));
                }
            }
            Some(_) => {}
        }
        self.assets = Some(state.assets.clone());

        // The project is compared once, by the update that carries it; the
        // engine's unchanged project is the same value and costs nothing.
        let project = self.state.as_ref().map(|previous| {
            ProjectUpdate::between(previous.project.as_deref(), state.shared.project.as_deref())
        });
        let changed = self.applied_sequence != state.applied_sequence
            || match (&self.state, &project) {
                (Some(previous), Some(project)) => {
                    !previous.same_but_timing_and_project(&state.shared)
                        || !matches!(project, ProjectUpdate::Unchanged)
                }
                _ => true,
            };
        let timing = (state.shared.beat, state.shared.render);
        if changed {
            // The first state goes whole, later ones with the project patched.
            messages.push(match project {
                None => ServerMessage::State {
                    state: state.shared.clone(),
                    applied_sequence: state.applied_sequence,
                },
                Some(project) => ServerMessage::Update {
                    state: Arc::new(state.shared.without_project()),
                    project,
                    applied_sequence: state.applied_sequence,
                },
            });
            self.state = Some(state.shared);
            self.applied_sequence = state.applied_sequence;
            self.timing = Some(timing);
        } else if self.timing != Some(timing) {
            let (beat, render) = timing;
            messages.push(ServerMessage::Timing { beat, render });
            self.timing = Some(timing);
        }
        messages
    }
}

/// The events of `log` from `from` on, and moves `from` past them. Without a
/// `from` yet, this only sets it: what happened before is not sent.
fn new_events<T: Clone>(log: &EventLog<T>, from: &mut Option<u64>) -> Option<Vec<T>> {
    let events = from.map(|from| log.since(from));
    *from = Some(log.next());
    events.filter(|events| !events.is_empty())
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::{
        engine::{InputLog, MidiLog},
        input::key_code::KeyCode,
    };
    use gled_model::input::event::InputEvent;
    use gled_protocol::midi::MidiMonitorEvent;

    fn client_state(log: &InputLog) -> ClientState {
        ClientState {
            shared: Arc::default(),
            applied_sequence: 0,
            assets: Arc::default(),
            preview: None,
            sound: Arc::default(),
            input_log: Arc::new(log.clone()),
            midi_log: Arc::default(),
            learn_log: Arc::default(),
        }
    }

    fn input_events(messages: &[ServerMessage]) -> Vec<InputEvent> {
        messages
            .iter()
            .filter_map(|message| match message {
                ServerMessage::InputEvents(events) => Some(events.clone()),
                _ => None,
            })
            .flatten()
            .collect()
    }

    #[test]
    fn input_events_start_with_the_subscription_and_none_is_lost() {
        let mut log = InputLog::default();
        log.push(InputEvent::Key(KeyCode::A));
        let mut outbox = Outbox::default();
        outbox.subscribe(Subscriptions {
            input_events: true,
            ..Subscriptions::default()
        });
        // What happened before subscribing is not sent.
        assert!(input_events(&outbox.messages(client_state(&log))).is_empty());

        // Several ticks between two updates: all their events arrive.
        log.push(InputEvent::Key(KeyCode::B));
        log.push(InputEvent::Key(KeyCode::C));
        assert_eq!(
            input_events(&outbox.messages(client_state(&log))),
            vec![InputEvent::Key(KeyCode::B), InputEvent::Key(KeyCode::C)]
        );
        assert!(input_events(&outbox.messages(client_state(&log))).is_empty());
    }

    #[test]
    fn the_midi_monitor_starts_with_the_kept_history() {
        let mut midi = MidiLog::default();
        midi.push(MidiMonitorEvent {
            timestamp_ms: 1,
            port_name: "APC40".to_owned(),
            label: "input connected".to_owned(),
            bytes: Vec::new(),
        });
        let mut state = client_state(&InputLog::default());
        state.midi_log = Arc::new(midi.clone());
        let mut outbox = Outbox::default();
        outbox.subscribe(Subscriptions {
            midi: true,
            ..Subscriptions::default()
        });
        let messages = outbox.messages(state);
        assert!(messages.iter().any(
            |message| matches!(message, ServerMessage::MidiMonitor(events) if events.len() == 1)
        ));
    }
}