mirror of
https://github.com/storytold/pdfcraft.git
synced 2026-10-31 21:48:09 -06:00
refactor(ui-egui): split control.rs into control/
New modules: plugin, requests, screenshot, pending, host, transport (non-wasm). transport_tests and screenshot_geometry_tests move with their code; lifecycle_tests and pointer_tests stay.
This commit is contained in:
aiongg
+17
-1783
File diff suppressed because it is too large
Load Diff
@@ -0,0 +1,188 @@
|
||||
use super::*;
|
||||
|
||||
impl Host for crate::PdfCraftApp {
|
||||
/// Cancel any in-progress pointer gesture in every open view.
|
||||
fn cancel_pointer_gestures(&mut self) {
|
||||
for view in &mut self.views {
|
||||
view.cancel_pointer_gestures();
|
||||
}
|
||||
}
|
||||
|
||||
/// The input target: active document, open dialog/prompt, palette and find state.
|
||||
///
|
||||
/// Synthetic input admitted for one generation is never replayed into another document or
|
||||
/// dialog. Only these identities are hashed, never text or document content.
|
||||
fn input_epoch(&self) -> u64 {
|
||||
use std::hash::{Hash as _, Hasher as _};
|
||||
let mut hasher = std::collections::hash_map::DefaultHasher::new();
|
||||
self.capture_epoch().hash(&mut hasher);
|
||||
// The focused widget is part of the input target: keys typed for one field must not land
|
||||
// in another after focus moves. Only the id value is hashed, never text.
|
||||
let focused = self.ctx.as_ref().and_then(|ctx| ctx.memory(|m| m.focused())).map(|id| id.value());
|
||||
focused.hash(&mut hasher);
|
||||
hasher.finish()
|
||||
}
|
||||
|
||||
/// The captured content: active document and open dialog/prompt, never keyboard focus.
|
||||
fn capture_epoch(&self) -> u64 {
|
||||
use std::hash::{Hash as _, Hasher as _};
|
||||
let mut hasher = std::collections::hash_map::DefaultHasher::new();
|
||||
let doc = self.active.and_then(|i| self.views.get(i)).map(|v| v.id.0);
|
||||
doc.hash(&mut hasher);
|
||||
format!(
|
||||
"{:?}/{:?}/{:?}/{:?}/{:?}",
|
||||
self.dialog,
|
||||
self.password_prompt.is_some(),
|
||||
self.palette_open,
|
||||
self.close_request.is_some(),
|
||||
self.active.and_then(|i| self.views.get(i)).and_then(|v| v.find.as_ref().map(|_| true)),
|
||||
)
|
||||
.hash(&mut hasher);
|
||||
hasher.finish()
|
||||
}
|
||||
|
||||
/// Control requests wait for the first UI pass that installs fonts; a font-gated early
|
||||
/// return cannot consume injected input or issue a screenshot.
|
||||
fn accepts_input(&self) -> bool {
|
||||
self.fonts_ready
|
||||
}
|
||||
|
||||
fn panel(&mut self, reset: bool, params: &Value) -> Reply {
|
||||
crate::side_panels::command(self, reset, params)
|
||||
}
|
||||
|
||||
fn quit(&mut self, ctx: &egui::Context) -> bool {
|
||||
self.request_quit(ctx)
|
||||
}
|
||||
|
||||
fn state(&self) -> Value {
|
||||
let active = self.active.and_then(|i| self.views.get(i));
|
||||
let docs: Vec<Value> = self
|
||||
.views
|
||||
.iter()
|
||||
.enumerate()
|
||||
.filter_map(|(i, v)| {
|
||||
let d = self.session.get(v.id)?;
|
||||
Some(json!({ "tab": i, "doc": v.id.0, "name": d.name, "path": d.path, "pages": d.info.pages.len(), "dirty": d.dirty, "active": self.active == Some(i) }))
|
||||
})
|
||||
.collect();
|
||||
#[cfg(not(target_arch = "wasm32"))]
|
||||
let pending_opens: Vec<Value> =
|
||||
self.pending_opens.iter().map(|pending| json!({ "token": pending.token, "name": pending.name, "path": pending.path })).collect();
|
||||
#[cfg(target_arch = "wasm32")]
|
||||
let pending_opens: Vec<Value> = Vec::new();
|
||||
json!({
|
||||
"side_panels": crate::side_panels::state(self),
|
||||
"documents": docs,
|
||||
"pending_opens": pending_opens,
|
||||
"active": active.map(|v| json!({
|
||||
"doc": v.id.0,
|
||||
"page": v.current + 1,
|
||||
"zoom_percent": (v.zoom * 100.0).round(),
|
||||
"fit": format!("{:?}", v.fit),
|
||||
"layout": format!("{:?}", v.layout),
|
||||
"continuous": v.continuous(),
|
||||
"page_gaps": v.show_page_gaps,
|
||||
"organize": v.organize,
|
||||
"grid_zoom": v.grid_zoom(),
|
||||
"thumbnail_zoom": v.thumbnail_zoom(),
|
||||
// Pages picked in the organize grid or the Pages panel (empty: the current page).
|
||||
"selected_pages": v.selected.iter().map(|p| p + 1).collect::<Vec<_>>(),
|
||||
"auto_scrolling": v.auto_scrolling(),
|
||||
"raster_memory": v.raster_memory(),
|
||||
"viewport": [v.viewport_rect().min.x, v.viewport_rect().min.y, v.viewport_rect().max.x, v.viewport_rect().max.y],
|
||||
"find_open": v.find.is_some(),
|
||||
"page_errors": v.page_errors().iter().map(|(p, e)| json!({ "page": p + 1, "error": e })).collect::<Vec<_>>(),
|
||||
// Where pages are on screen (points), to aim ui.click / ui.drag at page content.
|
||||
"pages_on_screen": v.visible_page_rects().iter().map(|(p, r)| json!({ "page": p + 1, "rect": [r.min.x, r.min.y, r.max.x, r.max.y] })).collect::<Vec<_>>(),
|
||||
"selected_comment": v.comments.selected.map(|(p, i)| json!({ "page": p + 1, "index": i + 1 })),
|
||||
"comment_composer_open": v.comments.composer.is_some(),
|
||||
})),
|
||||
// The Combine files tab: how its files show, and each grid thumbnail's state.
|
||||
"combine": {
|
||||
"showing": self.combine_showing(),
|
||||
"view": self.combine_view.as_str(),
|
||||
"zoom": self.combine_zoom,
|
||||
"files": self.combine_draft.iter().map(|f| f.name.clone()).collect::<Vec<_>>(),
|
||||
"selected": self.combine_selection(),
|
||||
"thumbnails": self.combine_thumbnails().iter().map(|t| format!("{t:?}")).collect::<Vec<_>>(),
|
||||
"expanded": self.combine_expanded().iter().filter_map(|i| self.combine_draft.get(*i).map(|f| f.name.clone())).collect::<Vec<_>>(),
|
||||
"preview": self.combine_preview().map(|(file, page, render)| json!({ "file": file, "page": page + 1, "render": format!("{render:?}") })),
|
||||
},
|
||||
"quick_tool": match self.quick_tool {
|
||||
crate::QuickTool::Measure(t) => format!("measure-{}", t.name()),
|
||||
crate::QuickTool::Select => "select".to_string(),
|
||||
crate::QuickTool::Hand => "hand".to_string(),
|
||||
crate::QuickTool::Crop => "crop".to_string(),
|
||||
crate::QuickTool::EnhanceCamera => "enhance-camera".to_string(),
|
||||
crate::QuickTool::Redact => "redact".to_string(),
|
||||
crate::QuickTool::AddText => "add-text".to_string(),
|
||||
crate::QuickTool::EditText => "edit-text".to_string(),
|
||||
crate::QuickTool::Link => "link".to_string(),
|
||||
crate::QuickTool::SignArea { certify: false } => "sign".to_string(),
|
||||
crate::QuickTool::SignArea { certify: true } => "certify".to_string(),
|
||||
crate::QuickTool::MarqueeZoom => "marquee-zoom".to_string(),
|
||||
crate::QuickTool::Snapshot => "snapshot".to_string(),
|
||||
crate::QuickTool::ColumnSelect => "column-select".to_string(),
|
||||
crate::QuickTool::Stamp(k) => format!("stamp-{}", k.name().trim_start_matches("PC").to_ascii_lowercase()),
|
||||
crate::QuickTool::CustomStamp(i) => format!("custom-stamp-{i}"),
|
||||
crate::QuickTool::Fill(f) => format!("fill-{}", f.command().trim_start_matches("sign.fill.")),
|
||||
crate::QuickTool::Field(f) => format!("field-{}", f.command().trim_start_matches("form.add.")),
|
||||
crate::QuickTool::Comment(t) => t.command().trim_start_matches("comment.").to_string(),
|
||||
},
|
||||
"home": self.active.is_none(),
|
||||
"mode": format!("{:?}", self.mode),
|
||||
"default_mode": self.default_mode,
|
||||
"comment_tool_pinned": self.comment_prefs.pinned,
|
||||
"left_panel": if self.left_open { json!(format!("{:?}", self.left)) } else { Value::Null },
|
||||
"tools_side": if self.side_panels.tools_on_right() { "right" } else { "left" },
|
||||
"right_panel": self.right.map(|r| format!("{r:?}")),
|
||||
"dialog": self.dialog.map(|d| format!("{d:?}")),
|
||||
"palette_open": self.palette_open,
|
||||
"theme": format!("{:?}", self.theme),
|
||||
"theme_preference": self.theme_preference,
|
||||
"language": self.language,
|
||||
"notice": self.toast.as_ref().map(|t| t.0.clone()),
|
||||
"progress": self.progress_notice.as_ref().map(|p| json!({ "label": p.label, "fraction": p.fraction })),
|
||||
"password_prompt": self.password_prompt.is_some(),
|
||||
"close_prompt": self.close_request.is_some(),
|
||||
})
|
||||
}
|
||||
|
||||
fn command(&mut self, id: &str) -> Reply {
|
||||
let spec = pdfcraft_engine::commands::command(id).ok_or_else(|| format!("unknown command {id:?} (see ui.commands)"))?;
|
||||
if !self.command_enabled(spec) {
|
||||
return Err(format!("{id} is disabled right now"));
|
||||
}
|
||||
self.execute(id);
|
||||
Ok(json!({ "ran": id }))
|
||||
}
|
||||
|
||||
fn commands(&self) -> Value {
|
||||
let mut list: Vec<Value> = pdfcraft_engine::commands::COMMANDS
|
||||
.iter()
|
||||
.map(|c| json!({ "id": c.id, "label": c.label, "menu": c.menu, "enabled": self.command_enabled(c), "shortcut": c.shortcut.map(|s| s.label(cfg!(target_os = "macos"))) }))
|
||||
.collect();
|
||||
list.push(json!({"id": "ui.panel", "label": "Side panel", "enabled": true, "params": "panel: tools or side; open: boolean; width: number (clamped to the panel's limits)"}));
|
||||
json!({ "commands": list })
|
||||
}
|
||||
|
||||
fn set(&mut self, key: &str, value: &str) -> Reply {
|
||||
self.set_option(key, value).map(|()| json!({ "set": key, "value": value }))
|
||||
}
|
||||
|
||||
#[cfg(not(target_arch = "wasm32"))]
|
||||
fn open(&mut self, path: &str) -> Reply {
|
||||
match self.open_path(path) {
|
||||
crate::OpenOutcome::Opened => Ok(json!({ "opened": path, "password_prompt": self.password_prompt.is_some() })),
|
||||
crate::OpenOutcome::Focused => Ok(json!({ "opened": path, "focused": true })),
|
||||
crate::OpenOutcome::Pending => Ok(json!({ "accepted": path, "pending": true })),
|
||||
crate::OpenOutcome::Failed => Err(self.toast.as_ref().map(|t| t.0.clone()).unwrap_or_else(|| format!("couldn't open {path}"))),
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(target_arch = "wasm32")]
|
||||
fn open(&mut self, _path: &str) -> Reply {
|
||||
Err("ui.open needs a file system; on the web, drop the file on the page".into())
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,176 @@
|
||||
use super::screenshot::{Geometry, MAX_IMAGE_BYTES, retried_reservation, screenshot_png};
|
||||
use super::*;
|
||||
|
||||
/// A screenshot pending in the `finish_pending` pass, with ownership of its reply.
|
||||
struct PendingCapture {
|
||||
slot: craft_ui::capture::CaptureSlot,
|
||||
callback: Option<egui::ScreenshotCallback>,
|
||||
region: Option<egui::Rect>,
|
||||
reply: Sender<Reply>,
|
||||
since: web_time::Instant,
|
||||
geometry: Geometry,
|
||||
reserved: usize,
|
||||
attempts: u8,
|
||||
}
|
||||
|
||||
impl PendingCapture {
|
||||
/// Convert a screenshot request into the pending form that later frames finish.
|
||||
fn into_pending(self) -> Pending {
|
||||
Pending::Screenshot {
|
||||
slot: self.slot,
|
||||
callback: self.callback,
|
||||
region: self.region,
|
||||
reply: self.reply,
|
||||
since: self.since,
|
||||
geometry: self.geometry,
|
||||
reserved: self.reserved,
|
||||
attempts: self.attempts,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// The state one `finish_pending` pass needs while resolving pending replies.
|
||||
struct PendingFinisher {
|
||||
now: web_time::Instant,
|
||||
passes: u64,
|
||||
current: Geometry,
|
||||
in_pass: bool,
|
||||
dropped: Vec<u64>,
|
||||
expired: Vec<u64>,
|
||||
reserved_total: usize,
|
||||
}
|
||||
|
||||
impl PendingFinisher {
|
||||
/// Answer or keep one queued-input request, preserving its one arrival deadline.
|
||||
fn resolve_frames(&self, until_pass: u64, batches: Vec<u64>, since: web_time::Instant, reply: Sender<Reply>, value: Value) -> Option<Pending> {
|
||||
if batches.iter().any(|id| self.expired.contains(id)) {
|
||||
let _ = reply.send(Err(not_drawn("handling the input")));
|
||||
return None;
|
||||
}
|
||||
if batches.iter().any(|id| self.dropped.contains(id)) {
|
||||
let _ = reply.send(Err("the document, dialog or focus changed before the input could be applied; nothing was done".into()));
|
||||
return None;
|
||||
}
|
||||
if self.passes >= until_pass {
|
||||
let _ = reply.send(Ok(value));
|
||||
return None;
|
||||
}
|
||||
if !self.in_pass && self.now.duration_since(since) > timeout() {
|
||||
let _ = reply.send(Err(not_drawn("handling the input")));
|
||||
return None;
|
||||
}
|
||||
Some(Pending::Frames { until_pass, batches, since, reply, value })
|
||||
}
|
||||
|
||||
/// Give one reservation back to the running aggregate budget.
|
||||
fn release_reservation(&mut self, reserved: usize) {
|
||||
self.reserved_total = self.reserved_total.saturating_sub(reserved);
|
||||
}
|
||||
|
||||
/// Resolve one screenshot request, keeping it only while its capture is still pending.
|
||||
fn finish_capture(&mut self, ctx: &egui::Context, capture: PendingCapture) -> Option<Pending> {
|
||||
// Completion cannot restart or bypass the admission deadline, even when the callback
|
||||
// arrived before this delayed tick or a resize would retry it.
|
||||
if self.now.duration_since(capture.since) > timeout() {
|
||||
capture.slot.cancel();
|
||||
self.release_reservation(capture.reserved);
|
||||
let _ = capture.reply.send(Err(not_drawn("the screenshot")));
|
||||
return None;
|
||||
}
|
||||
match capture.slot.take() {
|
||||
Some(Ok(image)) => self.finish_captured_image(ctx, capture, image),
|
||||
Some(Err(error)) => {
|
||||
self.release_reservation(capture.reserved);
|
||||
let _ = capture.reply.send(Err(format!("the screenshot could not be captured: {error}")));
|
||||
None
|
||||
}
|
||||
None => Some(capture.into_pending()),
|
||||
}
|
||||
}
|
||||
|
||||
/// Finish a screenshot whose callback produced an image.
|
||||
fn finish_captured_image(&mut self, ctx: &egui::Context, capture: PendingCapture, image: Arc<egui::ColorImage>) -> Option<Pending> {
|
||||
if self.current.epoch != capture.geometry.epoch {
|
||||
// The document/dialog changed: never save a capture of the previous context, and do
|
||||
// not retry it into the new one.
|
||||
self.release_reservation(capture.reserved);
|
||||
let _ = capture.reply.send(Err("the document or dialog changed while the screenshot was being captured; it was not saved".into()));
|
||||
return None;
|
||||
}
|
||||
if self.current.size != capture.geometry.size || self.current.pixels_per_point != capture.geometry.pixels_per_point {
|
||||
// A late completion after a resize/DPI change is never encoded: retry once with a
|
||||
// fresh slot, else report a defined error.
|
||||
return self.retry_captured_image(ctx, capture);
|
||||
}
|
||||
self.release_reservation(capture.reserved);
|
||||
let _ = capture.reply.send(screenshot_png(&image, capture.region, capture.geometry));
|
||||
None
|
||||
}
|
||||
|
||||
/// Retry one resize/DPI-stale capture with the current geometry.
|
||||
fn retry_captured_image(&mut self, ctx: &egui::Context, capture: PendingCapture) -> Option<Pending> {
|
||||
if capture.attempts >= 1 {
|
||||
self.release_reservation(capture.reserved);
|
||||
let _ = capture.reply.send(Err("the window geometry changed while the screenshot was being captured; it was not saved".into()));
|
||||
return None;
|
||||
}
|
||||
if capture.region.as_ref().is_some_and(|region| !self.current.contains_region(*region)) {
|
||||
self.release_reservation(capture.reserved);
|
||||
let _ = capture.reply.send(Err("the region must lie inside the window".into()));
|
||||
return None;
|
||||
}
|
||||
let Some(bytes) = self.current.image_bytes() else {
|
||||
self.release_reservation(capture.reserved);
|
||||
let _ = capture.reply.send(Err("the window has no capturable size".into()));
|
||||
return None;
|
||||
};
|
||||
let Some(new_total) = retried_reservation(self.reserved_total, capture.reserved, bytes) else {
|
||||
self.release_reservation(capture.reserved);
|
||||
let _ = capture.reply.send(Err("the screenshot no longer fits the screenshot budget".into()));
|
||||
return None;
|
||||
};
|
||||
self.reserved_total = new_total;
|
||||
let (slot, callback) = craft_ui::capture::CaptureSlot::new(ctx, egui::ViewportId::ROOT, MAX_IMAGE_BYTES);
|
||||
Some(Pending::Screenshot {
|
||||
slot,
|
||||
callback: Some(callback),
|
||||
region: capture.region,
|
||||
reply: capture.reply,
|
||||
since: capture.since,
|
||||
geometry: self.current,
|
||||
reserved: bytes,
|
||||
attempts: capture.attempts + 1,
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
impl Control {
|
||||
/// Resolve completed frame and screenshot requests while retaining those that still need a pass or a
|
||||
/// retry.
|
||||
pub(super) fn finish_pending(&mut self, ctx: &egui::Context, host: &mut impl Host) {
|
||||
let now = web_time::Instant::now();
|
||||
let (dropped, expired, in_pass) = self.take_finished_shared(ctx, host, now);
|
||||
let mut finisher = PendingFinisher {
|
||||
now,
|
||||
passes: self.passes(),
|
||||
current: Geometry::current(ctx, host.capture_epoch()),
|
||||
in_pass,
|
||||
dropped,
|
||||
expired,
|
||||
reserved_total: self.reserved_bytes(),
|
||||
};
|
||||
let mut keep = Vec::new();
|
||||
for pending in std::mem::take(&mut self.pending) {
|
||||
let kept = match pending {
|
||||
Pending::Frames { until_pass, batches, since, reply, value } => finisher.resolve_frames(until_pass, batches, since, reply, value),
|
||||
Pending::Screenshot { slot, callback, region, reply, since, geometry, reserved, attempts } => {
|
||||
finisher.finish_capture(ctx, PendingCapture { slot, callback, region, reply, since, geometry, reserved, attempts })
|
||||
}
|
||||
};
|
||||
if let Some(pending) = kept {
|
||||
keep.push(pending);
|
||||
}
|
||||
}
|
||||
self.pending = keep;
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,172 @@
|
||||
use super::requests::Handled;
|
||||
use super::*;
|
||||
|
||||
/// Track only presses that reached egui and have not yet been released. A pointer release is
|
||||
/// retained as an ownership marker, never as a timeout event.
|
||||
fn note_pressed(release: &mut Vec<egui::Event>, events: &[egui::Event]) {
|
||||
for event in events {
|
||||
match event {
|
||||
egui::Event::Key { key, pressed: true, .. }
|
||||
if !release.iter().any(|event| matches!(event, egui::Event::Key { key: seen, pressed: false, .. } if seen == key)) =>
|
||||
{
|
||||
release.push(egui::Event::Key { key: *key, physical_key: None, pressed: false, repeat: false, modifiers: egui::Modifiers::NONE });
|
||||
}
|
||||
egui::Event::PointerButton { pos, button, pressed: true, .. }
|
||||
if !release
|
||||
.iter()
|
||||
.any(|event| matches!(event, egui::Event::PointerButton { button: seen, pressed: false, .. } if seen == button)) =>
|
||||
{
|
||||
release.push(egui::Event::PointerButton { pos: *pos, button: *button, pressed: false, modifiers: egui::Modifiers::NONE });
|
||||
}
|
||||
egui::Event::ModifiersChanged(modifiers)
|
||||
if *modifiers != egui::Modifiers::NONE && !release.contains(&egui::Event::ModifiersChanged(egui::Modifiers::NONE)) =>
|
||||
{
|
||||
release.push(egui::Event::ModifiersChanged(egui::Modifiers::NONE));
|
||||
}
|
||||
egui::Event::Key { key, pressed: false, .. } => {
|
||||
release.retain(|event| !matches!(event, egui::Event::Key { key: seen, .. } if seen == key));
|
||||
}
|
||||
egui::Event::PointerButton { button, pressed: false, .. } => {
|
||||
release.retain(|event| !matches!(event, egui::Event::PointerButton { button: seen, .. } if seen == button));
|
||||
}
|
||||
egui::Event::ModifiersChanged(modifiers) if *modifiers == egui::Modifiers::NONE => {
|
||||
release.retain(|event| !matches!(event, egui::Event::ModifiersChanged(_)));
|
||||
}
|
||||
_ => {}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Cancel expired requests before egui computes interaction. Hidden logic calls this too, so
|
||||
/// replies do not wait for a visible frame. Never complete an abandoned pointer gesture.
|
||||
pub(super) fn expire_input(shared: &mut Shared, now: web_time::Instant) {
|
||||
shared.inject.retain_mut(|request| {
|
||||
if now.duration_since(request.since) <= timeout() {
|
||||
return true;
|
||||
}
|
||||
shared.expired.push(request.request);
|
||||
for event in request.release.drain(..) {
|
||||
if matches!(event, egui::Event::PointerButton { .. }) {
|
||||
shared.cancel_pointer = true;
|
||||
shared.cancel_gestures = true;
|
||||
} else {
|
||||
shared.cleanup.push(event);
|
||||
}
|
||||
}
|
||||
false
|
||||
});
|
||||
}
|
||||
|
||||
/// Synthetic input is bounded like screenshots: a local authenticated client must not be able to
|
||||
/// queue unbounded batches (one frame each) and hold the control channel's memory for ever.
|
||||
const MAX_QUEUED_BATCHES: usize = 512;
|
||||
|
||||
/// The egui plugin half of the control channel.
|
||||
pub struct ControlPlugin {
|
||||
pub(super) shared: Arc<Mutex<Shared>>,
|
||||
}
|
||||
|
||||
impl egui::Plugin for ControlPlugin {
|
||||
fn debug_name(&self) -> &'static str {
|
||||
"pdfcraft-control"
|
||||
}
|
||||
|
||||
/// Inject admitted control batches into the current pass, dropping stale requests, applying cancellations
|
||||
/// and scheduling repaints.
|
||||
fn input_hook(&mut self, ctx: &egui::Context, input: &mut egui::RawInput) {
|
||||
let Ok(mut s) = self.shared.lock() else { return };
|
||||
expire_input(&mut s, web_time::Instant::now());
|
||||
if std::mem::take(&mut s.cancel_pointer) {
|
||||
// input_hook precedes Context::begin_pass. stop_dragging sets the OLD snapshot's
|
||||
// drag_stopped; begin_pass replaces that snapshot without a release/click/drop.
|
||||
ctx.input_mut(|i| i.pointer = Default::default());
|
||||
ctx.stop_dragging();
|
||||
egui::DragAndDrop::clear_payload(ctx);
|
||||
}
|
||||
input.events.append(&mut s.cleanup);
|
||||
s.in_pass = true;
|
||||
let epoch = s.epoch;
|
||||
let Some(front) = s.inject.front_mut() else { return };
|
||||
if !front.started {
|
||||
if front.epoch != epoch {
|
||||
// The document, dialog or focus changed after this request was admitted, before
|
||||
// any of its input landed. Drop only this request; later requests are unrelated.
|
||||
let request = front.request;
|
||||
s.inject.pop_front();
|
||||
s.dropped.push(request);
|
||||
if !s.inject.is_empty() {
|
||||
ctx.request_repaint();
|
||||
}
|
||||
return;
|
||||
}
|
||||
front.started = true;
|
||||
}
|
||||
// A non-expired request's own earlier batch may have changed focus or opened a dialog.
|
||||
// Let it finish normally; expiry above cancels held pointer state instead of completing
|
||||
// an abandoned click/drop.
|
||||
let events = front.batches.pop_front();
|
||||
if let Some(events) = &events {
|
||||
// Remember the inverse of what this request presses as it is applied, so a timeout
|
||||
// after this batch can still release exactly those inputs.
|
||||
note_pressed(&mut front.release, events);
|
||||
}
|
||||
let done = front.batches.is_empty();
|
||||
if done {
|
||||
s.inject.pop_front();
|
||||
}
|
||||
if let Some(events) = events {
|
||||
input.events.extend(events);
|
||||
}
|
||||
if !s.inject.is_empty() {
|
||||
ctx.request_repaint();
|
||||
}
|
||||
}
|
||||
|
||||
/// Record presented passes and the accesskit tree and focus after a pass, so control replies observe only
|
||||
/// drawn frames.
|
||||
fn output_hook(&mut self, ctx: &egui::Context, output: &mut egui::FullOutput) {
|
||||
let Ok(mut s) = self.shared.lock() else { return };
|
||||
s.in_pass = false;
|
||||
// A discarded pass is layout work, not a frame the agent can observe. Only count passes
|
||||
// whose output is actually presented, so a request cannot be answered early. The last
|
||||
// pass of a frame is presented even when its own discard request was refused because
|
||||
// `max_passes` is reached; egui still reports that request in the output.
|
||||
let max_passes = ctx.options(|o| o.max_passes.get());
|
||||
let final_pass = output.platform_output.num_completed_passes >= max_passes;
|
||||
if output.platform_output.request_discard_reasons.is_empty() || final_pass {
|
||||
s.passes += 1;
|
||||
}
|
||||
let Some(update) = &output.platform_output.accesskit_update else { return };
|
||||
s.nodes = update.nodes.iter().cloned().collect();
|
||||
s.root = update.tree.as_ref().map(|t| t.root);
|
||||
s.focus = Some(update.focus);
|
||||
}
|
||||
}
|
||||
|
||||
impl Control {
|
||||
/// Queue one request's event batches (one per UI pass) and return its request id.
|
||||
///
|
||||
/// The request is stamped once, at admission: all of its batches share the input-target
|
||||
/// generation and the caller's single arrival deadline (`since`), so a deferred startup wait
|
||||
/// is not restarted here. A poisoned mutex or a full queue is an error, never a silently empty
|
||||
/// queue that would let the caller answer `Ok` for input that never happened.
|
||||
pub(super) fn inject(&self, batches: Vec<Vec<egui::Event>>, since: web_time::Instant) -> Result<Vec<u64>, String> {
|
||||
let mut s = self.shared.lock().map_err(|_| "control state poisoned".to_string())?;
|
||||
let queued: usize = s.inject.iter().map(|request| request.batches.len()).sum();
|
||||
if queued.saturating_add(batches.len()) > MAX_QUEUED_BATCHES {
|
||||
return Err(format!("too many synthetic input batches are already queued (limit {MAX_QUEUED_BATCHES})"));
|
||||
}
|
||||
s.next_batch += 1;
|
||||
let id = s.next_batch;
|
||||
let epoch = s.epoch;
|
||||
s.inject.push_back(RequestBatches { request: id, epoch, started: false, since, release: Vec::new(), batches: batches.into() });
|
||||
Ok(vec![id])
|
||||
}
|
||||
|
||||
/// Queue synthetic input batches and reply only after the requested number of frames has been presented.
|
||||
pub(super) fn after_input(&self, batches: Vec<Vec<egui::Event>>, value: Value, since: web_time::Instant) -> Result<Handled, String> {
|
||||
let frames = batches.len() as u32;
|
||||
let batches = self.inject(batches, since)?;
|
||||
Ok(Handled::AfterInput { batches, frames, value })
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,369 @@
|
||||
use super::screenshot::Geometry;
|
||||
use super::*;
|
||||
|
||||
impl Control {
|
||||
/// Validate and dispatch one control method, rejecting unknown methods and malformed parameters with
|
||||
/// actionable errors.
|
||||
pub(super) fn handle(&mut self, ctx: &egui::Context, host: &mut impl Host, method: &str, p: &Value, since: web_time::Instant) -> Handled {
|
||||
let str_param = |k: &str| p.get(k).and_then(Value::as_str).ok_or_else(|| format!("{method}: missing string parameter {k}"));
|
||||
let r: Result<Handled, String> = (|| match method {
|
||||
"ui.state" => {
|
||||
let mut state = host.state();
|
||||
// Viewport info is fresh in `logic`; the last UI pass's rectangle can be stale.
|
||||
let geometry = Geometry::current(ctx, host.capture_epoch());
|
||||
let size = geometry.size;
|
||||
state["window"] = json!({ "width": size.x, "height": size.y, "pixels_per_point": geometry.pixels_per_point });
|
||||
Ok(Handled::Now(Ok(state)))
|
||||
}
|
||||
"ui.commands" => Ok(Handled::Now(Ok(host.commands()))),
|
||||
"ui.panel" | "ui.dock.reset" => Ok(Handled::Now(host.panel(method == "ui.dock.reset", p))),
|
||||
"ui.command" if str_param("id")? == "ui.panel" => Ok(Handled::Now(host.panel(false, p.get("params").unwrap_or(p)))),
|
||||
"ui.command" => Ok(Handled::Now(host.command(str_param("id")?))),
|
||||
"ui.set" => {
|
||||
let value = match p.get("value") {
|
||||
Some(Value::String(s)) => s.clone(),
|
||||
Some(v) => v.to_string(),
|
||||
None => return Err("ui.set: missing parameter value".into()),
|
||||
};
|
||||
Ok(Handled::Now(host.set(str_param("key")?, &value)))
|
||||
}
|
||||
"ui.open" => Ok(Handled::Now(host.open(str_param("path")?))),
|
||||
"ui.inspect" => Ok(Handled::Now(Ok(self.inspect(p)))),
|
||||
"ui.click" => self.click(p, since),
|
||||
"ui.focus" => self.focus(p, since),
|
||||
"ui.resize" => {
|
||||
only_params(p, method, &["width", "height"])?;
|
||||
let dimension = |key: &str| -> Result<f32, String> {
|
||||
p.get(key)
|
||||
.and_then(Value::as_f64)
|
||||
.filter(|n| n.is_finite() && (320.0..=8192.0).contains(n))
|
||||
.map(|n| n as f32)
|
||||
.ok_or_else(|| format!("ui.resize: {key} must be between 320 and 8192 logical points"))
|
||||
};
|
||||
let size = egui::vec2(dimension("width")?, dimension("height")?);
|
||||
if cfg!(target_arch = "wasm32") {
|
||||
return Err("ui.resize needs a native window".into());
|
||||
}
|
||||
ctx.send_viewport_cmd(egui::ViewportCommand::InnerSize(size));
|
||||
ctx.request_repaint();
|
||||
Ok(Handled::AfterFrames(2, json!({ "requested": [size.x, size.y] })))
|
||||
}
|
||||
"ui.quit" => {
|
||||
only_params(p, method, &[])?;
|
||||
if cfg!(target_arch = "wasm32") {
|
||||
return Err("ui.quit needs a native window".into());
|
||||
}
|
||||
let quitting = host.quit(ctx);
|
||||
Ok(Handled::Now(Ok(json!({ "quitting": quitting, "needs_confirmation": !quitting }))))
|
||||
}
|
||||
"ui.drag" => self.drag(p, since),
|
||||
"ui.move" => {
|
||||
let point = |key: &str| -> Result<f32, String> {
|
||||
p.get(key)
|
||||
.and_then(Value::as_f64)
|
||||
.map(|n| n as f32)
|
||||
.filter(|n| n.is_finite())
|
||||
.ok_or_else(|| format!("ui.move: {key} must be a finite coordinate in points"))
|
||||
};
|
||||
let pos = egui::pos2(point("x")?, point("y")?);
|
||||
self.after_input(vec![vec![egui::Event::PointerMoved(pos)]], json!({ "moved": [pos.x, pos.y] }), since)
|
||||
}
|
||||
"ui.type" => {
|
||||
let text = str_param("text")?.to_string();
|
||||
self.after_input(vec![vec![egui::Event::Text(text)]], json!({ "typed": true }), since)
|
||||
}
|
||||
"ui.key" => {
|
||||
let name = str_param("key")?;
|
||||
let key = egui::Key::from_name(name)
|
||||
.ok_or_else(|| format!("ui.key: unknown key {name:?} (egui key names: A, Enter, Escape, ArrowDown, F5, …)"))?;
|
||||
let modifiers = modifiers(p.get("modifiers"))?;
|
||||
self.after_input(key_batches(key, modifiers), json!({ "key": name }), since)
|
||||
}
|
||||
"ui.screenshot" => {
|
||||
let region = match p.get("region") {
|
||||
None | Some(Value::Null) => None,
|
||||
Some(r) => Some(rect_param(r)?),
|
||||
};
|
||||
ctx.request_repaint();
|
||||
Ok(Handled::Screenshot(region))
|
||||
}
|
||||
other => Err(format!(
|
||||
"unknown method {other:?} (ui.state, ui.inspect, ui.click, ui.focus, ui.resize, ui.quit, ui.move, ui.drag, ui.type, ui.key, ui.command, ui.commands, ui.set, ui.open, ui.screenshot)"
|
||||
)),
|
||||
})();
|
||||
r.unwrap_or_else(|e| Handled::Now(Err(e)))
|
||||
}
|
||||
|
||||
fn inspect(&self, p: &Value) -> Value {
|
||||
let query = p.get("query").and_then(Value::as_str).map(str::to_lowercase);
|
||||
let role = p.get("role").and_then(Value::as_str).map(str::to_lowercase);
|
||||
let focused = p.get("focused").and_then(Value::as_bool);
|
||||
let limit = p.get("limit").and_then(Value::as_u64).unwrap_or(500) as usize;
|
||||
let Ok(s) = self.shared.lock() else { return json!({ "widgets": [] }) };
|
||||
let mut out = Vec::new();
|
||||
let mut total = 0usize;
|
||||
let mut stack: Vec<(NodeId, usize)> = s.root.map(|r| vec![(r, 0)]).unwrap_or_default();
|
||||
while let Some((id, depth)) = stack.pop() {
|
||||
let Some(node) = s.nodes.get(&id) else { continue };
|
||||
for c in node.children().iter().rev() {
|
||||
stack.push((*c, depth + 1));
|
||||
}
|
||||
let w = widget(id, node, depth, s.focus == Some(id));
|
||||
let text =
|
||||
format!("{} {} {}", w["label"].as_str().unwrap_or(""), w["value"].as_str().unwrap_or(""), w["description"].as_str().unwrap_or(""))
|
||||
.to_lowercase();
|
||||
let role_ok = role.as_ref().is_none_or(|r| w["role"].as_str().is_some_and(|x| x.to_lowercase() == *r));
|
||||
let query_ok = query.as_ref().is_none_or(|q| text.contains(q.as_str()));
|
||||
let focus_ok = focused.is_none_or(|want| (s.focus == Some(id)) == want);
|
||||
if role_ok && query_ok && focus_ok && id != s.root.unwrap_or(id) {
|
||||
total += 1;
|
||||
if out.len() < limit {
|
||||
out.push(w);
|
||||
}
|
||||
}
|
||||
}
|
||||
json!({ "widgets": out, "count": total, "truncated": total > out.len() })
|
||||
}
|
||||
|
||||
/// Press at `from`, move to `to` in `steps` frames, release (drawing, selecting text, moving).
|
||||
fn drag(&mut self, p: &Value, since: web_time::Instant) -> Result<Handled, String> {
|
||||
let point = |k: &str| -> Result<egui::Pos2, String> {
|
||||
match p.get(k).and_then(Value::as_array).map(|a| a.iter().filter_map(Value::as_f64).collect::<Vec<_>>()).as_deref() {
|
||||
Some([x, y]) => Some(egui::pos2(*x as f32, *y as f32))
|
||||
.filter(|pos| pos.is_finite())
|
||||
.ok_or_else(|| format!("ui.drag: {k} must be [x, y], finite coordinates in points")),
|
||||
_ => Err(format!("ui.drag: {k} must be [x, y] in points")),
|
||||
}
|
||||
};
|
||||
let (from, to) = (point("from")?, point("to")?);
|
||||
if !(to - from).is_finite() {
|
||||
return Err("ui.drag: from and to must be a finite distance apart, in points".into());
|
||||
}
|
||||
let steps = p.get("steps").and_then(Value::as_u64).unwrap_or(8).clamp(1, 200) as usize;
|
||||
let modifiers = modifiers(p.get("modifiers"))?;
|
||||
let which = pointer_button(p, "ui.drag")?;
|
||||
// The events carry the modifiers and so does the input state: gestures that read
|
||||
// `ui.input(|i| i.modifiers)` (a Ctrl-drag copy, an Alt-drag column select) see them.
|
||||
let held = |mut events: Vec<egui::Event>| {
|
||||
if modifiers != egui::Modifiers::NONE {
|
||||
events.insert(0, egui::Event::ModifiersChanged(modifiers));
|
||||
}
|
||||
events
|
||||
};
|
||||
let button = |pos, pressed| egui::Event::PointerButton { pos, button: which, pressed, modifiers };
|
||||
let mut frames = vec![held(vec![egui::Event::PointerMoved(from)]), held(vec![button(from, true)])];
|
||||
for k in 1..=steps {
|
||||
frames.push(held(vec![egui::Event::PointerMoved(from + (to - from) * (k as f32 / steps as f32))]));
|
||||
}
|
||||
frames.push(held(vec![button(to, false)]));
|
||||
if modifiers != egui::Modifiers::NONE {
|
||||
frames.push(vec![egui::Event::ModifiersChanged(egui::Modifiers::NONE)]);
|
||||
}
|
||||
self.after_input(frames, json!({ "dragged": [[from.x, from.y], [to.x, to.y]] }), since)
|
||||
}
|
||||
|
||||
/// Click at a point, or on an accesskit widget by id or unique enabled label.
|
||||
fn click(&mut self, p: &Value, since: web_time::Instant) -> Result<Handled, String> {
|
||||
if let (Some(x), Some(y)) = (p.get("x").and_then(Value::as_f64), p.get("y").and_then(Value::as_f64)) {
|
||||
let point = |key: &str, n: f64| {
|
||||
Some(n as f32).filter(|n| n.is_finite()).ok_or_else(|| format!("ui.click: {key} must be a finite coordinate in points"))
|
||||
};
|
||||
let pos = egui::pos2(point("x", x)?, point("y", y)?);
|
||||
let which = pointer_button(p, "ui.click")?;
|
||||
let button = |pressed| egui::Event::PointerButton { pos, button: which, pressed, modifiers: egui::Modifiers::NONE };
|
||||
// Keep hover/layout warmup, then make an instantaneous click even on slow frames.
|
||||
let frames = vec![vec![egui::Event::PointerMoved(pos)], vec![button(true), button(false)]];
|
||||
return self.after_input(frames, json!({ "clicked": [x, y] }), since);
|
||||
}
|
||||
let s = self.shared.lock().map_err(|_| "control state poisoned")?;
|
||||
let id = match (p.get("id"), p.get("label").and_then(Value::as_str)) {
|
||||
(Some(id), _) => {
|
||||
let id: u64 =
|
||||
id.as_u64().or_else(|| id.as_str().and_then(|s| s.parse().ok())).ok_or("ui.click: id must be a widget id from ui.inspect")?;
|
||||
let id = NodeId(id);
|
||||
if !s.nodes.contains_key(&id) {
|
||||
return Err(format!("ui.click: no widget with id {} on screen (run ui.inspect again)", id.0));
|
||||
}
|
||||
id
|
||||
}
|
||||
(None, Some(label)) => {
|
||||
let want = label.to_lowercase();
|
||||
let matches: Vec<NodeId> = s
|
||||
.nodes
|
||||
.iter()
|
||||
.filter(|(_, n)| n.supports_action(Action::Click) && !n.is_disabled() && n.label().is_some_and(|l| l.to_lowercase() == want))
|
||||
.map(|(id, _)| *id)
|
||||
.collect();
|
||||
match matches.as_slice() {
|
||||
[one] => *one,
|
||||
[] => return Err(format!("ui.click: no enabled clickable widget labelled {label:?} (try ui.inspect with query)")),
|
||||
many => return Err(format!("ui.click: {} widgets are labelled {label:?}; click one by id", many.len())),
|
||||
}
|
||||
}
|
||||
(None, None) => return Err("ui.click: pass id, label, or x and y".into()),
|
||||
};
|
||||
let node = &s.nodes[&id];
|
||||
if node.is_disabled() {
|
||||
return Err(format!("ui.click: widget {} ({}) is disabled", id.0, node.label().unwrap_or("unlabelled")));
|
||||
}
|
||||
let label = node.label().map(str::to_owned);
|
||||
let action = accesskit::ActionRequest { action: Action::Click, target_tree: accesskit::TreeId::ROOT, target_node: id, data: None };
|
||||
drop(s);
|
||||
let frames = vec![vec![egui::Event::AccessKitActionRequest(action)]];
|
||||
self.after_input(frames, json!({ "clicked": id.0.to_string(), "label": label }), since)
|
||||
}
|
||||
|
||||
/// Focus an accesskit widget by id or unique enabled label.
|
||||
fn focus(&mut self, p: &Value, since: web_time::Instant) -> Result<Handled, String> {
|
||||
only_params(p, "ui.focus", &["id", "label"])?;
|
||||
let s = self.shared.lock().map_err(|_| "control state poisoned")?;
|
||||
let id = match (p.get("id"), p.get("label")) {
|
||||
(Some(id), None) => {
|
||||
NodeId(id.as_u64().or_else(|| id.as_str().and_then(|s| s.parse().ok())).ok_or("ui.focus: id must be a widget id from ui.inspect")?)
|
||||
}
|
||||
(None, Some(Value::String(label))) => {
|
||||
let matches: Vec<NodeId> = s
|
||||
.nodes
|
||||
.iter()
|
||||
.filter(|(_, n)| n.supports_action(Action::Focus) && !n.is_disabled() && n.label().is_some_and(|l| l.eq_ignore_ascii_case(label)))
|
||||
.map(|(id, _)| *id)
|
||||
.collect();
|
||||
match matches.as_slice() {
|
||||
[one] => *one,
|
||||
[] => return Err(format!("ui.focus: no enabled focusable widget labelled {label:?}")),
|
||||
many => return Err(format!("ui.focus: {} widgets are labelled {label:?}; use a widget id", many.len())),
|
||||
}
|
||||
}
|
||||
_ => return Err("ui.focus: pass exactly one widget id or label".into()),
|
||||
};
|
||||
let node = s.nodes.get(&id).ok_or_else(|| format!("ui.focus: no widget with id {} on screen (run ui.inspect again)", id.0))?;
|
||||
if node.is_disabled() || !node.supports_action(Action::Focus) {
|
||||
return Err(format!("ui.focus: widget {} is disabled or not focusable", id.0));
|
||||
}
|
||||
let action = accesskit::ActionRequest { action: Action::Focus, target_tree: accesskit::TreeId::ROOT, target_node: id, data: None };
|
||||
drop(s);
|
||||
let frames = vec![vec![egui::Event::AccessKitActionRequest(action)]];
|
||||
self.after_input(frames, json!({ "focus_requested": id.0.to_string() }), since)
|
||||
}
|
||||
}
|
||||
|
||||
pub(super) enum Handled {
|
||||
Now(Reply),
|
||||
/// Answer after this many accepted UI passes, without injecting input.
|
||||
AfterFrames(u32, Value),
|
||||
AfterInput {
|
||||
batches: Vec<u64>,
|
||||
frames: u32,
|
||||
value: Value,
|
||||
},
|
||||
Screenshot(Option<egui::Rect>),
|
||||
}
|
||||
|
||||
fn widget(id: NodeId, n: &accesskit::Node, depth: usize, focused: bool) -> Value {
|
||||
let rect = n.bounds().map(|b| [b.x0, b.y0, b.x1, b.y1]);
|
||||
let mut w = json!({
|
||||
// A string: 64-bit ids don't survive JSON parsers that use doubles.
|
||||
"id": id.0.to_string(),
|
||||
"role": format!("{:?}", n.role()),
|
||||
"depth": depth,
|
||||
"enabled": !n.is_disabled(),
|
||||
"clickable": n.supports_action(Action::Click),
|
||||
"focusable": n.supports_action(Action::Focus),
|
||||
});
|
||||
let mut more = serde_json::Map::new();
|
||||
let obj = &mut more;
|
||||
if let Some(l) = n.label() {
|
||||
obj.insert("label".into(), json!(l));
|
||||
}
|
||||
if let Some(v) = n.value() {
|
||||
obj.insert("value".into(), json!(v));
|
||||
}
|
||||
if let Some(d) = n.description() {
|
||||
obj.insert("description".into(), json!(d));
|
||||
}
|
||||
if let Some(r) = rect {
|
||||
obj.insert("rect".into(), json!(r));
|
||||
}
|
||||
if let Some(t) = n.toggled() {
|
||||
obj.insert("toggled".into(), json!(format!("{t:?}").to_lowercase()));
|
||||
}
|
||||
if let Some(s) = n.is_selected() {
|
||||
obj.insert("selected".into(), json!(s));
|
||||
}
|
||||
if focused {
|
||||
obj.insert("focused".into(), json!(true));
|
||||
}
|
||||
if let Value::Object(m) = &mut w {
|
||||
m.extend(more);
|
||||
}
|
||||
w
|
||||
}
|
||||
|
||||
fn only_params(p: &Value, method: &str, allowed: &[&str]) -> Result<(), String> {
|
||||
// The JSON-RPC transport represents omitted params as null.
|
||||
if p.is_null() && allowed.is_empty() {
|
||||
return Ok(());
|
||||
}
|
||||
let object = p.as_object().ok_or_else(|| format!("{method}: params must be an object"))?;
|
||||
if let Some(key) = object.keys().find(|key| !allowed.contains(&key.as_str())) {
|
||||
return Err(format!("{method}: unknown parameter {key:?}"));
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn pointer_button(p: &Value, method: &str) -> Result<egui::PointerButton, String> {
|
||||
match p.get("button").and_then(Value::as_str).unwrap_or("primary") {
|
||||
"primary" | "left" => Ok(egui::PointerButton::Primary),
|
||||
"secondary" | "right" => Ok(egui::PointerButton::Secondary),
|
||||
"middle" => Ok(egui::PointerButton::Middle),
|
||||
other => Err(format!("{method}: unknown button {other:?} (primary, secondary, middle)")),
|
||||
}
|
||||
}
|
||||
|
||||
/// The events for one synthetic key press. egui updates `InputState.modifiers` only from
|
||||
/// `ModifiersChanged`, not from the `modifiers` field on a key event, so the held set goes in
|
||||
/// before the press and is released after the key-up. Without this, handlers that read
|
||||
/// `i.modifiers` (Shift+Tab, ⌘Enter) would see stale state.
|
||||
pub(super) fn key_batches(key: egui::Key, modifiers: egui::Modifiers) -> Vec<Vec<egui::Event>> {
|
||||
let key_event = |pressed| egui::Event::Key { key, physical_key: None, pressed, repeat: false, modifiers };
|
||||
let mut press = vec![key_event(true)];
|
||||
if modifiers != egui::Modifiers::NONE {
|
||||
press.insert(0, egui::Event::ModifiersChanged(modifiers));
|
||||
}
|
||||
let mut release = vec![key_event(false)];
|
||||
if modifiers != egui::Modifiers::NONE {
|
||||
release.push(egui::Event::ModifiersChanged(egui::Modifiers::NONE));
|
||||
}
|
||||
vec![press, release]
|
||||
}
|
||||
|
||||
fn modifiers(v: Option<&Value>) -> Result<egui::Modifiers, String> {
|
||||
let mut m = egui::Modifiers::NONE;
|
||||
for name in v.and_then(Value::as_array).into_iter().flatten() {
|
||||
match name.as_str().unwrap_or("") {
|
||||
"command" | "cmd" => m.command = true,
|
||||
"shift" => m.shift = true,
|
||||
"alt" | "option" => m.alt = true,
|
||||
"ctrl" | "control" => m.ctrl = true,
|
||||
"mac_cmd" => m.mac_cmd = true,
|
||||
other => return Err(format!("ui.key: unknown modifier {other:?} (command, shift, alt, ctrl)")),
|
||||
}
|
||||
}
|
||||
// "command" is ⌘ on macOS and Ctrl elsewhere, as egui expects.
|
||||
if m.command {
|
||||
if cfg!(target_os = "macos") {
|
||||
m.mac_cmd = true;
|
||||
} else {
|
||||
m.ctrl = true;
|
||||
}
|
||||
}
|
||||
Ok(m)
|
||||
}
|
||||
|
||||
fn rect_param(v: &Value) -> Result<egui::Rect, String> {
|
||||
let a: Vec<f32> = v.as_array().map(|a| a.iter().filter_map(Value::as_f64).map(|f| f as f32).collect()).unwrap_or_default();
|
||||
match a.as_slice() {
|
||||
[x0, y0, x1, y1] if x1 > x0 && y1 > y0 => Ok(egui::Rect::from_min_max(egui::pos2(*x0, *y0), egui::pos2(*x1, *y1))),
|
||||
_ => Err("region must be [x0, y0, x1, y1] in points".into()),
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,284 @@
|
||||
use super::*;
|
||||
|
||||
/// Screenshot admission limits: bounded requests, per-image bytes and aggregate retained bytes.
|
||||
const MAX_PENDING_SCREENSHOTS: usize = 4;
|
||||
|
||||
pub(super) const MAX_IMAGE_BYTES: usize = 128 * 1024 * 1024;
|
||||
|
||||
const MAX_TOTAL_BYTES: usize = 256 * 1024 * 1024;
|
||||
|
||||
/// One image's encode working copies are bounded too: the RGBA buffer fed to the PNG encoder,
|
||||
/// the encoded PNG, and the base64 string in the reply. They are made one capture at a time,
|
||||
/// after the retained image budget above has been checked. Peak encoding memory is therefore at
|
||||
/// most the retained aggregate plus one crop, one PNG and one base64 string; it is not hidden
|
||||
/// behind the retained-image limit.
|
||||
const MAX_ENCODED_BYTES: usize = MAX_IMAGE_BYTES;
|
||||
|
||||
const MAX_BASE64_BYTES: usize = 192 * 1024 * 1024;
|
||||
|
||||
/// Replace one pending capture's reservation with `bytes`, keeping the other reservations.
|
||||
///
|
||||
/// The drain retries captures one at a time: using a single total taken before the loop would
|
||||
/// let several retries each see room that no longer exists together, and the aggregate would
|
||||
/// grow past [`MAX_TOTAL_BYTES`].
|
||||
pub(super) fn retried_reservation(total: usize, old: usize, bytes: usize) -> Option<usize> {
|
||||
let without_this = total.saturating_sub(old);
|
||||
let new_total = without_this.saturating_add(bytes);
|
||||
(bytes <= MAX_IMAGE_BYTES && new_total <= MAX_TOTAL_BYTES).then_some(new_total)
|
||||
}
|
||||
|
||||
/// The window content a screenshot request was admitted and issued for.
|
||||
///
|
||||
/// Only the viewport's local size and scale matter: the window's position on the monitor does not
|
||||
/// change a single captured pixel, and keyboard focus is not part of the captured content.
|
||||
#[derive(Clone, Copy, Debug, PartialEq)]
|
||||
pub(super) struct Geometry {
|
||||
pub(super) size: egui::Vec2,
|
||||
pub(super) pixels_per_point: f32,
|
||||
/// Active document and open dialog/prompt identity, never the focused widget.
|
||||
pub(super) epoch: u64,
|
||||
}
|
||||
|
||||
impl Geometry {
|
||||
/// Capture the current viewport size and effective pixels-per-point for the request epoch.
|
||||
pub(super) fn current(ctx: &egui::Context, epoch: u64) -> Self {
|
||||
// Viewport info is refreshed by `run_logic`; the last UI pass's `viewport_rect` can be
|
||||
// stale while the window is hidden. Fall back to it only when the host cannot say. Only
|
||||
// its size is kept: `inner_rect` is in monitor space and its origin may move.
|
||||
let (rect, native) = ctx.input(|i| {
|
||||
let viewport = i.viewport();
|
||||
(viewport.inner_rect.unwrap_or_else(|| i.viewport_rect()), viewport.native_pixels_per_point)
|
||||
});
|
||||
let native = native.or_else(|| ctx.native_pixels_per_point()).unwrap_or(1.0);
|
||||
Self { size: rect.size(), pixels_per_point: native * ctx.zoom_factor(), epoch }
|
||||
}
|
||||
|
||||
/// Checked bytes an image of this geometry needs, before the capture is issued.
|
||||
pub(super) fn image_bytes(self) -> Option<usize> {
|
||||
let width = (self.size.x * self.pixels_per_point).ceil();
|
||||
let height = (self.size.y * self.pixels_per_point).ceil();
|
||||
if !(width.is_finite() && height.is_finite()) || width <= 0.0 || height <= 0.0 {
|
||||
return None;
|
||||
}
|
||||
checked_image_bytes(width as usize, height as usize)
|
||||
}
|
||||
|
||||
/// Whether `region` is a non-empty rectangle inside this viewport, in logical points.
|
||||
pub(super) fn contains_region(self, region: egui::Rect) -> bool {
|
||||
region.min.x.is_finite()
|
||||
&& region.min.y.is_finite()
|
||||
&& region.max.x.is_finite()
|
||||
&& region.max.y.is_finite()
|
||||
&& region.min.x >= 0.0
|
||||
&& region.min.y >= 0.0
|
||||
&& region.max.x > region.min.x
|
||||
&& region.max.y > region.min.y
|
||||
&& region.max.x <= self.size.x
|
||||
&& region.max.y <= self.size.y
|
||||
}
|
||||
}
|
||||
|
||||
/// Return the byte size of an RGBA image of the given dimensions, or `None` when the multiplication
|
||||
/// overflows.
|
||||
fn checked_image_bytes(width: usize, height: usize) -> Option<usize> {
|
||||
width.checked_mul(height)?.checked_mul(core::mem::size_of::<egui::Color32>())
|
||||
}
|
||||
|
||||
impl Control {
|
||||
/// Return the total screenshot bytes reserved by pending captures without overflowing.
|
||||
pub(super) fn reserved_bytes(&self) -> usize {
|
||||
self.pending
|
||||
.iter()
|
||||
.filter_map(|p| if let Pending::Screenshot { reserved, .. } = p { Some(*reserved) } else { None })
|
||||
.fold(0usize, usize::saturating_add)
|
||||
}
|
||||
|
||||
/// Count the screenshot requests still waiting for a capture.
|
||||
fn pending_screenshots(&self) -> usize {
|
||||
self.pending.iter().filter(|p| matches!(p, Pending::Screenshot { .. })).count()
|
||||
}
|
||||
|
||||
/// Admit one screenshot request within the count, per-image and aggregate limits.
|
||||
pub(super) fn admit_screenshot(
|
||||
&mut self,
|
||||
ctx: &egui::Context,
|
||||
host: &mut impl Host,
|
||||
region: Option<egui::Rect>,
|
||||
reply: Sender<Reply>,
|
||||
since: web_time::Instant,
|
||||
) {
|
||||
if self.pending_screenshots() >= MAX_PENDING_SCREENSHOTS {
|
||||
let _ = reply.send(Err(format!("too many screenshots are already pending (limit {MAX_PENDING_SCREENSHOTS})")));
|
||||
return;
|
||||
}
|
||||
let geometry = Geometry::current(ctx, host.capture_epoch());
|
||||
let Some(bytes) = geometry.image_bytes() else {
|
||||
let _ = reply.send(Err("the window has no capturable size".into()));
|
||||
return;
|
||||
};
|
||||
if let Some(region) = region
|
||||
&& !geometry.contains_region(region)
|
||||
{
|
||||
let _ = reply.send(Err("the region must lie inside the window".into()));
|
||||
return;
|
||||
}
|
||||
if bytes > MAX_IMAGE_BYTES {
|
||||
let _ = reply.send(Err(format!("the screenshot would use {bytes} bytes, over the {MAX_IMAGE_BYTES}-byte image limit")));
|
||||
return;
|
||||
}
|
||||
if self.reserved_bytes().saturating_add(bytes) > MAX_TOTAL_BYTES {
|
||||
let _ = reply.send(Err(format!("the screenshot would exceed the {MAX_TOTAL_BYTES}-byte aggregate screenshot budget")));
|
||||
return;
|
||||
}
|
||||
let (slot, callback) = craft_ui::capture::CaptureSlot::new(ctx, egui::ViewportId::ROOT, MAX_IMAGE_BYTES);
|
||||
self.pending.push(Pending::Screenshot { slot, callback: Some(callback), region, reply, since, geometry, reserved: bytes, attempts: 0 });
|
||||
}
|
||||
}
|
||||
|
||||
/// Encode (a region of) a screenshot as PNG; the reply carries base64 data and the size.
|
||||
///
|
||||
/// The image scale is derived from the returned pixels and the logical viewport the capture was
|
||||
/// admitted for, never from the current `Context`: egui-wgpu's surface fit may have shrunk the
|
||||
/// surface (and a resize/DPI change has already been rejected before this call).
|
||||
pub(super) fn screenshot_png(image: &egui::ColorImage, region: Option<egui::Rect>, geometry: Geometry) -> Reply {
|
||||
let [w, h] = image.size;
|
||||
let view = geometry.size;
|
||||
if w == 0 || h == 0 {
|
||||
return Err("the screenshot has no pixels".into());
|
||||
}
|
||||
if !(view.x.is_finite() && view.y.is_finite()) || view.x <= 0.0 || view.y <= 0.0 {
|
||||
return Err("the window has no capturable size".into());
|
||||
}
|
||||
if region.is_some_and(|region| !geometry.contains_region(region)) {
|
||||
return Err("the region must lie inside the window".into());
|
||||
}
|
||||
let scale_x = w as f32 / view.x;
|
||||
let scale_y = h as f32 / view.y;
|
||||
// A fitted surface shrinks both sides by one factor; a different x/y scale means the window
|
||||
// changed shape and this image cannot be cropped honestly.
|
||||
if (scale_x - scale_y).abs() > 0.01 * scale_x.max(1.0) {
|
||||
return Err("the screenshot's aspect ratio does not match the window".into());
|
||||
}
|
||||
let (x0, y0, x1, y1) = match region {
|
||||
Some(r) => {
|
||||
let px = |v: f32, scale: f32, max: usize| ((v * scale).round().max(0.0) as usize).min(max);
|
||||
(px(r.min.x, scale_x, w), px(r.min.y, scale_y, h), px(r.max.x, scale_x, w), px(r.max.y, scale_y, h))
|
||||
}
|
||||
None => (0, 0, w, h),
|
||||
};
|
||||
if x1 <= x0 || y1 <= y0 {
|
||||
return Err("the region is outside the window".into());
|
||||
}
|
||||
let crop_bytes = (x1 - x0)
|
||||
.checked_mul(y1 - y0)
|
||||
.and_then(|pixels| pixels.checked_mul(core::mem::size_of::<egui::Color32>()))
|
||||
.ok_or_else(|| "the screenshot region is too large".to_string())?;
|
||||
if crop_bytes > MAX_ENCODED_BYTES {
|
||||
return Err(format!("the screenshot region would need {crop_bytes} bytes to encode, over the {MAX_ENCODED_BYTES}-byte encode limit"));
|
||||
}
|
||||
let mut rgba = Vec::with_capacity(crop_bytes);
|
||||
for y in y0..y1 {
|
||||
for px in &image.pixels[y * w + x0..y * w + x1] {
|
||||
rgba.extend_from_slice(&px.to_srgba_unmultiplied());
|
||||
}
|
||||
}
|
||||
let (cw, ch) = ((x1 - x0) as u32, (y1 - y0) as u32);
|
||||
let mut out = Vec::new();
|
||||
let mut enc = png::Encoder::new(&mut out, cw, ch);
|
||||
enc.set_color(png::ColorType::Rgba);
|
||||
enc.set_depth(png::BitDepth::Eight);
|
||||
let mut writer = enc.write_header().map_err(|e| e.to_string())?;
|
||||
writer.write_image_data(&rgba).map_err(|e| e.to_string())?;
|
||||
writer.finish().map_err(|e| e.to_string())?;
|
||||
if out.len() > MAX_ENCODED_BYTES {
|
||||
return Err(format!("the encoded screenshot is {} bytes, over the {MAX_ENCODED_BYTES}-byte encode limit", out.len()));
|
||||
}
|
||||
let encoded = out.len().div_ceil(3).checked_mul(4).ok_or_else(|| "the encoded screenshot is too large".to_string())?;
|
||||
if encoded > MAX_BASE64_BYTES {
|
||||
return Err(format!("the screenshot's base64 reply would be {encoded} bytes, over the {MAX_BASE64_BYTES}-byte limit"));
|
||||
}
|
||||
use base64::Engine as _;
|
||||
Ok(json!({
|
||||
"png_base64": base64::engine::general_purpose::STANDARD.encode(out),
|
||||
"width": cw,
|
||||
"height": ch,
|
||||
"pixels_per_point": scale_x,
|
||||
}))
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod screenshot_geometry_tests {
|
||||
use super::*;
|
||||
|
||||
#[test]
|
||||
fn image_bytes_are_checked_against_overflow_and_limits() {
|
||||
assert_eq!(checked_image_bytes(2, 3), Some(24));
|
||||
assert_eq!(checked_image_bytes(usize::MAX, 2), None);
|
||||
let big = Geometry { size: egui::vec2(10_000.0, 10_000.0), pixels_per_point: 8.0, epoch: 0 };
|
||||
assert!(big.image_bytes().is_some_and(|bytes| bytes > MAX_IMAGE_BYTES));
|
||||
}
|
||||
|
||||
/// The crop is derived from the returned pixels and the admitted logical rect, not from a
|
||||
/// current `pixels_per_point`: a fitted surface may return a smaller image.
|
||||
#[test]
|
||||
fn region_crop_uses_the_capture_geometry() {
|
||||
let image = egui::ColorImage::filled([200, 100], egui::Color32::RED);
|
||||
let geometry = Geometry { size: egui::vec2(100.0, 50.0), pixels_per_point: 2.0, epoch: 7 };
|
||||
let reply =
|
||||
screenshot_png(&image, Some(egui::Rect::from_min_max(egui::pos2(10.0, 5.0), egui::pos2(60.0, 30.0))), geometry).expect("a valid crop");
|
||||
assert_eq!(reply["width"], 100);
|
||||
assert_eq!(reply["height"], 50);
|
||||
assert_eq!(reply["pixels_per_point"], 2.0);
|
||||
}
|
||||
|
||||
/// A retry replaces one reservation in a running aggregate. Three 50 MiB captures that each
|
||||
/// retry to 100 MiB must not all be admitted against a stale 150 MiB snapshot.
|
||||
#[test]
|
||||
fn retries_keep_one_running_aggregate_budget() {
|
||||
const MIB: usize = 1024 * 1024;
|
||||
let first = retried_reservation(3 * 50 * MIB, 50 * MIB, 100 * MIB).expect("the first retry fits");
|
||||
let second = retried_reservation(first, 50 * MIB, 100 * MIB).expect("the second retry fits");
|
||||
assert_eq!(second, 250 * MIB);
|
||||
assert_eq!(retried_reservation(second, 50 * MIB, 100 * MIB), None, "the third would pass 256 MiB");
|
||||
assert_eq!(retried_reservation(0, 0, MAX_IMAGE_BYTES + 1), None, "a single image over the per-image limit never fits");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn an_image_with_another_aspect_ratio_is_rejected() {
|
||||
let image = egui::ColorImage::filled([200, 100], egui::Color32::RED);
|
||||
let geometry = Geometry { size: egui::vec2(100.0, 60.0), pixels_per_point: 2.0, epoch: 0 };
|
||||
assert!(screenshot_png(&image, None, geometry).is_err());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn an_outside_region_is_rejected_not_panicking() {
|
||||
let image = egui::ColorImage::filled([20, 10], egui::Color32::RED);
|
||||
let geometry = Geometry { size: egui::vec2(10.0, 5.0), pixels_per_point: 2.0, epoch: 0 };
|
||||
assert!(screenshot_png(&image, Some(egui::Rect::from_min_max(egui::pos2(11.0, 6.0), egui::pos2(15.0, 9.0))), geometry).is_err());
|
||||
}
|
||||
|
||||
/// A region that a shrink now cuts off is refused, not silently clamped to a smaller crop.
|
||||
#[test]
|
||||
fn a_region_cut_by_a_shrink_is_rejected_instead_of_cropped() {
|
||||
let image = egui::ColorImage::filled([20, 10], egui::Color32::RED);
|
||||
let shrunk = Geometry { size: egui::vec2(10.0, 5.0), pixels_per_point: 2.0, epoch: 0 };
|
||||
let region = egui::Rect::from_min_max(egui::pos2(8.0, 4.0), egui::pos2(15.0, 9.0));
|
||||
assert!(!shrunk.contains_region(region));
|
||||
let error = screenshot_png(&image, Some(region), shrunk).expect_err("a cut region must not become a smaller image");
|
||||
assert!(error.contains("inside the window"), "{error}");
|
||||
}
|
||||
|
||||
/// Screenshot identity is the captured content, not the window's monitor-space origin or the
|
||||
/// focused widget: only the capture epoch and the viewport-local size/scale matter.
|
||||
#[test]
|
||||
fn capture_geometry_compares_size_scale_and_capture_epoch_only() {
|
||||
let admitted = Geometry { size: egui::vec2(800.0, 600.0), pixels_per_point: 2.0, epoch: 5 };
|
||||
assert_eq!(admitted, Geometry { size: egui::vec2(800.0, 600.0), pixels_per_point: 2.0, epoch: 5 });
|
||||
assert_ne!(admitted, Geometry { size: egui::vec2(801.0, 600.0), pixels_per_point: 2.0, epoch: 5 });
|
||||
assert_ne!(admitted, Geometry { size: egui::vec2(800.0, 600.0), pixels_per_point: 1.5, epoch: 5 });
|
||||
assert_ne!(admitted, Geometry { size: egui::vec2(800.0, 600.0), pixels_per_point: 2.0, epoch: 6 });
|
||||
assert!(admitted.contains_region(egui::Rect::from_min_max(egui::pos2(0.0, 0.0), egui::pos2(800.0, 600.0))));
|
||||
assert!(!admitted.contains_region(egui::Rect::from_min_max(egui::pos2(0.0, 0.0), egui::pos2(801.0, 600.0))));
|
||||
assert!(!admitted.contains_region(egui::Rect::from_min_max(egui::pos2(-1.0, 0.0), egui::pos2(10.0, 10.0))));
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,615 @@
|
||||
use super::*;
|
||||
|
||||
// ---- transport (native only) -----------------------------------------------------------------
|
||||
|
||||
/// Where a running app's control channel can be reached (written to the `--control` file).
|
||||
#[cfg(not(target_arch = "wasm32"))]
|
||||
pub struct Endpoint {
|
||||
pub port: u16,
|
||||
pub token: String,
|
||||
}
|
||||
|
||||
/// What a connection may do before it authenticates, so an unauthenticated local client cannot
|
||||
/// hold a thread for ever or exhaust them (#1806).
|
||||
#[cfg(not(target_arch = "wasm32"))]
|
||||
#[derive(Clone, Copy)]
|
||||
struct Limits {
|
||||
/// How long a connection has to authenticate, counted from its acceptance: an absolute
|
||||
/// deadline, so a client feeding one byte at a time cannot extend it.
|
||||
auth_deadline: std::time::Duration,
|
||||
/// How many connections may wait for authentication at once; further ones are refused.
|
||||
max_unauthenticated: usize,
|
||||
/// How long a write may block on a client that never reads its replies.
|
||||
write_timeout: std::time::Duration,
|
||||
/// How long a connection refused after a reply is kept half-open, discarding input, so the
|
||||
/// reply is not lost to a reset (see [`close_after_reply`]).
|
||||
linger: std::time::Duration,
|
||||
/// How many refused connections may linger at once. A lingering connection has given back
|
||||
/// its place among [`Self::max_unauthenticated`], so this is what bounds their threads;
|
||||
/// beyond it a refused connection is closed at once.
|
||||
max_lingering: usize,
|
||||
}
|
||||
|
||||
#[cfg(not(target_arch = "wasm32"))]
|
||||
const LIMITS: Limits = Limits {
|
||||
auth_deadline: std::time::Duration::from_secs(10),
|
||||
max_unauthenticated: 8,
|
||||
write_timeout: std::time::Duration::from_secs(30),
|
||||
linger: std::time::Duration::from_secs(1),
|
||||
max_lingering: 8,
|
||||
};
|
||||
|
||||
/// One of a limited number of places, given back when dropped: a place among
|
||||
/// [`Limits::max_unauthenticated`] (given back when the connection authenticates or is refused,
|
||||
/// and on every way its thread can end), or among [`Limits::max_lingering`].
|
||||
#[cfg(not(target_arch = "wasm32"))]
|
||||
struct Slot(std::sync::Arc<std::sync::atomic::AtomicUsize>);
|
||||
|
||||
#[cfg(not(target_arch = "wasm32"))]
|
||||
impl Slot {
|
||||
fn take(waiting: &std::sync::Arc<std::sync::atomic::AtomicUsize>, max: usize) -> Option<Self> {
|
||||
use std::sync::atomic::Ordering;
|
||||
let mut n = waiting.load(Ordering::Acquire);
|
||||
loop {
|
||||
if n >= max {
|
||||
return None;
|
||||
}
|
||||
match waiting.compare_exchange_weak(n, n.saturating_add(1), Ordering::AcqRel, Ordering::Acquire) {
|
||||
Ok(_) => return Some(Slot(waiting.clone())),
|
||||
Err(now) => n = now,
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(not(target_arch = "wasm32"))]
|
||||
impl Drop for Slot {
|
||||
fn drop(&mut self) {
|
||||
self.0.fetch_sub(1, std::sync::atomic::Ordering::AcqRel);
|
||||
}
|
||||
}
|
||||
|
||||
/// The reading half of a connection, with the authentication deadline: every read before it
|
||||
/// gets the time left as its timeout, and none at all once it has passed.
|
||||
#[cfg(not(target_arch = "wasm32"))]
|
||||
struct Reading {
|
||||
stream: std::net::TcpStream,
|
||||
until: Option<std::time::Instant>,
|
||||
}
|
||||
|
||||
#[cfg(not(target_arch = "wasm32"))]
|
||||
impl std::io::Read for Reading {
|
||||
fn read(&mut self, buf: &mut [u8]) -> std::io::Result<usize> {
|
||||
if let Some(until) = self.until {
|
||||
let left = until.saturating_duration_since(std::time::Instant::now());
|
||||
if left.is_zero() {
|
||||
return Err(std::io::Error::new(std::io::ErrorKind::TimedOut, "the authentication deadline passed"));
|
||||
}
|
||||
self.stream.set_read_timeout(Some(left))?;
|
||||
}
|
||||
self.stream.read(buf)
|
||||
}
|
||||
}
|
||||
|
||||
/// Close a connection whose last reply has just been written without losing that reply.
|
||||
///
|
||||
/// Closing a socket with input still unread makes the kernel send a reset instead of a FIN, and
|
||||
/// the reset can discard the reply before the client reads it (Linux does): a client still
|
||||
/// sending when it is refused would never learn why. So send FIN first, then read and discard
|
||||
/// whatever arrives until the client closes its side or `linger` passes, and only then close.
|
||||
///
|
||||
/// Only [`Limits::max_lingering`] connections linger at once: `place` is one of those places,
|
||||
/// taken by [`refuse`] before the reply was written, and held for the drain only. Without one
|
||||
/// the connection is closed at once after the FIN, accepting the risk of a reset rather than
|
||||
/// holding another thread.
|
||||
#[cfg(not(target_arch = "wasm32"))]
|
||||
fn close_after_reply(reading: &mut Reading, place: Option<Slot>, linger: std::time::Duration) {
|
||||
use std::io::Read;
|
||||
let _ = reading.stream.shutdown(std::net::Shutdown::Write);
|
||||
let Some(_place) = place else { return };
|
||||
reading.until = Some(std::time::Instant::now() + linger);
|
||||
let mut sink = [0u8; 1024];
|
||||
// Each read is bounded by the time left, and none is made once `linger` has passed.
|
||||
while matches!(reading.read(&mut sink), Ok(n) if n > 0) {}
|
||||
}
|
||||
|
||||
/// A connection is being refused before `auth`: give back its place among those waiting to
|
||||
/// authenticate, so it does not keep a real client out while it lingers, and take a place to
|
||||
/// linger in if one is free. Done before the reply is written, so a client that has read the
|
||||
/// reply sees the places as they will be.
|
||||
#[cfg(not(target_arch = "wasm32"))]
|
||||
fn refuse(slot: &mut Option<Slot>, lingering: &std::sync::Arc<std::sync::atomic::AtomicUsize>, limits: Limits) -> Option<Slot> {
|
||||
drop(slot.take());
|
||||
Slot::take(lingering, limits.max_lingering)
|
||||
}
|
||||
|
||||
/// Listen on `127.0.0.1` (random port) and forward authenticated requests to `client`.
|
||||
///
|
||||
/// Protocol: newline-delimited JSON-RPC 2.0. The first request on a connection must be
|
||||
/// `{"method": "auth", "params": {"token": …}}`; anything else closes the connection, and so does
|
||||
/// a line longer than [`MAX_AUTH_LINE`] before that, or [`LIMITS`] passing without it.
|
||||
#[cfg(not(target_arch = "wasm32"))]
|
||||
pub fn serve(client: ControlClient) -> std::io::Result<Endpoint> {
|
||||
serve_with(client, LIMITS)
|
||||
}
|
||||
|
||||
/// How long the accepting thread waits after a failed accept. Accepting can keep failing, for
|
||||
/// example while the process has no file descriptors left, and retrying at once would keep a
|
||||
/// core busy until it recovers. The single-instance listener waits the same way.
|
||||
#[cfg(not(target_arch = "wasm32"))]
|
||||
const ACCEPT_RETRY: std::time::Duration = std::time::Duration::from_millis(100);
|
||||
|
||||
/// The connections `incoming` accepts, calling `wait` after each failed accept.
|
||||
#[cfg(not(target_arch = "wasm32"))]
|
||||
fn accepted<T>(incoming: impl Iterator<Item = std::io::Result<T>>, mut wait: impl FnMut()) -> impl Iterator<Item = T> {
|
||||
incoming.filter_map(move |stream| stream.map_err(|_| wait()).ok())
|
||||
}
|
||||
|
||||
#[cfg(not(target_arch = "wasm32"))]
|
||||
fn serve_with(client: ControlClient, limits: Limits) -> std::io::Result<Endpoint> {
|
||||
use std::io::{BufReader, Write};
|
||||
let listener = std::net::TcpListener::bind(("127.0.0.1", 0))?;
|
||||
let port = listener.local_addr()?.port();
|
||||
let token = random_token()?;
|
||||
let expected = token.clone();
|
||||
let waiting = std::sync::Arc::new(std::sync::atomic::AtomicUsize::new(0));
|
||||
let lingering = std::sync::Arc::new(std::sync::atomic::AtomicUsize::new(0));
|
||||
std::thread::Builder::new().name("pdfcraft-control".into()).spawn(move || {
|
||||
for stream in accepted(listener.incoming(), || std::thread::sleep(ACCEPT_RETRY)) {
|
||||
let _ = stream.set_write_timeout(Some(limits.write_timeout));
|
||||
// Admitted only while fewer than the limit wait to authenticate: refused on the
|
||||
// accepting thread, so a refused connection costs no thread at all.
|
||||
let Some(slot) = Slot::take(&waiting, limits.max_unauthenticated) else {
|
||||
let mut write = stream;
|
||||
let e = rpc_error(Value::Null, -32003, "too many connections waiting to authenticate; try again later");
|
||||
let _ = writeln!(write, "{e}").and_then(|_| write.flush());
|
||||
continue;
|
||||
};
|
||||
let client = client.clone();
|
||||
let expected = expected.clone();
|
||||
let lingering = lingering.clone();
|
||||
let _ = std::thread::Builder::new().name("pdfcraft-control-conn".into()).spawn(move || {
|
||||
// Held until `auth` succeeds; given back however this thread ends.
|
||||
let mut slot = Some(slot);
|
||||
let Ok(read) = stream.try_clone() else { return };
|
||||
let mut write = stream;
|
||||
let mut authed = false;
|
||||
let mut reader = BufReader::new(Reading { stream: read, until: Some(std::time::Instant::now() + limits.auth_deadline) });
|
||||
let mut buf = Vec::new();
|
||||
loop {
|
||||
// Until `auth` succeeds, lines are capped at MAX_AUTH_LINE; afterwards they
|
||||
// are read whole, as before.
|
||||
let max = if authed { usize::MAX } else { MAX_AUTH_LINE };
|
||||
match read_line_limited(&mut reader, &mut buf, max) {
|
||||
Ok(LineRead::Line) => {}
|
||||
// The deadline passed before `auth`: say so and close.
|
||||
Err(e) if !authed && matches!(e.kind(), std::io::ErrorKind::TimedOut | std::io::ErrorKind::WouldBlock) => {
|
||||
let secs = limits.auth_deadline.as_secs_f64();
|
||||
let e = rpc_error(Value::Null, -32004, &format!("not authenticated within {secs} s"));
|
||||
let place = refuse(&mut slot, &lingering, limits);
|
||||
if writeln!(write, "{e}").and_then(|_| write.flush()).is_ok() {
|
||||
close_after_reply(reader.get_mut(), place, limits.linger);
|
||||
}
|
||||
break;
|
||||
}
|
||||
// End of stream or a failed read ends the connection, as before.
|
||||
Ok(LineRead::End) | Err(_) => break,
|
||||
Ok(LineRead::TooLong) => {
|
||||
let e = rpc_error(Value::Null, -32600, &format!("request line longer than {max} bytes before auth"));
|
||||
let place = refuse(&mut slot, &lingering, limits);
|
||||
if writeln!(write, "{e}").and_then(|_| write.flush()).is_ok() {
|
||||
close_after_reply(reader.get_mut(), place, limits.linger);
|
||||
}
|
||||
break;
|
||||
}
|
||||
}
|
||||
// Bytes that aren't UTF-8 end the connection, as `BufRead::lines` did.
|
||||
let Ok(line) = std::str::from_utf8(&buf) else { break };
|
||||
if line.trim().is_empty() {
|
||||
continue;
|
||||
}
|
||||
let (reply, close) = match serde_json::from_str::<Value>(line) {
|
||||
// Before `auth`, like any other first request, this closes the connection.
|
||||
Err(e) => (rpc_error(Value::Null, -32700, &format!("parse error: {e}")), !authed),
|
||||
Ok(msg) => {
|
||||
let id = msg.get("id").cloned().unwrap_or(Value::Null);
|
||||
let method = msg.get("method").and_then(Value::as_str).unwrap_or("");
|
||||
let params = msg.get("params").cloned().unwrap_or(Value::Null);
|
||||
if !authed {
|
||||
let ok =
|
||||
method == "auth" && params.get("token").and_then(Value::as_str).is_some_and(|t| constant_time_eq(t, &expected));
|
||||
authed = ok;
|
||||
if ok {
|
||||
// Authenticated: no deadline from here on, and its place is free.
|
||||
reader.get_mut().until = None;
|
||||
let _ = reader.get_mut().stream.set_read_timeout(None);
|
||||
drop(slot.take());
|
||||
(json!({ "jsonrpc": "2.0", "id": id, "result": { "ok": true } }), false)
|
||||
} else {
|
||||
(rpc_error(id, -32001, "authenticate first: auth {token}"), true)
|
||||
}
|
||||
} else {
|
||||
let rx = client.send(method, params);
|
||||
match rx.recv_timeout(std::time::Duration::from_secs(30)) {
|
||||
Ok(Ok(v)) => (json!({ "jsonrpc": "2.0", "id": id, "result": v }), false),
|
||||
Ok(Err(e)) => (rpc_error(id, -32000, &e), false),
|
||||
Err(_) => (rpc_error(id, -32002, "the app did not answer within 30 s"), false),
|
||||
}
|
||||
}
|
||||
}
|
||||
};
|
||||
// `close` is only ever set before `auth`.
|
||||
let place = if close { refuse(&mut slot, &lingering, limits) } else { None };
|
||||
if writeln!(write, "{reply}").and_then(|_| write.flush()).is_err() {
|
||||
break;
|
||||
}
|
||||
if close {
|
||||
close_after_reply(reader.get_mut(), place, limits.linger);
|
||||
break;
|
||||
}
|
||||
}
|
||||
});
|
||||
}
|
||||
})?;
|
||||
Ok(Endpoint { port, token })
|
||||
}
|
||||
|
||||
/// Longest line read before a connection has authenticated. The only acceptable request then is
|
||||
/// `auth`, about a hundred bytes; a longer line is refused without buffering the rest of it.
|
||||
#[cfg(not(target_arch = "wasm32"))]
|
||||
const MAX_AUTH_LINE: usize = 4 * 1024;
|
||||
|
||||
/// What [`read_line_limited`] found.
|
||||
#[cfg(not(target_arch = "wasm32"))]
|
||||
#[derive(Debug, PartialEq, Eq)]
|
||||
enum LineRead {
|
||||
/// A line, now in the buffer (newline and a trailing `\r` stripped).
|
||||
Line,
|
||||
/// End of stream.
|
||||
End,
|
||||
/// The line is longer than the limit; at most limit + 1 bytes of it were read.
|
||||
TooLong,
|
||||
}
|
||||
|
||||
/// Read one `\n`-terminated line of at most `max` bytes into `buf`.
|
||||
#[cfg(not(target_arch = "wasm32"))]
|
||||
fn read_line_limited(reader: &mut impl std::io::BufRead, buf: &mut Vec<u8>, max: usize) -> std::io::Result<LineRead> {
|
||||
use std::io::{BufRead, Read};
|
||||
buf.clear();
|
||||
// `take` bounds what is buffered for one line: never more than `max` + 1 bytes.
|
||||
let limit = u64::try_from(max).unwrap_or(u64::MAX).saturating_add(1);
|
||||
if reader.take(limit).read_until(b'\n', buf)? == 0 {
|
||||
return Ok(LineRead::End);
|
||||
}
|
||||
if buf.last() == Some(&b'\n') {
|
||||
buf.pop();
|
||||
if buf.last() == Some(&b'\r') {
|
||||
buf.pop();
|
||||
}
|
||||
return Ok(LineRead::Line);
|
||||
}
|
||||
// No newline: either the limit was passed, or this is a last line before end of stream.
|
||||
Ok(if buf.len() > max { LineRead::TooLong } else { LineRead::Line })
|
||||
}
|
||||
|
||||
#[cfg(not(target_arch = "wasm32"))]
|
||||
fn rpc_error(id: Value, code: i64, message: &str) -> Value {
|
||||
json!({ "jsonrpc": "2.0", "id": id, "error": { "code": code, "message": message } })
|
||||
}
|
||||
|
||||
#[cfg(not(target_arch = "wasm32"))]
|
||||
fn constant_time_eq(a: &str, b: &str) -> bool {
|
||||
a.len() == b.len() && a.bytes().zip(b.bytes()).fold(0u8, |acc, (x, y)| acc | (x ^ y)) == 0
|
||||
}
|
||||
|
||||
/// 128 random bits from the OS, hex-encoded.
|
||||
#[cfg(not(target_arch = "wasm32"))]
|
||||
fn random_token() -> std::io::Result<String> {
|
||||
let mut bytes = [0u8; 16];
|
||||
getrandom::fill(&mut bytes).map_err(|e| std::io::Error::other(e.to_string()))?;
|
||||
Ok(bytes.iter().map(|b| format!("{b:02x}")).collect())
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod transport_tests {
|
||||
/// A failed accept is followed by a wait before the next one, so a listener that keeps
|
||||
/// failing does not spin, and the connections around the failures still come through.
|
||||
#[cfg(not(target_arch = "wasm32"))]
|
||||
#[test]
|
||||
fn a_failed_accept_waits_before_the_next() {
|
||||
let failed = || Err(std::io::Error::other("no file descriptors left"));
|
||||
let incoming = vec![failed(), Ok(1), failed(), failed(), Ok(2)];
|
||||
let mut waits = 0;
|
||||
let connections: Vec<i32> = super::accepted(incoming.into_iter(), || waits += 1).collect();
|
||||
assert_eq!(connections, [1, 2]);
|
||||
assert_eq!(waits, 3, "one wait per failed accept");
|
||||
}
|
||||
|
||||
/// Before authenticating, a line that isn't JSON gets a parse error and closes the
|
||||
/// connection, as [`super::serve`] documents for any first request other than `auth`.
|
||||
#[cfg(not(target_arch = "wasm32"))]
|
||||
#[test]
|
||||
fn malformed_input_before_auth_closes_the_connection() {
|
||||
use std::io::{BufRead, BufReader, ErrorKind, Write};
|
||||
let (_control, client) = super::attach(&egui::Context::default());
|
||||
let ep = super::serve(client).unwrap();
|
||||
let s = std::net::TcpStream::connect(("127.0.0.1", ep.port)).unwrap();
|
||||
s.set_read_timeout(Some(std::time::Duration::from_secs(5))).unwrap();
|
||||
writeln!(s.try_clone().unwrap(), "not json").unwrap();
|
||||
let mut r = BufReader::new(s);
|
||||
let mut first = String::new();
|
||||
r.read_line(&mut first).unwrap();
|
||||
let reply: serde_json::Value = serde_json::from_str(&first).unwrap();
|
||||
assert_eq!(reply["error"]["code"], -32700, "{first}");
|
||||
// Closed: end of stream (or a reset), not a connection left waiting for more input.
|
||||
let mut next = String::new();
|
||||
let closed = match r.read_line(&mut next) {
|
||||
Ok(n) => n == 0,
|
||||
Err(e) => matches!(e.kind(), ErrorKind::ConnectionReset | ErrorKind::ConnectionAborted),
|
||||
};
|
||||
assert!(closed, "the connection stayed open: {next:?}");
|
||||
}
|
||||
|
||||
#[cfg(not(target_arch = "wasm32"))]
|
||||
#[test]
|
||||
fn lines_are_read_up_to_the_limit() {
|
||||
use super::{LineRead, read_line_limited};
|
||||
let mut buf = Vec::new();
|
||||
let mut input: &[u8] = b"aaaaaaaaaa\nbb\r\ncc";
|
||||
assert_eq!(read_line_limited(&mut input, &mut buf, 10).unwrap(), LineRead::Line);
|
||||
assert_eq!(buf, b"aaaaaaaaaa", "a line exactly at the limit is accepted");
|
||||
assert_eq!(read_line_limited(&mut input, &mut buf, 10).unwrap(), LineRead::Line);
|
||||
assert_eq!(buf, b"bb", "CRLF is stripped");
|
||||
assert_eq!(read_line_limited(&mut input, &mut buf, 10).unwrap(), LineRead::Line);
|
||||
assert_eq!(buf, b"cc", "a last line without a newline is still read");
|
||||
assert_eq!(read_line_limited(&mut input, &mut buf, 10).unwrap(), LineRead::End, "then end of stream");
|
||||
let long = [b'x'; 100];
|
||||
let mut input: &[u8] = &long;
|
||||
assert_eq!(read_line_limited(&mut input, &mut buf, 10).unwrap(), LineRead::TooLong);
|
||||
assert_eq!(input.len(), 100 - 11, "at most limit + 1 bytes are read; the rest is left");
|
||||
}
|
||||
|
||||
/// Before authenticating, a line longer than [`super::MAX_AUTH_LINE`] is refused as soon as
|
||||
/// the limit is passed (no newline needed) and the connection is closed.
|
||||
#[cfg(not(target_arch = "wasm32"))]
|
||||
#[test]
|
||||
fn an_overlong_line_before_auth_is_refused() {
|
||||
use std::io::{BufRead, BufReader, ErrorKind, Write};
|
||||
let (_control, client) = super::attach(&egui::Context::default());
|
||||
let ep = super::serve(client).unwrap();
|
||||
let s = std::net::TcpStream::connect(("127.0.0.1", ep.port)).unwrap();
|
||||
s.set_read_timeout(Some(std::time::Duration::from_secs(5))).unwrap();
|
||||
// Exactly one byte over the limit, without a newline: the server reads it all, so the
|
||||
// close below is a clean end of stream.
|
||||
s.try_clone().unwrap().write_all(&vec![b'x'; super::MAX_AUTH_LINE + 1]).unwrap();
|
||||
let mut r = BufReader::new(s);
|
||||
let mut first = String::new();
|
||||
r.read_line(&mut first).unwrap();
|
||||
let reply: serde_json::Value = serde_json::from_str(&first).unwrap();
|
||||
assert_eq!(reply["error"]["code"], -32600, "{first}");
|
||||
let mut next = String::new();
|
||||
let closed = match r.read_line(&mut next) {
|
||||
Ok(n) => n == 0,
|
||||
Err(e) => matches!(e.kind(), ErrorKind::ConnectionReset | ErrorKind::ConnectionAborted),
|
||||
};
|
||||
assert!(closed, "the connection stayed open: {next:?}");
|
||||
}
|
||||
|
||||
/// Small limits for the tests below: a 300 ms deadline, two places to wait in.
|
||||
#[cfg(not(target_arch = "wasm32"))]
|
||||
const TEST_LIMITS: super::Limits = super::Limits {
|
||||
auth_deadline: std::time::Duration::from_millis(300),
|
||||
max_unauthenticated: 2,
|
||||
write_timeout: std::time::Duration::from_secs(5),
|
||||
linger: std::time::Duration::from_secs(1),
|
||||
max_lingering: 2,
|
||||
};
|
||||
|
||||
/// The next line the server sends, as JSON, or `None` when it closed the connection instead.
|
||||
#[cfg(not(target_arch = "wasm32"))]
|
||||
fn next_line(r: &mut std::io::BufReader<std::net::TcpStream>) -> Option<serde_json::Value> {
|
||||
use std::io::BufRead;
|
||||
let mut line = String::new();
|
||||
match r.read_line(&mut line) {
|
||||
Ok(0) | Err(_) => None,
|
||||
Ok(_) => Some(serde_json::from_str(&line).unwrap()),
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(not(target_arch = "wasm32"))]
|
||||
fn connect(port: u16) -> std::net::TcpStream {
|
||||
let s = std::net::TcpStream::connect(("127.0.0.1", port)).unwrap();
|
||||
s.set_read_timeout(Some(std::time::Duration::from_secs(5))).unwrap();
|
||||
s
|
||||
}
|
||||
|
||||
/// #1806: a connection that never authenticates is closed at the deadline, counted from its
|
||||
/// acceptance, and a client feeding a byte at a time cannot push that deadline back.
|
||||
#[cfg(not(target_arch = "wasm32"))]
|
||||
#[test]
|
||||
fn an_unauthenticated_connection_is_closed_at_the_deadline() {
|
||||
use std::io::{BufReader, Write};
|
||||
let (_control, client) = super::attach(&egui::Context::default());
|
||||
let ep = super::serve_with(client, TEST_LIMITS).unwrap();
|
||||
// Idle.
|
||||
let started = std::time::Instant::now();
|
||||
let mut r = BufReader::new(connect(ep.port));
|
||||
let reply = next_line(&mut r).expect("the server says why before closing");
|
||||
assert_eq!(reply["error"]["code"], -32004, "{reply}");
|
||||
let took = started.elapsed();
|
||||
assert!(took >= std::time::Duration::from_millis(250) && took < std::time::Duration::from_secs(3), "{took:?}");
|
||||
assert!(next_line(&mut r).is_none(), "then the connection is closed");
|
||||
// Dripping: a byte every 50 ms for as long as the server keeps the connection.
|
||||
let started = std::time::Instant::now();
|
||||
let s = connect(ep.port);
|
||||
let mut w = s.try_clone().unwrap();
|
||||
let drip = std::thread::spawn(move || {
|
||||
while w.write_all(b"x").is_ok() {
|
||||
std::thread::sleep(std::time::Duration::from_millis(50));
|
||||
if started.elapsed() > std::time::Duration::from_secs(3) {
|
||||
break;
|
||||
}
|
||||
}
|
||||
});
|
||||
let mut r = BufReader::new(s);
|
||||
let reply = next_line(&mut r).expect("the server says why before closing");
|
||||
assert_eq!(reply["error"]["code"], -32004, "{reply}");
|
||||
let took = started.elapsed();
|
||||
assert!(took < std::time::Duration::from_secs(2), "the deadline is absolute: {took:?}");
|
||||
drop(r);
|
||||
drip.join().unwrap();
|
||||
}
|
||||
|
||||
/// A client refused while the server still has its input unread gets the reason, not a
|
||||
/// reset: closing on top of unread input sends a reset, which can discard the reply before
|
||||
/// the client reads it (Linux does). Here a line far over the limit leaves most of itself
|
||||
/// unread when the server refuses it.
|
||||
#[cfg(not(target_arch = "wasm32"))]
|
||||
#[test]
|
||||
fn a_client_refused_with_input_unread_gets_the_reason() {
|
||||
use std::io::{BufReader, Write};
|
||||
let (_control, client) = super::attach(&egui::Context::default());
|
||||
let ep = super::serve_with(client, TEST_LIMITS).unwrap();
|
||||
let s = connect(ep.port);
|
||||
let mut w = s.try_clone().unwrap();
|
||||
w.write_all(&vec![b'x'; 16 * super::MAX_AUTH_LINE]).unwrap();
|
||||
// Let the server refuse it (and, before the fix, close) before reading the reply.
|
||||
std::thread::sleep(std::time::Duration::from_millis(200));
|
||||
let mut r = BufReader::new(s);
|
||||
let reply = next_line(&mut r).expect("the server says why before closing");
|
||||
assert_eq!(reply["error"]["code"], -32600, "{reply}");
|
||||
}
|
||||
|
||||
/// A refused connection gives back its place among those waiting to authenticate before it
|
||||
/// lingers, so lingering does not keep real clients out.
|
||||
#[cfg(not(target_arch = "wasm32"))]
|
||||
#[test]
|
||||
fn a_lingering_connection_holds_no_place_to_authenticate() {
|
||||
use std::io::{BufReader, Write};
|
||||
// A linger far longer than the checks below take, even on a loaded machine.
|
||||
let limits = super::Limits { linger: std::time::Duration::from_secs(5), ..TEST_LIMITS };
|
||||
let (_control, client) = super::attach(&egui::Context::default());
|
||||
let ep = super::serve_with(client, limits).unwrap();
|
||||
let auth = serde_json::json!({ "jsonrpc": "2.0", "id": 1, "method": "auth", "params": { "token": ep.token } });
|
||||
// Refused with most of its line unread, and left open: it lingers for `linger`.
|
||||
let refused = connect(ep.port);
|
||||
refused.try_clone().unwrap().write_all(&vec![b'x'; 16 * super::MAX_AUTH_LINE]).unwrap();
|
||||
let mut r = BufReader::new(refused.try_clone().unwrap());
|
||||
assert_eq!(next_line(&mut r).expect("refused with a reason")["error"]["code"], -32600);
|
||||
// Well within the linger, both places are free for new connections.
|
||||
let started = std::time::Instant::now();
|
||||
let others: Vec<_> = (0..limits.max_unauthenticated).map(|_| connect(ep.port)).collect();
|
||||
for s in &others {
|
||||
// A refused one has already been closed, so its write may fail; its reply says why.
|
||||
let _ = writeln!(s.try_clone().unwrap(), "{auth}");
|
||||
let reply = next_line(&mut BufReader::new(s.try_clone().unwrap())).expect("a reply");
|
||||
assert_eq!(reply["result"]["ok"], true, "admitted while the refused one lingers: {reply}");
|
||||
}
|
||||
assert!(started.elapsed() < limits.linger, "checked while it lingered: {:?}", started.elapsed());
|
||||
drop(refused);
|
||||
}
|
||||
|
||||
/// Only [`super::Limits::max_lingering`] refused connections linger at once: one more still
|
||||
/// gets its reply, and is then closed at once instead of lingering.
|
||||
#[cfg(not(target_arch = "wasm32"))]
|
||||
#[test]
|
||||
fn only_so_many_refused_connections_linger() {
|
||||
use std::io::{BufReader, Write};
|
||||
// A linger far longer than the checks below take, even on a loaded machine.
|
||||
let limits = super::Limits { linger: std::time::Duration::from_secs(5), max_lingering: 1, ..TEST_LIMITS };
|
||||
let (_control, client) = super::attach(&egui::Context::default());
|
||||
let ep = super::serve_with(client, limits).unwrap();
|
||||
let bad = r#"{"jsonrpc": "2.0", "id": 1, "method": "page_count"}"#;
|
||||
// The first refused connection takes the only lingering place and keeps it.
|
||||
let first = connect(ep.port);
|
||||
writeln!(first.try_clone().unwrap(), "{bad}").unwrap();
|
||||
let mut r1 = BufReader::new(first.try_clone().unwrap());
|
||||
assert_eq!(next_line(&mut r1).expect("refused with a reason")["error"]["code"], -32001);
|
||||
// The second still gets its reason (it has nothing unread, so no reset)...
|
||||
let second = connect(ep.port);
|
||||
let mut w2 = second.try_clone().unwrap();
|
||||
writeln!(w2, "{bad}").unwrap();
|
||||
let mut r2 = BufReader::new(second.try_clone().unwrap());
|
||||
assert_eq!(next_line(&mut r2).expect("refused with a reason")["error"]["code"], -32001);
|
||||
// ...and is closed outright, not lingering: writing to it soon fails, where a lingering
|
||||
// connection would read and discard for the whole of `linger`.
|
||||
let started = std::time::Instant::now();
|
||||
let closed = loop {
|
||||
if w2.write_all(b"x").is_err() {
|
||||
break true;
|
||||
}
|
||||
if started.elapsed() > limits.linger {
|
||||
break false;
|
||||
}
|
||||
std::thread::sleep(std::time::Duration::from_millis(20));
|
||||
};
|
||||
assert!(closed && started.elapsed() < limits.linger / 2, "it lingered: {:?}", started.elapsed());
|
||||
drop(first);
|
||||
}
|
||||
|
||||
/// #1806: only so many connections may wait to authenticate at once. One more is refused
|
||||
/// on the spot; once a waiting one authenticates or times out, its place is free again.
|
||||
#[cfg(not(target_arch = "wasm32"))]
|
||||
#[test]
|
||||
fn only_so_many_connections_may_wait_to_authenticate() {
|
||||
use std::io::{BufReader, Write};
|
||||
let (_control, client) = super::attach(&egui::Context::default());
|
||||
let ep = super::serve_with(client, TEST_LIMITS).unwrap();
|
||||
let auth = serde_json::json!({ "jsonrpc": "2.0", "id": 1, "method": "auth", "params": { "token": ep.token } });
|
||||
let first = connect(ep.port);
|
||||
let second = connect(ep.port);
|
||||
// The kernel accepts a connection before the listener admits it: give the listener a
|
||||
// moment to admit both, well within their deadline.
|
||||
std::thread::sleep(std::time::Duration::from_millis(100));
|
||||
// The third finds both places taken.
|
||||
let mut r3 = BufReader::new(connect(ep.port));
|
||||
let reply = next_line(&mut r3).expect("refused with a reason");
|
||||
assert_eq!(reply["error"]["code"], -32003, "{reply}");
|
||||
assert!(next_line(&mut r3).is_none(), "and closed");
|
||||
// The second authenticates: its place is free, so a new connection is admitted.
|
||||
let mut r2 = BufReader::new(second.try_clone().unwrap());
|
||||
let mut w2 = second.try_clone().unwrap();
|
||||
writeln!(w2, "{auth}").unwrap();
|
||||
assert_eq!(next_line(&mut r2).unwrap()["result"]["ok"], true);
|
||||
let fourth = connect(ep.port);
|
||||
let mut r4 = BufReader::new(fourth.try_clone().unwrap());
|
||||
let mut w4 = fourth.try_clone().unwrap();
|
||||
writeln!(w4, "{auth}").unwrap();
|
||||
assert_eq!(next_line(&mut r4).unwrap()["result"]["ok"], true, "admitted and authenticated");
|
||||
// The first was closed at the deadline (its place freed on that way out too): after it,
|
||||
// another connection is admitted even though the first never authenticated.
|
||||
drop(first);
|
||||
std::thread::sleep(TEST_LIMITS.auth_deadline + std::time::Duration::from_millis(100));
|
||||
let fifth = connect(ep.port);
|
||||
let mut r5 = BufReader::new(fifth.try_clone().unwrap());
|
||||
let mut w5 = fifth.try_clone().unwrap();
|
||||
writeln!(w5, "{auth}").unwrap();
|
||||
assert_eq!(next_line(&mut r5).unwrap()["result"]["ok"], true);
|
||||
}
|
||||
|
||||
/// After `auth` the limit no longer applies: a long line is read as before (here it isn't
|
||||
/// JSON, so it gets a parse error and the connection stays open).
|
||||
#[cfg(not(target_arch = "wasm32"))]
|
||||
#[test]
|
||||
fn long_lines_after_auth_are_still_read() {
|
||||
use std::io::{BufRead, BufReader, Write};
|
||||
let (_control, client) = super::attach(&egui::Context::default());
|
||||
let ep = super::serve(client).unwrap();
|
||||
let s = std::net::TcpStream::connect(("127.0.0.1", ep.port)).unwrap();
|
||||
s.set_read_timeout(Some(std::time::Duration::from_secs(5))).unwrap();
|
||||
let mut w = s.try_clone().unwrap();
|
||||
let mut r = BufReader::new(s);
|
||||
let reply = |r: &mut BufReader<std::net::TcpStream>| {
|
||||
let mut line = String::new();
|
||||
r.read_line(&mut line).unwrap();
|
||||
serde_json::from_str::<serde_json::Value>(&line).unwrap()
|
||||
};
|
||||
let auth = serde_json::json!({ "jsonrpc": "2.0", "id": 1, "method": "auth", "params": { "token": ep.token } });
|
||||
writeln!(w, "{auth}").unwrap();
|
||||
assert_eq!(reply(&mut r)["result"]["ok"], true);
|
||||
writeln!(w, "{}", "x".repeat(4 * super::MAX_AUTH_LINE)).unwrap();
|
||||
assert_eq!(reply(&mut r)["error"]["code"], -32700, "a long line after auth is parsed, not refused");
|
||||
writeln!(w, "y").unwrap();
|
||||
assert_eq!(reply(&mut r)["error"]["code"], -32700, "and the connection is still open");
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user