gled-core 2.31.2

//! The OSC network loop: receives messages, keeps the subscribers of the state
//! feedback and sends them the engine's state.

mod scene;
mod snapshot;
#[cfg(test)]
mod tests;

use rosc::{OscMessage, OscPacket, OscType};
use std::collections::HashMap;
use std::net::{SocketAddr, UdpSocket};
use std::sync::Arc;
use std::sync::atomic::Ordering;
use std::time::{Duration, Instant};
use std::{io, thread};
use strum::EnumString;

use tracing::{debug, trace, warn};

use crate::action::UiAction;

use super::parse::parse_message;
use super::{OSCHandler, OscStateSnapshot};
use snapshot::broadcast_snapshot;

static SUBSCRIBER_TTL: Duration = Duration::from_secs(30);

/// An OSC client asking for state feedback, parsed from the message address.
#[derive(Debug, Clone, Copy, PartialEq, Eq, EnumString)]
enum ClientInterestCommand {
    #[strum(serialize = "/osc/state/subscribe")]
    Subscribe,
    #[strum(serialize = "/osc/state/unsubscribe")]
    Unsubscribe,
    #[strum(serialize = "/osc/state/heartbeat")]
    Heartbeat,
}

pub(super) fn start_network_loop(handler: Arc<OSCHandler>) {
    thread::Builder::new()
        .name("gled:artnet:osc".to_string())
        .spawn(move || {
            let mut buf = [0u8; rosc::decoder::MTU];
            let mut subscribers: HashMap<SocketAddr, Instant> = HashMap::new();
            trace!("Starting osc network receiving loop");

            let state_receiver = crate::output_state::new_receiver();

            while handler.running.load(Ordering::Relaxed) {
                subscribers.retain(|_, last_seen| last_seen.elapsed() < SUBSCRIBER_TTL);

                let mut latest_snapshot = None;
                while let Ok(Some(state)) = state_receiver.try_recv() {
                    latest_snapshot = Some(OscStateSnapshot {
                        project: state.project,
                        selected_scene_instance: state.selected_scene_instance,
                        blackout: state.blackout,
                        beats_per_minute: state.beats_per_minute,
                        beat_progression: state.beat_progression,
                    });
                }
                if let Some(snapshot) = latest_snapshot
                    && !subscribers.is_empty()
                {
                    broadcast_snapshot(&handler.socket, &subscribers, &snapshot);
                }

                trace!("Waiting for osc package...");
                let (size, src) = match handler.socket.recv_from(&mut buf) {
                    Ok(received) => received,
                    Err(err)
                        if matches!(
                            err.kind(),
                            io::ErrorKind::WouldBlock | io::ErrorKind::TimedOut
                        ) =>
                    {
                        continue;
                    }
                    Err(err) => {
                        debug!("Could not receive osc package ({err:?}). Retrying in 10ms...");
                        thread::sleep(Duration::from_millis(10));
                        continue;
                    }
                };

                crate::network_stats::add_incoming_bytes(size);

                trace!("Received packet with size {} from: {}", size, src);
                let Ok((_, packet)) = rosc::decoder::decode_udp(&buf[..size]) else {
                    warn!("Invalid osc packet from {src}");
                    continue;
                };

                match packet {
                    OscPacket::Message(msg) => {
                        trace!("OSC address: {}", msg.addr);
                        trace!("OSC arguments: {:?}", msg.args);
                        if let Ok(command) = msg.addr.parse::<ClientInterestCommand>() {
                            match command {
                                ClientInterestCommand::Subscribe => {
                                    subscribers.insert(src, Instant::now());
                                }
                                ClientInterestCommand::Unsubscribe => {
                                    subscribers.remove(&src);
                                }
                                ClientInterestCommand::Heartbeat => {
                                    subscribers.entry(src).and_modify(|t| *t = Instant::now());
                                }
                            }
                            continue;
                        }

                        if let Ok(action) = parse_message(&msg) {
                            UiAction::enqueue(action);
                        } else {
                            warn!("Invalid osc message: {} with args {:?}", msg.addr, msg.args);
                        }
                    }
                    OscPacket::Bundle(bundle) => {
                        warn!("OSC Bundle are not supported currently: {:?}", bundle);
                    }
                }
            }
        })
        .expect("Could not spawn osc receiver thread");
}

/// A count or an index as an OSC int, saturating at `i32::MAX`.
fn osc_int(value: usize) -> OscType {
    OscType::Int(i32::try_from(value).unwrap_or(i32::MAX))
}

fn send_feedback(socket: &UdpSocket, addr: SocketAddr, path: &str, args: Vec<OscType>) {
    let packet = OscPacket::Message(OscMessage {
        addr: path.to_string(),
        args,
    });
    let Ok(bytes) = rosc::encoder::encode(&packet) else {
        return;
    };
    match socket.send_to(&bytes, addr) {
        Ok(count) => {
            crate::network_stats::add_outgoing_bytes(count);
        }
        Err(err) => {
            debug!("Could not send OSC feedback to {addr}: {err:?}");
        }
    }
}