brep_app/automation/

queue.rs

//! The command queue: the channel between a host and the app.
//!
//! A host calls [`AutomationQueue::submit`] with an [`Envelope`] and gets a
//! receiver for the [`Reply`]. The app drains the queue at three fixed points
//! in the frame (§4.4): [`drain_input`](AutomationQueue::drain_input) in
//! `raw_input_hook`, then [`drain_app`](AutomationQueue::drain_app) for
//! `Phase::Mutate` at the top of `ui` and for `Phase::Read` at the bottom. A
//! command waits in the queue until its phase comes round.
use crate::automation::command::{lookup, Ctx, Envelope, Handler, Notice, NoticeKind, Outcome, Phase, Reply};
use crate::automation::pointer::Pointer;
use std::collections::VecDeque;
use std::sync::mpsc::{channel, Receiver, Sender};
use std::sync::{Arc, Mutex};

pub struct Pending {
    pub envelope: Envelope,
    pub reply: Sender<Reply>,
}

struct Awaiting {
    token: u64,
    region: crate::automation::cmd_capture::Region,
    pending: Pending,
}

#[derive(Default)]
struct Inner {
    pending: VecDeque<Pending>,
    awaiting: Vec<Awaiting>,
    notices: Vec<Notice>,
    pointer: Pointer,
    next_token: u64,
    frame: u64,
    poisoned: Option<String>,
}

#[derive(Default)]
pub struct AutomationQueue {
    inner: Mutex<Inner>,
    ctx: Mutex<Option<egui::Context>>,
}

impl AutomationQueue {
    pub fn new() -> Arc<Self> {
        Arc::new(Self::default())
    }

    /// Give the queue the egui context so a submit can wake an idle frame loop.
    pub fn attach(&self, ctx: &egui::Context) {
        *self.ctx.lock().unwrap() = Some(ctx.clone());
    }

    /// Enqueue a command. The reply arrives on the returned receiver once the
    /// app has reached the command's phase (or at once, for an unknown command
    /// or a poisoned session).
    pub fn submit(&self, envelope: Envelope) -> Receiver<Reply> {
        let (tx, rx) = channel();
        let mut inner = self.inner.lock().unwrap();
        if let Some(why) = &inner.poisoned {
            let _ = tx.send(Reply::err(envelope.id, inner.frame, format!("session poisoned: {why}"), vec![]));
            return rx;
        }
        if lookup(&envelope.cmd).is_none() {
            let _ = tx.send(Reply::err(envelope.id, inner.frame, format!("unknown command `{}`", envelope.cmd), vec![]));
            return rx;
        }
        inner.pending.push_back(Pending { envelope, reply: tx });
        drop(inner);
        if let Some(ctx) = self.ctx.lock().unwrap().as_ref() {
            ctx.request_repaint();
        }
        rx
    }

    pub fn push_notice(&self, kind: NoticeKind, text: impl Into<String>) {
        let mut inner = self.inner.lock().unwrap();
        let frame = inner.frame;
        inner.notices.push(Notice { kind, frame, text: text.into() });
    }

    pub fn take_notices(&self) -> Vec<Notice> {
        std::mem::take(&mut self.inner.lock().unwrap().notices)
    }

    pub fn pointer(&self) -> Pointer {
        self.inner.lock().unwrap().pointer.clone()
    }

    pub fn frame(&self) -> u64 {
        self.inner.lock().unwrap().frame
    }

    pub fn has_pending(&self) -> bool {
        let inner = self.inner.lock().unwrap();
        !inner.pending.is_empty() || !inner.awaiting.is_empty()
    }

    /// Mark the session unusable (a panic unwound through the app). Every
    /// pending and future command is answered with the reason.
    pub fn poison(&self, why: impl Into<String>) {
        let why = why.into();
        let mut inner = self.inner.lock().unwrap();
        inner.poisoned = Some(why.clone());
        let frame = inner.frame;
        for p in inner.pending.drain(..) {
            let _ = p.reply.send(Reply::err(p.envelope.id, frame, format!("session poisoned: {why}"), vec![]));
        }
        for a in inner.awaiting.drain(..) {
            let _ = a.pending.reply.send(Reply::err(a.pending.envelope.id, frame, format!("session poisoned: {why}"), vec![]));
        }
    }

    pub fn is_poisoned(&self) -> bool {
        self.inner.lock().unwrap().poisoned.is_some()
    }

