1
//! One task owns the peer, playback clock, LED, counters, and packet buffers.
2
use crate::{
3
    error::{Error, Result},
4
    platform::{self, Board, MusicBytes, Peer, PeerState},
5
};
6
use radio_core::{
7
    music::{FRAME_MS, Pack},
8
    playback::Playback,
9
    protocol::{self, Ack, Color, Command, ROBOT_ID, SPECTRUM_ID, Telemetry},
10
};
11
use std::{
12
    sync::mpsc::{Receiver, SyncSender},
13
    thread,
14
    time::Duration,
15
};
16
 
17
pub(crate) enum Request {
18
    Answer(String, SyncSender<Result<()>>),
19
    Start(SyncSender<Result<()>>),
20
}
21
pub(crate) enum Event {
22
    Offer(String),
23
    Connected,
24
    Lost,
25
}
26
 
27
// SDP-bearing messages intentionally have no Debug implementation.
28
 
29
#[derive(Debug, Default)]
30
struct Counters {
31
    audio: u64,
32
    data: u64,
33
    rejected: u64,
34
    sequence: u64,
35
}
36
 
37
pub(crate) fn run(
38
    mut board: Board,
39
    requests: Receiver<Request>,
40
    events: SyncSender<Event>,
41
) -> Result<()> {
42
    let storage = MusicBytes::load()?;
43
    let pack =
44
        Pack::parse(storage.as_slice()).map_err(|_| Error::new("music pack validation failed"))?;
45
    platform::log(&format!(
46
        "Music validated: frames={} duration_ms={}",
47
        pack.frame_count(),
48
        pack.frame_count().get() * FRAME_MS
49
    ));
50
    let mut peer = Peer::open()?;
51
    let mut playback = None;
52
    let mut led = Color([0, 12, 4]);
53
    let mut counters = Counters::default();
54
    let mut next_telemetry = 0;
55
    let mut previous = PeerState::Connecting;
56
    let mut offer_buffer = vec![0; 16_000];
57
    let mut offer_sent = false;
58
    let mut command_buffer = [0; protocol::COMMAND_LIMIT];
59
    let mut json_buffer = Vec::with_capacity(768);
60
    loop {
61
        if let Ok(request) = requests.try_recv() {
62
            match request {
63
                Request::Answer(sdp, reply) => {
64
                    let _ = reply.send(peer.answer(&sdp));
65
                }
66
                Request::Start(reply) => {
67
                    let result = if playback.is_some() {
68
                        Err(Error::new("playback already started"))
69
                    } else {
70
                        peer.create_channels()
71
                    };
72
                    if result.is_ok() {
73
                        playback = Some(Playback::new(pack.frame_count(), platform::now_us()));
74
                    }
75
                    let _ = reply.send(result);
76
                }
77
            }
78
        }
79
        peer.poll();
80
        if !offer_sent {
81
            let length = peer.offer(&mut offer_buffer)?;
82
            if length > 0 {
83
                let offer = std::str::from_utf8(&offer_buffer[..length])
84
                    .map_err(|_| Error::new("invalid local SDP"))?;
85
                events
86
                    .try_send(Event::Offer(offer.to_owned()))
87
                    .map_err(|_| Error::new("signaling event queue unavailable"))?;
88
                offer_sent = true;
89
                offer_buffer.fill(0);
90
                // Release setup-only storage before the steady-state packet loop.
91
                offer_buffer = Vec::new();
92
            }
93
        }
94
        let state = peer.state();
95
        if state != previous {
96
            if state == PeerState::Connected {
97
                events
98
                    .try_send(Event::Connected)
99
                    .map_err(|_| Error::new("signaling event queue unavailable"))?;
100
                platform::log("WebRTC audio and SCTP connected");
101
            } else if state == PeerState::Lost {
102
                let _ = events.try_send(Event::Lost);
103
                return Err(Error::new("WebRTC transport disconnected"));
104
            }
105
            previous = state;
106
        }
107
        if let Some(player) = playback.as_mut().filter(|_| state == PeerState::Connected) {
108
            // Bound per-tick work even if a controller floods the reliable stream.
109
            let length = peer.command(&mut command_buffer)?;
110
            if length > 0 {
111
                match Command::parse(&command_buffer[..length]) {
112
                    Ok(command) => {
113
                        let mut result = 0;
114
                        if let Some(color) = command.led {
115
                            if board.set_led(color).is_ok() {
116
                                led = color;
117
                            } else {
118
                                result = -1;
119
                            }
120
                        }
121
                        if let Some(action) = command.action {
122
                            player.apply(action);
123
                        }
124
                        let ack = Ack {
125
                            event: "ack",
126
                            command_id: command.command_id.as_deref(),
127
                            result,
128
                            led,
129
                            paused: player.paused(),
130
                        };
131
                        send_json(&mut peer, &ack, &mut json_buffer, &mut counters)?;
132
                    }
133
                    Err(_) => counters.rejected = counters.rejected.saturating_add(1),
134
                }
135
            }
136
            let now = platform::now_us();
137
            if let Some(due) = player.due(now) {
138
                let frame = pack
139
                    .frame(due.index)
140
                    .ok_or(Error::new("playback index out of range"))?;
141
                let opus = if due.paused {
142
                    protocol::SILENCE
143
                } else {
144
                    frame.opus
145
                };
146
                if peer.audio(due.pts_ms, opus).is_err() {
147
                    counters.audio = counters.audio.saturating_add(1);
148
                }
149
                if due.spectrum
150
                    && peer
151
                        .data(SPECTRUM_ID, true, &protocol::spectrum(due, frame.bands))
152
                        .is_err()
153
                {
154
                    counters.data = counters.data.saturating_add(1);
155
                }
156
            }
157
            if now >= next_telemetry {
158
                next_telemetry = now + 500_000;
159
                counters.sequence = counters.sequence.wrapping_add(1);
160
                let telemetry = Telemetry {
161
                    event: "telemetry",
162
                    firmware: "rust",
163
                    sequence: counters.sequence,
164
                    random: platform::random() % 1000,
165
                    uptime_ms: now / 1000,
166
                    position_ms: player.position_ms(),
167
                    duration_ms: player.duration_ms(),
168
                    paused: player.paused(),
169
                    rssi: platform::rssi(),
170
                    heap: platform::heap_free(),
171
                    audio_errors: counters.audio,
172
                    data_errors: counters.data,
173
                    skipped_frames: player.skipped_frames(),
174
                    rejected_commands: counters.rejected,
175
                    dropped_commands: peer.dropped(),
176
                    stack_free: platform::stack_free(),
177
                    led,
178
                };
179
                send_json(&mut peer, &telemetry, &mut json_buffer, &mut counters)?;
180
            }
181
        }
182
        thread::sleep(Duration::from_millis(1));
183
    }
184
}
185
 
186
fn send_json(
187
    peer: &mut Peer,
188
    value: &impl serde::Serialize,
189
    buffer: &mut Vec<u8>,
190
    counters: &mut Counters,
191
) -> Result<()> {
192
    buffer.clear();
193
    serde_json::to_writer(&mut *buffer, value)
194
        .map_err(|_| Error::new("packet serialization failed"))?;
195
    if peer.data(ROBOT_ID, false, buffer).is_err() {
196
        counters.data = counters.data.saturating_add(1);
197
    }
198
    Ok(())
199
}