open_stomp/
queue.rs

1//! Request pacing (docs/03): the pedal handles one SysEx request at a time, so requests and
2//! writes are queued and each waits for its reply or ack before the next is sent, like the
3//! official app. Live CCs/PCs carry no reply and need no queue.
4//!
5//! [`RequestQueue`] is the pure state machine (no I/O, time passed in) that
6//! [`device`](crate::device) drives; use it directly to talk to the pedal over your own MIDI
7//! transport:
8//!
9//! ```
10//! use std::time::{Duration, Instant};
11//! use open_stomp::queue::{RequestQueue, handshake};
12//! use open_stomp::Msg;
13//!
14//! let mut q = RequestQueue::new(Duration::from_secs(1));
15//! q.start_sync(handshake());
16//! // loop: send what the queue releases ...
17//! while let Some(req) = q.pop() {
18//!     // out.send(req.bytes())?;
19//!     q.sent(req, Instant::now());
20//! }
21//! // ... and feed it every complete message from the pedal
22//! let reply = [0xF0, 0x00, 0x21, 0x7A, 0x10, 0x56, 0x31, 0xF7];
23//! assert!(q.on_message(&Msg::parse(&reply)).is_some());
24//! assert_eq!(q.progress(), (1, handshake().len()));
25//! ```
26
27use std::collections::VecDeque;
28use std::time::{Duration, Instant};
29
30use crate::midi::{Cmd, Msg, req};
31
32/// What answers a request.
33#[derive(Clone, Copy, Debug, PartialEq, Eq)]
34pub enum Expect {
35    /// Short frame reply of this kind with a payload (`0x10` version, `0x60` product).
36    Short(u8),
37    /// NEP reply (`dir` 2) to this cmd.
38    Reply(u8),
39    /// NEP write ack (`dir` 3) for this cmd.
40    Ack(u8),
41}
42
43impl Expect {
44    /// What the pedal sends back for an outgoing message, `None` for CCs/PCs (no reply).
45    pub fn of(bytes: &[u8]) -> Option<Expect> {
46        Some(match Msg::parse(bytes) {
47            Msg::Short { kind: 0x00, .. } => Expect::Short(0x10),
48            Msg::Short { kind, .. } => Expect::Short(kind),
49            Msg::Nep { cmd, dir: 1, .. } => Expect::Ack(cmd),
50            Msg::Nep { cmd, .. } => Expect::Reply(cmd),
51            _ => return None,
52        })
53    }
54
55    pub fn matches(self, msg: &Msg) -> bool {
56        match (self, msg) {
57            (Expect::Short(k), Msg::Short { kind, payload }) => *kind == k && !payload.is_empty(),
58            (Expect::Reply(c), Msg::Nep { cmd, dir, .. }) => *cmd == c && *dir == 2,
59            (Expect::Ack(c), Msg::Nep { cmd, dir, .. }) => *cmd == c && *dir == 3,
60            _ => false,
61        }
62    }
63}
64
65/// One queued outgoing message.
66#[derive(Clone, Debug, PartialEq, Eq)]
67pub struct Request {
68    bytes: Vec<u8>,
69    expect: Option<Expect>,
70    /// Part of the startup sync (counts towards its progress).
71    sync: bool,
72}
73
74impl Request {
75    pub fn bytes(&self) -> &[u8] {
76        &self.bytes
77    }
78
79    /// `None`: nothing comes back (CC/PC), done once sent.
80    pub fn expect(&self) -> Option<Expect> {
81        self.expect
82    }
83
84    pub fn is_sync(&self) -> bool {
85        self.sync
86    }
87}
88
89/// How an in-flight request ended.
90#[derive(Clone, Debug, PartialEq, Eq)]
91pub struct Done {
92    pub request: Request,
93    /// The reply arrived (`false`: timed out).
94    pub answered: bool,
95}
96
97/// Queue of SysEx requests and writes with at most one in flight.
98#[derive(Debug)]
99pub struct RequestQueue {
100    queue: VecDeque<Request>,
101    in_flight: Option<(Request, Instant)>,
102    timeout: Duration,
103    sync_total: usize,
104    sync_done: usize,
105}
106
107/// The official app's startup reads (docs/03), all read-only. Presets come before the edit
108/// buffer so the active preset can be identified by matching the edit buffer against them
109/// (see [`DeviceState`](crate::DeviceState)).
110pub fn handshake() -> Vec<Vec<u8>> {
111    let mut reqs = vec![req::version(), req::product(), req::settings(), req::cur_state(), req::drum()];
112    reqs.extend((0..10).map(req::ir_slot));
113    reqs.extend((0..10).map(req::user_amp));
114    reqs.extend((0..99).map(req::preset));
115    reqs.extend([req::edit_buffer(), req::looper(), req::tuner()]);
116    reqs
117}
118
119/// Follow-up reads after pedal-side changes that carry no data themselves: a preset saved on the
120/// pedal (re-read it and the edit buffer) and the tuner toggled (read its state).
121pub fn follow_up(msg: &Msg) -> Vec<Vec<u8>> {
122    match msg {
123        Msg::Notify { what, payload } if *what == Cmd::Preset as u8 => {
124            let i = payload.first().copied().unwrap_or(0);
125            vec![req::preset(i), req::edit_buffer()]
126        }
127        Msg::Notify { what, .. } if *what == Cmd::Tuner as u8 => vec![req::tuner()],
128        _ => vec![],
129    }
130}
131
132impl RequestQueue {
133    /// `timeout`: how long to wait for a reply before moving on (the official app uses 1 s).
134    pub fn new(timeout: Duration) -> Self {
135        RequestQueue { queue: VecDeque::new(), in_flight: None, timeout, sync_total: 0, sync_done: 0 }
136    }
137
138    /// Drop everything queued and in flight, and any sync in progress.
139    pub fn clear(&mut self) {
140        self.queue.clear();
141        self.in_flight = None;
142        self.sync_total = 0;
143        self.sync_done = 0;
144    }
145
146    /// Start a sync: replaces the queue with `reqs`, whose replies count towards [`progress`].
147    ///
148    /// [`progress`]: RequestQueue::progress
149    pub fn start_sync(&mut self, reqs: Vec<Vec<u8>>) {
150        self.clear();
151        self.sync_total = reqs.len();
152        for bytes in reqs {
153            self.read(bytes);
154        }
155    }
156
157    /// Queue a read. Ignored when the same request is already waiting.
158    pub fn read(&mut self, bytes: Vec<u8>) {
159        if !self.queue.iter().any(|r| r.bytes == bytes) {
160            let expect = Expect::of(&bytes);
161            self.queue.push_back(Request { bytes, expect, sync: self.sync_total > 0 });
162        }
163    }
164
165    /// Queue a write (or any message that must keep its place after queued SysEx). Settings are
166    /// rewritten whole on every change, so a queued settings write is replaced by the newer one.
167    pub fn write(&mut self, bytes: Vec<u8>) {
168        let expect = Expect::of(&bytes);
169        if let Some(Expect::Ack(c)) = expect
170            && c == Cmd::System as u8
171            && let Some(r) = self.queue.iter_mut().find(|r| r.expect == expect)
172        {
173            r.bytes = bytes;
174            return;
175        }
176        self.queue.push_back(Request { bytes, expect, sync: false });
177    }
178
179    /// The next message to send, if nothing is in flight. Pass it to [`sent`] once transmitted.
180    ///
181    /// [`sent`]: RequestQueue::sent
182    pub fn pop(&mut self) -> Option<Request> {
183        if self.in_flight.is_some() {
184            return None;
185        }
186        self.queue.pop_front()
187    }
188
189    /// Record that `r` went out at `now`; it is in flight until answered or timed out.
190    pub fn sent(&mut self, r: Request, now: Instant) {
191        if r.expect.is_some() {
192            self.in_flight = Some((r, now));
193        }
194    }
195
196    /// Feed a complete incoming message. Returns the request it answers, if any.
197    pub fn on_message(&mut self, msg: &Msg) -> Option<Done> {
198        let (r, _) = self.in_flight.as_ref()?;
199        if !r.expect.is_some_and(|e| e.matches(msg)) {
200            return None;
201        }
202        let (request, _) = self.in_flight.take()?;
203        self.finish(request, true)
204    }
205
206    /// Expire the in-flight request if it has waited longer than the timeout.
207    pub fn expire(&mut self, now: Instant) -> Option<Done> {
208        let (_, t) = self.in_flight.as_ref()?;
209        if now.saturating_duration_since(*t) <= self.timeout {
210            return None;
211        }
212        let (request, _) = self.in_flight.take()?;
213        self.finish(request, false)
214    }
215
216    fn finish(&mut self, request: Request, answered: bool) -> Option<Done> {
217        if request.sync && self.sync_total > 0 {
218            self.sync_done += 1;
219            if self.sync_done >= self.sync_total {
220                self.sync_total = 0;
221                self.sync_done = 0;
222            }
223        }
224        Some(Done { request, answered })
225    }
226
227    /// `(done, total)` of the current sync; `(0, 0)` when none is running.
228    pub fn progress(&self) -> (usize, usize) {
229        (self.sync_done, self.sync_total)
230    }
231
232    pub fn in_flight(&self) -> Option<&Request> {
233        self.in_flight.as_ref().map(|(r, _)| r)
234    }
235
236    /// Messages waiting (not counting the one in flight).
237    pub fn len(&self) -> usize {
238        self.queue.len()
239    }
240
241    pub fn is_empty(&self) -> bool {
242        self.queue.is_empty() && self.in_flight.is_none()
243    }
244}
245
246#[cfg(test)]
247mod tests {
248    use super::*;
249    use crate::midi::host;
250    use crate::records::Settings;
251
252    fn t0() -> Instant {
253        Instant::now()
254    }
255
256    #[test]
257    fn one_in_flight_until_answered() {
258        let mut q = RequestQueue::new(Duration::from_secs(1));
259        q.read(req::settings());
260        q.read(req::drum());
261        q.read(req::settings()); // duplicate, ignored
262        assert_eq!(q.len(), 2);
263        let r = q.pop().unwrap();
264        assert_eq!(r.expect(), Some(Expect::Reply(0x14)));
265        q.sent(r, t0());
266        assert!(q.pop().is_none());
267        // unrelated traffic doesn't release it
268        assert!(q.on_message(&Msg::parse(&[0xB0, 0x12, 5])).is_none());
269        assert!(q.on_message(&Msg::parse(&req::settings())).is_none()); // dir 0, not a reply
270        let done = q.on_message(&Msg::Nep { cmd: 0x14, dir: 2, payload: vec![0; 20] }).unwrap();
271        assert!(done.answered);
272        assert_eq!(q.pop().unwrap().bytes(), &req::drum()[..]);
273    }
274
275    #[test]
276    fn live_messages_are_not_in_flight() {
277        let mut q = RequestQueue::new(Duration::from_secs(1));
278        q.write(host::program(3));
279        q.write(host::cc(0x12, 50));
280        let r = q.pop().unwrap();
281        assert_eq!(r.expect(), None);
282        q.sent(r, t0());
283        assert!(q.in_flight().is_none());
284        assert!(q.pop().is_some());
285    }
286
287    #[test]
288    fn timeout_moves_on_and_counts_sync() {
289        let mut q = RequestQueue::new(Duration::from_millis(100));
290        q.start_sync(vec![req::version(), req::product()]);
291        let start = t0();
292        let r = q.pop().unwrap();
293        assert_eq!(r.expect(), Some(Expect::Short(0x10)));
294        q.sent(r, start);
295        assert!(q.expire(start + Duration::from_millis(50)).is_none());
296        let done = q.expire(start + Duration::from_millis(150)).unwrap();
297        assert!(!done.answered && done.request.is_sync());
298        assert_eq!(q.progress(), (1, 2));
299        let r = q.pop().unwrap();
300        q.sent(r, start);
301        // product reply: F0 00 21 7A 60 <name> F7
302        let reply = [0xF0, 0, 0x21, 0x7A, 0x60, b'J', b'M', b'E', 0xF7];
303        assert!(q.on_message(&Msg::parse(&reply)).unwrap().answered);
304        assert_eq!(q.progress(), (0, 0));
305    }
306
307    #[test]
308    fn settings_writes_coalesce() {
309        let mut q = RequestQueue::new(Duration::from_secs(1));
310        q.write(host::program(1));
311        let mut s = Settings { raw: vec![0; 17], ..Default::default() };
312        q.write(host::settings_write(&s));
313        s.input_trim = 13;
314        q.write(host::settings_write(&s));
315        assert_eq!(q.len(), 2);
316        q.pop();
317        assert_eq!(q.pop().unwrap().bytes(), &host::settings_write(&s)[..]);
318    }
319
320    #[test]
321    fn handshake_matches_official_app() {
322        let h = handshake();
323        assert_eq!(h.len(), 5 + 10 + 10 + 99 + 3);
324        assert!(h.iter().all(|r| Expect::of(r).is_some()), "every handshake read expects a reply");
325    }
326
327    #[test]
328    fn follow_ups() {
329        let saved = Msg::Notify { what: 0x0B, payload: vec![7] };
330        assert_eq!(follow_up(&saved), vec![req::preset(7), req::edit_buffer()]);
331        assert_eq!(follow_up(&Msg::Notify { what: 0x6F, payload: vec![] }), vec![req::tuner()]);
332        assert!(follow_up(&Msg::Pc { ch: 0, prog: 1 }).is_empty());
333    }
334}
open_stomp/queue.rs

open_stomp/
queue.rs