    /// Phase 1 of the frame: apply input commands as `egui::Event`s and
    /// complete captures whose `Event::Screenshot` arrived. `view` is the
    /// viewport rect of the last frame (for `region: viewport` crops).
    pub fn drain_input(&self, raw: &mut egui::RawInput, frame: u64, ppp: f32, view: Option<egui::Rect>) {
        let mut inner = self.inner.lock().unwrap();
        inner.frame = frame;

        // Screenshots that landed in this frame's input.
        if !inner.awaiting.is_empty() {
            let mut done = Vec::new();
            for event in &raw.events {
                if let egui::Event::Screenshot { user_data, image, .. } = event {
                    let token = user_data.data.as_ref().and_then(|d| d.downcast_ref::<u64>()).copied();
                    if let Some(token) = token {
                        if let Some(pos) = inner.awaiting.iter().position(|a| a.token == token) {
                            let a = inner.awaiting.remove(pos);
                            done.push((a, image.clone()));
                        }
                    }
                }
            }
            for (a, image) in done {
                let notices = std::mem::take(&mut inner.notices);
                let reply = match crate::automation::cmd_capture::encode_capture(&image, ppp, view, &a.region) {
                    Ok((json, png)) => {
                        let mut r = Reply::ok(a.pending.envelope.id, frame, json, notices);
                        r.blob = Some(png);
                        r
                    }
                    Err(e) => Reply::err(a.pending.envelope.id, frame, e, notices),
                };
                let _ = a.pending.reply.send(reply);
            }
        }

        // Input-phase commands, in order.
        let mut keep = VecDeque::new();
        let mut events = Vec::new();
        while let Some(p) = inner.pending.pop_front() {
            let Some(spec) = lookup(&p.envelope.cmd) else { continue };
            match (&spec.handler, spec.phase) {
                (Handler::Input(f), Phase::Input) => {
                    let result = f(&mut inner.pointer, p.envelope.args.clone(), &mut events);
                    let notices = std::mem::take(&mut inner.notices);
                    let reply = match result {
                        Ok(v) => Reply::ok(p.envelope.id, frame, v, notices),
                        Err(e) => Reply::err(p.envelope.id, frame, e, notices),
                    };
                    let _ = p.reply.send(reply);
                }
                _ => keep.push_back(p),
            }
        }
        inner.pending = keep;
        raw.modifiers = inner.pointer.modifiers.egui();
        raw.events.extend(events);
    }

    /// Phases 2 and 3 of the frame. Runs every pending command registered for
    /// `phase`, in order.
    pub fn drain_app(&self, phase: Phase, ctx: &mut Ctx<'_>, frame: u64) {
        let batch: Vec<Pending> = {
            let mut inner = self.inner.lock().unwrap();
            inner.frame = frame;
            let mut keep = VecDeque::new();
            let mut batch = Vec::new();
            while let Some(p) = inner.pending.pop_front() {
                match lookup(&p.envelope.cmd) {
                    Some(spec) if spec.phase == phase => batch.push(p),
                    _ => keep.push_back(p),
                }
            }
            inner.pending = keep;
            batch
        };
        for p in batch {
            let spec = lookup(&p.envelope.cmd).expect("checked at submit");
            let Handler::App(f) = &spec.handler else { continue };
            let result = f(ctx, p.envelope.args.clone());
            let mut inner = self.inner.lock().unwrap();
            let notices = std::mem::take(&mut inner.notices);
            match result {
                Ok(Outcome::Done(v)) => {
                    let _ = p.reply.send(Reply::ok(p.envelope.id, frame, v, notices));
                }
                Ok(Outcome::Blob { json, bytes }) => {
                    let mut r = Reply::ok(p.envelope.id, frame, json, notices);
                    r.blob = Some(bytes);
                    let _ = p.reply.send(r);
                }
                Ok(Outcome::AwaitScreenshot { token, region }) => {
                    inner.notices = notices;
                    inner.awaiting.push(Awaiting { token, region, pending: p });
                }
                Err(e) => {
                    let _ = p.reply.send(Reply::err(p.envelope.id, frame, e, notices));
                }
            }
        }
    }

    /// A fresh token for a screenshot request.
    pub fn next_token(&self) -> u64 {
        let mut inner = self.inner.lock().unwrap();
        inner.next_token += 1;
        inner.next_token
    }
}