use std::cell::RefCell;
use std::collections::VecDeque;
use std::rc::Rc;
use brep_render::brep_kernel::HistoryRequest;
use brep_render::runner::{
CamJob, CamProgress, CamReply, CamSimJob, CamSimProgress, CamSimReply, Command, HistoryRunner, MeasureQuery, MeasureReply, MeshImportPhaseReply, MeshImportReply, MeshImportRequest, RegionFitJob, RegionFitReply, ApplyRegionChoicesJob, ApplyRegionChoicesReply, Reply,
RunProgress, RunReply, SheetLinesReply, SheetLinesRequest, StepProbeReply, StepProbeRequest,
TopologyReply, TopologyRequest, PluginActionRequest, PluginActionReply, PluginInstallRequest, PluginInstallReply,
};
use wasm_bindgen::prelude::*;
use wasm_bindgen::JsCast;
use web_sys::MessageEvent;
pub struct WorkerRunner {
worker: web_sys::Worker,
inbox: Rc<RefCell<VecDeque<Reply>>>,
_onmessage: Closure<dyn FnMut(MessageEvent)>,
run_buf: VecDeque<RunReply>,
query_buf: VecDeque<MeasureReply>,
mesh_import_buf: VecDeque<MeshImportReply>,
mesh_phase_buf: VecDeque<MeshImportPhaseReply>,
region_fit_buf: VecDeque<RegionFitReply>,
apply_choices_buf: VecDeque<ApplyRegionChoicesReply>,
step_probe_buf: VecDeque<StepProbeReply>,
topology_buf: VecDeque<TopologyReply>,
sheet_lines_buf: VecDeque<SheetLinesReply>,
cam_buf: VecDeque<CamReply>,
cam_progress_buf: VecDeque<CamProgress>,
cam_sim_buf: VecDeque<CamSimReply>,
cam_sim_progress_buf: VecDeque<CamSimProgress>,
progress_buf: VecDeque<RunProgress>,
in_flight: bool,
pending_run: Option<(HistoryRequest, u64, u64)>,
sent_library_revision: Option<u64>,
library_requested: bool,
sent_plugin_revision: Option<u64>,
plugin_action_buf: VecDeque<PluginActionReply>,
plugin_install_buf: VecDeque<PluginInstallReply>,
pending_install: Option<PluginInstallRequest>,
}
thread_local! {
static REPLY_WAKE: RefCell<Option<Rc<dyn Fn()>>> = const { RefCell::new(None) };
}
pub fn set_reply_wake(wake: Option<Rc<dyn Fn()>>) {
REPLY_WAKE.with(|w| *w.borrow_mut() = wake);
}
impl WorkerRunner {
pub fn new() -> Self {
let worker = spawn_worker();
let inbox: Rc<RefCell<VecDeque<Reply>>> = Rc::new(RefCell::new(VecDeque::new()));
let inbox_cb = inbox.clone();
let onmessage = Closure::wrap(Box::new(move |event: MessageEvent| {
if let Some(text) = event.data().as_string() {
match serde_json::from_str::<Reply>(&text) {
Ok(reply) => inbox_cb.borrow_mut().push_back(reply),
Err(error) => log::error!("worker reply parse failed: {error}"),
}
}
let wake = REPLY_WAKE.with(|w| w.borrow().clone());
if let Some(wake) = wake {
wake();
}
}) as Box<dyn FnMut(MessageEvent)>);
worker.set_onmessage(Some(onmessage.as_ref().unchecked_ref()));
Self {
worker,
inbox,
_onmessage: onmessage,
run_buf: VecDeque::new(),
query_buf: VecDeque::new(),
mesh_import_buf: VecDeque::new(),
mesh_phase_buf: VecDeque::new(),
region_fit_buf: VecDeque::new(),
apply_choices_buf: VecDeque::new(),
step_probe_buf: VecDeque::new(),
topology_buf: VecDeque::new(),
sheet_lines_buf: VecDeque::new(),
cam_buf: VecDeque::new(),
cam_progress_buf: VecDeque::new(),
cam_sim_buf: VecDeque::new(),
cam_sim_progress_buf: VecDeque::new(),
progress_buf: VecDeque::new(),
in_flight: false,
pending_run: None,
sent_library_revision: None,
library_requested: false,
sent_plugin_revision: None,
plugin_action_buf: VecDeque::new(),
plugin_install_buf: VecDeque::new(),
pending_install: None,
}
}
fn post(&self, command: &Command) -> Result<(), JsValue> {
let json = serde_json::to_string(command).expect("serialize worker command");
self.worker.post_message(&JsValue::from_str(&json))
}
fn post_run(&mut self, request: HistoryRequest, generation: u64, library_revision: u64) {
self.post(&Command::Run {
request,
generation,
parts_library_revision: library_revision,
})
.unwrap_or_else(|e| panic!("post run to history worker failed: {e:?}"));
self.in_flight = true;
}
fn drain(&mut self) {
loop {
let next = self.inbox.borrow_mut().pop_front();
let Some(reply) = next else { break };
match reply {
Reply::PluginAction(reply) => self.plugin_action_buf.push_back(reply),
Reply::ValidatedPlugins(reply) => {
if self.pending_install.as_ref().is_some_and(|p| p.id == reply.id) { self.pending_install = None; }
self.plugin_install_buf.push_back(reply);
}
Reply::Run(run) => {
self.run_buf.push_back(run);
self.in_flight = false;
if let Some((request, generation, revision)) = self.pending_run.take() {
self.post_run(request, generation, revision);
}
}
Reply::Query(query) => self.query_buf.push_back(query),
Reply::MeshImport(reply) => self.mesh_import_buf.push_back(reply),
Reply::MeshImportPhase(reply) => self.mesh_phase_buf.push_back(reply),
Reply::RegionFit(reply) => self.region_fit_buf.push_back(reply),
Reply::ApplyRegionChoices(reply) => self.apply_choices_buf.push_back(reply),
Reply::StepProbe(reply) => self.step_probe_buf.push_back(reply),
Reply::Topology(reply) => self.topology_buf.push_back(reply),
Reply::SheetLines(reply) => self.sheet_lines_buf.push_back(reply),
Reply::CamGenerate(reply) => self.cam_buf.push_back(reply),
Reply::CamProgress(progress) => self.cam_progress_buf.push_back(progress),
Reply::CamSimulate(reply) => self.cam_sim_buf.push_back(reply),
Reply::CamSimProgress(progress) => self.cam_sim_progress_buf.push_back(progress),
Reply::Progress(progress) => self.progress_buf.push_back(progress),
Reply::SessionExport { .. } => {}
Reply::NeedPartsLibrary => {
self.in_flight = false;
self.sent_library_revision = None;
self.library_requested = true;
self.pending_run = None;
}
}
}
}
}
impl Default for WorkerRunner {
fn default() -> Self {
Self::new()
}
}
fn spawn_worker() -> web_sys::Worker {
let options = web_sys::WorkerOptions::new();
options.set_type(web_sys::WorkerType::Module);
web_sys::Worker::new_with_options("./worker.js", &options)
.unwrap_or_else(|e| panic!("failed to spawn history worker (./worker.js): {e:?}"))
}
impl Drop for WorkerRunner {
fn drop(&mut self) {
self.worker.terminate();
}
}
impl HistoryRunner for WorkerRunner {
fn submit_plugin_install(&mut self, request: PluginInstallRequest) -> bool {
if self.post(&Command::ValidatePlugins(request.clone())).is_err() { return false; }
self.pending_install = Some(request);
true
}
fn poll_plugin_install(&mut self) -> Option<PluginInstallReply> {
self.drain();
self.plugin_install_buf.pop_front()
}
fn sync_plugins(&mut self, revision: u64, fetch: &mut dyn FnMut() -> serde_json::Value) {
if self.sent_plugin_revision != Some(revision) {
self.post(&Command::SetPlugins { revision, packages: fetch() })
.unwrap_or_else(|e| panic!("Failed to provision plugin worker: {e:?}"));
self.sent_plugin_revision = Some(revision);
}
}
fn submit_plugin_action(&mut self, request: PluginActionRequest) -> bool {
self.post(&Command::PluginAction(request)).is_ok()
}
fn poll_plugin_action(&mut self) -> Option<PluginActionReply> {
self.drain();
self.plugin_action_buf.pop_front()
}
fn submit_run(&mut self, request: HistoryRequest, generation: u64) {
let revision = self.sent_library_revision.unwrap_or(0);
if self.in_flight {
self.pending_run = Some((request, generation, revision));
} else {
self.post_run(request, generation, revision);
}
}
fn sync_parts_library(
&mut self,
revision: u64,
fetch: &mut dyn FnMut() -> brep_render::brep_kernel::PartsLibraryMap,
) {
if self.sent_library_revision == Some(revision) {
return;
}
if self
.post(&Command::SetPartsLibrary {
library: fetch(),
revision,
})
.is_ok()
{
self.sent_library_revision = Some(revision);
}
}
fn poll_library_request(&mut self) -> bool {
self.drain();
std::mem::take(&mut self.library_requested)
}
fn poll_run(&mut self) -> Option<RunReply> {
self.drain();
self.run_buf.pop_front()
}
fn submit_query(&mut self, query: MeasureQuery) {
let _ = self.post(&Command::Query(query));
}
fn poll_query(&mut self) -> Option<MeasureReply> {
self.drain();
self.query_buf.pop_front()
}
fn submit_mesh_import(&mut self, request: MeshImportRequest) {
let _ = self.post(&Command::MeshImport(request));
}
fn poll_mesh_import(&mut self) -> Option<MeshImportReply> {
self.drain();
self.mesh_import_buf.pop_front()
}
fn poll_mesh_import_phase(&mut self) -> Option<MeshImportPhaseReply> {
self.drain();
self.mesh_phase_buf.pop_front()
}
fn submit_region_fit(&mut self, job: RegionFitJob) -> bool {
self.post(&Command::RegionFit(job)).is_ok()
}
fn poll_region_fit(&mut self) -> Option<RegionFitReply> {
self.drain();
self.region_fit_buf.pop_front()
}
fn submit_apply_region_choices(&mut self, job: ApplyRegionChoicesJob) -> bool {
self.post(&Command::ApplyRegionChoices(job)).is_ok()
}
fn poll_apply_region_choices(&mut self) -> Option<ApplyRegionChoicesReply> {
self.drain();
self.apply_choices_buf.pop_front()
}
fn submit_step_probe(&mut self, request: StepProbeRequest) {
let _ = self.post(&Command::StepProbe(request));
}
fn poll_step_probe(&mut self) -> Option<StepProbeReply> {
self.drain();
self.step_probe_buf.pop_front()
}
fn submit_topology(&mut self, request: TopologyRequest) {
let _ = self.post(&Command::Topology(request));
}
fn poll_topology(&mut self) -> Option<TopologyReply> {
self.drain();
self.topology_buf.pop_front()
}
fn submit_sheet_lines(&mut self, request: SheetLinesRequest) -> bool {
self.post(&Command::SheetLines(request)).is_ok()
}
fn poll_sheet_lines(&mut self) -> Option<SheetLinesReply> {
self.drain();
self.sheet_lines_buf.pop_front()
}
fn submit_cam(&mut self, job: CamJob) -> bool {
self.post(&Command::CamGenerate(job)).is_ok()
}
fn poll_cam(&mut self) -> Option<CamReply> {
self.drain();
self.cam_buf.pop_front()
}
fn poll_cam_progress(&mut self) -> Option<CamProgress> {
self.drain();
let latest = self.cam_progress_buf.pop_back();
self.cam_progress_buf.clear();
latest
}
fn submit_cam_sim(&mut self, job: CamSimJob) -> bool {
self.post(&Command::CamSimulate(job)).is_ok()
}
fn poll_cam_sim(&mut self) -> Option<CamSimReply> {
self.drain();
self.cam_sim_buf.pop_front()
}
fn poll_cam_sim_progress(&mut self) -> Option<CamSimProgress> {
self.drain();
let latest = self.cam_sim_progress_buf.pop_back();
self.cam_sim_progress_buf.clear();
latest
}
fn poll_progress(&mut self) -> Option<RunProgress> {
self.drain();
let latest = self.progress_buf.pop_back();
self.progress_buf.clear();
latest
}
fn cancel(&mut self) -> bool {
self.worker.terminate();
self.worker = spawn_worker();
self.worker
.set_onmessage(Some(self._onmessage.as_ref().unchecked_ref()));
self.inbox.borrow_mut().clear();
self.run_buf.clear();
self.query_buf.clear();
self.mesh_import_buf.clear();
self.mesh_phase_buf.clear();
self.region_fit_buf.clear();
self.apply_choices_buf.clear();
self.step_probe_buf.clear();
self.topology_buf.clear();
self.sheet_lines_buf.clear();
self.progress_buf.clear();
self.pending_run = None;
self.in_flight = false;
self.sent_library_revision = None;
self.library_requested = false;
self.sent_plugin_revision = None;
self.plugin_action_buf.clear();
if let Some(request) = self.pending_install.clone() {
if self.post(&Command::ValidatePlugins(request.clone())).is_err() {
self.pending_install = None;
self.plugin_install_buf.push_back(PluginInstallReply { id: request.id, result: Err("Worker restart could not resume plugin validation".into()) });
}
}
true
}
fn reset(&mut self) {
let _ = self.post(&Command::Reset);
self.run_buf.clear();
self.query_buf.clear();
self.mesh_import_buf.clear();
self.mesh_phase_buf.clear();
self.region_fit_buf.clear();
self.apply_choices_buf.clear();
self.step_probe_buf.clear();
self.topology_buf.clear();
self.sheet_lines_buf.clear();
self.progress_buf.clear();
self.pending_run = None;
self.in_flight = false;
self.sent_library_revision = None;
self.library_requested = false;
self.sent_plugin_revision = None;
self.plugin_action_buf.clear();
}
}
#[wasm_bindgen]
pub fn worker_entry() {
brep_render::brep_kernel::panic_hook::set_once();
let scope: web_sys::DedicatedWorkerGlobalScope = js_sys::global().unchecked_into();
let scope_reply = scope.clone();
let runner = Rc::new(RefCell::new(brep_render::pipeline::SceneRunner::new()));
let onmessage = Closure::wrap(Box::new(move |event: MessageEvent| {
let Some(text) = event.data().as_string() else {
return;
};
let command = match serde_json::from_str::<Command>(&text) {
Ok(command) => command,
Err(error) => {
log::error!("worker command parse failed: {error}");
return;
}
};
let reply = {
let mut resident = runner.borrow_mut();
let mut progress = |reply: Reply| {
let json = serde_json::to_string(&reply).expect("serialize worker progress");
let _ = scope_reply.post_message(&JsValue::from_str(&json));
true
};
brep_render::runner::process_command(&mut resident, command, &mut progress)
};
if let Some(reply) = reply {
let json = serde_json::to_string(&reply).expect("serialize worker reply");
let _ = scope_reply.post_message(&JsValue::from_str(&json));
}
}) as Box<dyn FnMut(MessageEvent)>);
scope.set_onmessage(Some(onmessage.as_ref().unchecked_ref()));
onmessage.forget();
}