1#![cfg(not(target_arch = "wasm32"))]
2
3use std::cell::RefCell;
4use std::path::PathBuf;
5use std::rc::Rc;
6use std::sync::{mpsc, Arc};
7
8use crate::core::{map_entries, ExceptionInfo, Promise, TraceFrame, Value};
9use crate::invoke_hta::InvokeHtaError;
10use crate::lang::data::Symbol;
11use crate::lang::protocol::INamespaced;
12use crate::{
13 EvaluationId, InProcessSandboxProvider, Runtime, SandboxId, SandboxSpec, SandboxStatus,
14 SessionId, SessionKernel,
15};
16
17mod arguments;
18mod documentation;
19mod kernel;
20use arguments::{
21 keyword, optional_string as optional_string_argument, string as string_argument,
22 strings as strings_argument, strings_value, tap_value,
23};
24pub use documentation::{Documentation, DocumentationValue};
25use kernel::kernel_call;
26
27const RUNTIME_BROKER_STACK_SIZE: usize = if cfg!(debug_assertions) {
31 64 * 1024 * 1024
32} else {
33 8 * 1024 * 1024
34};
35const MAX_DIAGNOSTIC_DATA_BYTES: usize = 16 * 1024;
36
37#[derive(Clone, Copy)]
38enum RuntimeBootstrap {
39 Full,
40 Core,
41 Source,
42}
43
44enum Request {
45 Eval {
46 session: String,
47 source: String,
48 reply: mpsc::Sender<Result<String, String>>,
49 },
50 EvalDiagnostic {
51 session: String,
52 source: String,
53 reply: mpsc::Sender<Result<String, RuntimeDiagnostic>>,
54 },
55 Namespace {
56 session: String,
57 reply: mpsc::Sender<Result<String, String>>,
58 },
59 Complete {
60 session: String,
61 prefix: String,
62 reply: mpsc::Sender<Result<Vec<String>, String>>,
63 },
64 Doc {
65 session: String,
66 symbol: String,
67 reply: mpsc::Sender<Result<Documentation, String>>,
68 },
69 Create {
70 session: String,
71 reply: mpsc::Sender<Result<String, String>>,
72 },
73 Close {
74 session: String,
75 reply: mpsc::Sender<Result<String, String>>,
76 },
77 List {
78 reply: mpsc::Sender<Result<Vec<String>, String>>,
79 },
80 Info {
81 session: String,
82 reply: mpsc::Sender<Result<String, String>>,
83 },
84 RegisterResource {
85 name: String,
86 source: String,
87 reply: mpsc::Sender<Result<(), String>>,
88 },
89 RemoveResource {
90 name: String,
91 reply: mpsc::Sender<Result<(), String>>,
92 },
93 ListResources {
94 reply: mpsc::Sender<Result<Vec<String>, String>>,
95 },
96 InstallModule {
97 session: String,
98 manifest: String,
99 module: crate::wasmtime_provider::CompiledWasmModule,
100 reply: mpsc::Sender<Result<String, String>>,
101 },
102 InvokeModule {
103 session: String,
104 namespace: String,
105 export: String,
106 arguments: Vec<u8>,
107 reply: mpsc::Sender<Result<Vec<u8>, String>>,
108 },
109 InvokeHta {
110 session: String,
111 qualified_var: String,
112 arguments: Vec<u8>,
113 reply: mpsc::Sender<Result<Vec<u8>, InvokeHtaError>>,
114 },
115 SandboxOpen {
116 spec: SandboxSpec,
117 reply: mpsc::Sender<Result<SandboxId, String>>,
118 },
119 SandboxEval {
120 sandbox: SandboxId,
121 source: String,
122 started: mpsc::Sender<Result<EvaluationId, String>>,
123 reply: mpsc::Sender<Result<String, String>>,
124 },
125 SandboxCall {
126 sandbox: SandboxId,
127 callable: String,
128 arguments: Vec<u8>,
129 started: mpsc::Sender<Result<EvaluationId, String>>,
130 reply: mpsc::Sender<Result<Vec<u8>, String>>,
131 },
132 SandboxCancel {
133 sandbox: SandboxId,
134 evaluation: Option<EvaluationId>,
135 reply: mpsc::Sender<Result<bool, String>>,
136 },
137 SandboxStatus {
138 sandbox: SandboxId,
139 reply: mpsc::Sender<Result<SandboxStatus, String>>,
140 },
141 SandboxClose {
142 sandbox: SandboxId,
143 reply: mpsc::Sender<Result<(), String>>,
144 },
145 Shutdown,
146}
147
148struct BrokerHandle {
149 sender: mpsc::Sender<Request>,
150}
151
152impl Drop for BrokerHandle {
153 fn drop(&mut self) {
154 let _ = self.sender.send(Request::Shutdown);
155 }
156}
157
158#[derive(Clone)]
159pub struct RuntimeBroker {
160 handle: Arc<BrokerHandle>,
161}
162
163#[derive(Clone, Debug)]
167pub(crate) struct RuntimeDiagnostic {
168 pub message: String,
169 pub exception: Option<RuntimeException>,
170 pub frames: Vec<TraceFrame>,
171}
172
173#[derive(Clone, Debug)]
177pub(crate) struct RuntimeException {
178 pub message: String,
179 pub class: Option<String>,
180 pub code: Option<String>,
181 pub data: String,
182 pub cause: Option<Box<RuntimeException>>,
183 pub throws: Vec<crate::core::ExceptionSite>,
184}
185
186impl RuntimeDiagnostic {
187 fn message(message: String) -> Self {
188 Self {
189 message,
190 exception: None,
191 frames: Vec::new(),
192 }
193 }
194}
195
196fn exception_attribute(exception: &ExceptionInfo, name: &str) -> Option<Value> {
197 let key = Value::Keyword(name.into());
198 map_entries(exception.data.as_ref())?
199 .into_iter()
200 .find_map(|(candidate, value)| (candidate == key).then_some(value))
201}
202
203fn bounded_diagnostic_data(value: String) -> String {
204 if value.len() <= MAX_DIAGNOSTIC_DATA_BYTES {
205 return value;
206 }
207 let mut end = MAX_DIAGNOSTIC_DATA_BYTES.saturating_sub(3);
208 while end > 0 && !value.is_char_boundary(end) {
209 end -= 1;
210 }
211 format!("{}...", &value[..end])
212}
213
214fn exception_snapshot(exception: &ExceptionInfo) -> RuntimeException {
215 let provenance = exception.provenance.borrow();
216 RuntimeException {
217 message: exception.message.clone(),
218 class: exception_attribute(exception, "ex/class").map(|value| value.display()),
219 code: exception_attribute(exception, "ex/code").map(|value| value.display()),
220 data: bounded_diagnostic_data(exception.data.display()),
221 cause: exception.cause.as_deref().and_then(|value| match value {
222 Value::ExceptionInfo(cause) => Some(Box::new(exception_snapshot(cause))),
223 _ => None,
224 }),
225 throws: provenance.throws.clone(),
226 }
227}
228
229fn captured_exception(value: Option<Value>) -> Option<RuntimeException> {
230 match value {
231 Some(Value::ExceptionInfo(exception)) => Some(exception_snapshot(&exception)),
232 _ => None,
233 }
234}
235
236impl RuntimeBroker {
237 pub fn start() -> Result<Self, String> {
238 Self::start_with_bootstrap(None, false, false, false, RuntimeBootstrap::Full)
239 }
240
241 pub fn start_core() -> Result<Self, String> {
246 Self::start_with_bootstrap(None, false, false, false, RuntimeBootstrap::Core)
247 }
248
249 pub fn start_with(
250 root: Option<PathBuf>,
251 native_sockets: bool,
252 allow_process: bool,
253 allow_postgres: bool,
254 ) -> Result<Self, String> {
255 Self::start_with_bootstrap(
256 root,
257 native_sockets,
258 allow_process,
259 allow_postgres,
260 RuntimeBootstrap::Full,
261 )
262 }
263
264 pub fn start_with_backend(
269 root: Option<PathBuf>,
270 native_sockets: bool,
271 allow_process: bool,
272 allow_postgres: bool,
273 execution_backend: &str,
274 ) -> Result<Self, String> {
275 Self::start_with_bootstrap_and_backend(
276 root,
277 native_sockets,
278 allow_process,
279 allow_postgres,
280 RuntimeBootstrap::Full,
281 execution_backend,
282 )
283 }
284
285 pub fn start_with_backend_and_source_catalog(
288 root: Option<PathBuf>,
289 native_sockets: bool,
290 allow_process: bool,
291 allow_postgres: bool,
292 execution_backend: &str,
293 source_catalog: crate::project::SourceCatalog,
294 ) -> Result<Self, String> {
295 Self::start_with_bootstrap_and_backend_and_catalog(
296 root,
297 native_sockets,
298 allow_process,
299 allow_postgres,
300 RuntimeBootstrap::Full,
301 execution_backend,
302 Some(source_catalog),
303 )
304 }
305
306 pub fn start_with_source_catalog(
310 root: Option<PathBuf>,
311 native_sockets: bool,
312 allow_process: bool,
313 allow_postgres: bool,
314 execution_backend: &str,
315 source_catalog: crate::project::SourceCatalog,
316 ) -> Result<Self, String> {
317 Self::start_with_bootstrap_and_backend_and_catalog(
318 root,
319 native_sockets,
320 allow_process,
321 allow_postgres,
322 RuntimeBootstrap::Source,
323 execution_backend,
324 Some(source_catalog),
325 )
326 }
327
328 fn start_with_bootstrap(
329 root: Option<PathBuf>,
330 native_sockets: bool,
331 allow_process: bool,
332 allow_postgres: bool,
333 bootstrap: RuntimeBootstrap,
334 ) -> Result<Self, String> {
335 Self::start_with_bootstrap_and_backend(
336 root,
337 native_sockets,
338 allow_process,
339 allow_postgres,
340 bootstrap,
341 "interpreter",
342 )
343 }
344
345 fn start_with_bootstrap_and_backend(
346 root: Option<PathBuf>,
347 native_sockets: bool,
348 allow_process: bool,
349 allow_postgres: bool,
350 bootstrap: RuntimeBootstrap,
351 execution_backend: &str,
352 ) -> Result<Self, String> {
353 Self::start_with_bootstrap_and_backend_and_catalog(
354 root,
355 native_sockets,
356 allow_process,
357 allow_postgres,
358 bootstrap,
359 execution_backend,
360 None,
361 )
362 }
363
364 fn start_with_bootstrap_and_backend_and_catalog(
365 root: Option<PathBuf>,
366 native_sockets: bool,
367 allow_process: bool,
368 allow_postgres: bool,
369 bootstrap: RuntimeBootstrap,
370 execution_backend: &str,
371 source_catalog: Option<crate::project::SourceCatalog>,
372 ) -> Result<Self, String> {
373 crate::validate_execution_backend(execution_backend)?;
374 if allow_postgres {
375 return Err(
376 "PostgreSQL support is not included in the core hara-native crate".to_owned(),
377 );
378 }
379 let execution_backend = execution_backend.to_owned();
380 let (sender, receiver) = mpsc::channel();
381 std::thread::Builder::new()
382 .name("hara-runtime-broker".into())
383 .stack_size(RUNTIME_BROKER_STACK_SIZE)
384 .spawn(move || {
385 run(
386 receiver,
387 root,
388 native_sockets,
389 allow_process,
390 allow_postgres,
391 bootstrap,
392 execution_backend,
393 source_catalog,
394 )
395 })
396 .map_err(|error| format!("runtime broker failed: {error}"))?;
397 Ok(Self {
398 handle: Arc::new(BrokerHandle { sender }),
399 })
400 }
401
402 pub fn eval(&self, session: &str, source: &str) -> Result<String, String> {
403 self.call(|reply| Request::Eval {
404 session: session.into(),
405 source: source.into(),
406 reply,
407 })
408 }
409
410 pub(crate) fn eval_diagnostic(
411 &self,
412 session: &str,
413 source: &str,
414 ) -> Result<String, RuntimeDiagnostic> {
415 let (reply, response) = mpsc::channel();
416 self.handle
417 .sender
418 .send(Request::EvalDiagnostic {
419 session: session.into(),
420 source: source.into(),
421 reply,
422 })
423 .map_err(|_| RuntimeDiagnostic::message("runtime broker is closed".into()))?;
424 response.recv().map_err(|_| {
425 RuntimeDiagnostic::message("runtime broker stopped without a response".into())
426 })?
427 }
428
429 pub fn namespace(&self, session: &str) -> Result<String, String> {
430 self.call(|reply| Request::Namespace {
431 session: session.into(),
432 reply,
433 })
434 }
435
436 pub fn complete(&self, session: &str, prefix: &str) -> Result<Vec<String>, String> {
437 self.call(|reply| Request::Complete {
438 session: session.into(),
439 prefix: prefix.into(),
440 reply,
441 })
442 }
443
444 pub fn documentation(&self, session: &str, symbol: &str) -> Result<Documentation, String> {
445 self.call(|reply| Request::Doc {
446 session: session.into(),
447 symbol: symbol.into(),
448 reply,
449 })
450 }
451
452 pub fn create(&self, session: &str) -> Result<String, String> {
453 self.call(|reply| Request::Create {
454 session: session.into(),
455 reply,
456 })
457 }
458
459 pub fn close(&self, session: &str) -> Result<String, String> {
460 self.call(|reply| Request::Close {
461 session: session.into(),
462 reply,
463 })
464 }
465
466 pub fn list(&self) -> Result<Vec<String>, String> {
467 self.call(|reply| Request::List { reply })
468 }
469
470 pub fn info(&self, session: &str) -> Result<String, String> {
471 self.call(|reply| Request::Info {
472 session: session.into(),
473 reply,
474 })
475 }
476
477 pub fn register_resource(&self, name: &str, source: &str) -> Result<(), String> {
478 self.call(|reply| Request::RegisterResource {
479 name: name.into(),
480 source: source.into(),
481 reply,
482 })
483 }
484
485 pub fn remove_resource(&self, name: &str) -> Result<(), String> {
486 self.call(|reply| Request::RemoveResource {
487 name: name.into(),
488 reply,
489 })
490 }
491
492 pub fn resources(&self) -> Result<Vec<String>, String> {
493 self.call(|reply| Request::ListResources { reply })
494 }
495
496 pub fn install_module(
497 &self,
498 session: &str,
499 manifest: &str,
500 module: &crate::wasmtime_provider::CompiledWasmModule,
501 ) -> Result<String, String> {
502 self.call(|reply| Request::InstallModule {
503 session: session.into(),
504 manifest: manifest.into(),
505 module: module.clone(),
506 reply,
507 })
508 }
509
510 pub fn invoke_hta(
511 &self,
512 session: &str,
513 qualified_var: &str,
514 arguments: &[u8],
515 ) -> Result<Vec<u8>, InvokeHtaError> {
516 let (reply, response) = mpsc::channel();
517 self.handle
518 .sender
519 .send(Request::InvokeHta {
520 session: session.into(),
521 qualified_var: qualified_var.into(),
522 arguments: arguments.into(),
523 reply,
524 })
525 .map_err(|_| InvokeHtaError::BrokerClosed)?;
526 response.recv().map_err(|_| InvokeHtaError::BrokerStopped)?
527 }
528
529 pub fn invoke_module(
530 &self,
531 session: &str,
532 namespace: &str,
533 export: &str,
534 arguments: &[u8],
535 ) -> Result<Vec<u8>, String> {
536 self.call(|reply| Request::InvokeModule {
537 session: session.into(),
538 namespace: namespace.into(),
539 export: export.into(),
540 arguments: arguments.into(),
541 reply,
542 })
543 }
544
545 fn sandbox_open(&self, spec: SandboxSpec) -> Result<SandboxId, String> {
546 self.call(|reply| Request::SandboxOpen { spec, reply })
547 }
548
549 fn sandbox_eval_receiver(
550 &self,
551 sandbox: SandboxId,
552 source: &str,
553 ) -> Result<(EvaluationId, mpsc::Receiver<Result<String, String>>), String> {
554 let (reply, response) = mpsc::channel();
555 let (started_reply, started_response) = mpsc::channel();
556 self.handle
557 .sender
558 .send(Request::SandboxEval {
559 sandbox,
560 source: source.into(),
561 started: started_reply,
562 reply,
563 })
564 .map_err(|_| "runtime broker is closed".to_owned())?;
565 let evaluation = started_response.recv().map_err(|_| {
566 "runtime broker stopped before starting sandbox evaluation".to_owned()
567 })??;
568 Ok((evaluation, response))
569 }
570
571 fn sandbox_call_receiver(
572 &self,
573 sandbox: SandboxId,
574 callable: &str,
575 arguments: &[u8],
576 ) -> Result<(EvaluationId, mpsc::Receiver<Result<Vec<u8>, String>>), String> {
577 let (reply, response) = mpsc::channel();
578 let (started_reply, started_response) = mpsc::channel();
579 self.handle
580 .sender
581 .send(Request::SandboxCall {
582 sandbox,
583 callable: callable.into(),
584 arguments: arguments.into(),
585 started: started_reply,
586 reply,
587 })
588 .map_err(|_| "runtime broker is closed".to_owned())?;
589 let evaluation = started_response
590 .recv()
591 .map_err(|_| "runtime broker stopped before starting sandbox call".to_owned())??;
592 Ok((evaluation, response))
593 }
594
595 fn sandbox_cancel(&self, sandbox: SandboxId) -> Result<bool, String> {
596 self.call(|reply| Request::SandboxCancel {
597 sandbox,
598 evaluation: None,
599 reply,
600 })
601 }
602
603 fn sandbox_cancel_evaluation(
604 &self,
605 sandbox: SandboxId,
606 evaluation: EvaluationId,
607 ) -> Result<bool, String> {
608 self.call(|reply| Request::SandboxCancel {
609 sandbox,
610 evaluation: Some(evaluation),
611 reply,
612 })
613 }
614
615 fn sandbox_status(&self, sandbox: SandboxId) -> Result<SandboxStatus, String> {
616 self.call(|reply| Request::SandboxStatus { sandbox, reply })
617 }
618
619 fn sandbox_close(&self, sandbox: SandboxId) -> Result<(), String> {
620 self.call(|reply| Request::SandboxClose { sandbox, reply })
621 }
622
623 fn call<T>(
624 &self,
625 request: impl FnOnce(mpsc::Sender<Result<T, String>>) -> Request,
626 ) -> Result<T, String> {
627 let (reply, response) = mpsc::channel();
628 self.handle
629 .sender
630 .send(request(reply))
631 .map_err(|_| "runtime broker is closed".to_owned())?;
632 response
633 .recv()
634 .map_err(|_| "runtime broker stopped without a response".to_owned())?
635 }
636}
637
638fn runtime(
639 root: Option<&PathBuf>,
640 native_sockets: bool,
641 allow_process: bool,
642 allow_postgres: bool,
643 bootstrap: RuntimeBootstrap,
644 execution_backend: &str,
645 source_catalog: Option<&crate::project::SourceCatalog>,
646) -> Runtime {
647 let mut runtime = match bootstrap {
648 RuntimeBootstrap::Full => Runtime::new(),
649 RuntimeBootstrap::Core | RuntimeBootstrap::Source => Runtime::core(),
650 };
651 if let Some(source_catalog) = source_catalog {
652 runtime.register_source_catalog(source_catalog);
653 }
654 if matches!(bootstrap, RuntimeBootstrap::Source) {
655 runtime
656 .bootstrap_source_foundation()
657 .expect("source Foundation bootstrap must be valid");
658 }
659 if let Some(root) = root {
660 runtime.install_native_file_provider(root.to_string_lossy().as_ref());
661 }
662 if native_sockets {
663 runtime.install_native_socket_provider();
664 }
665 if allow_process {
666 runtime.install_native_process_provider();
667 }
668 runtime
672 .configure_execution_backend(execution_backend)
673 .expect("validated execution backend must configure");
674 let _ = allow_postgres;
675 runtime
676}
677
678fn run(
679 receiver: mpsc::Receiver<Request>,
680 root: Option<PathBuf>,
681 native_sockets: bool,
682 allow_process: bool,
683 allow_postgres: bool,
684 bootstrap: RuntimeBootstrap,
685 execution_backend: String,
686 source_catalog: Option<crate::project::SourceCatalog>,
687) {
688 let runtime_root = root.clone();
689 let runtime_backend = execution_backend.clone();
690 let runtime_catalog = source_catalog;
691 let runtime_factory: Rc<dyn Fn() -> Runtime> = Rc::new(move || {
692 runtime(
693 runtime_root.as_ref(),
694 native_sockets,
695 allow_process,
696 allow_postgres,
697 bootstrap,
698 &runtime_backend,
699 runtime_catalog.as_ref(),
700 )
701 });
702 let root_runtime = runtime_factory();
703 let mut kernel = SessionKernel::with_runtime_factory(root_runtime, runtime_factory);
704 kernel.register_sandbox_provider(Rc::new(InProcessSandboxProvider));
705 while let Ok(request) = receiver.recv() {
706 match request {
707 Request::Eval {
708 session,
709 source,
710 reply,
711 } => {
712 let result = broker_session_id(&session).and_then(|id| {
713 let runtime = kernel.session_mut(&id)?.runtime_mut()?;
714 if execution_backend == "direct-native" {
715 runtime.eval_native(&source)
716 } else {
717 runtime.eval_native_traced(&source)
718 }
719 });
720 let _ = reply.send(result);
721 }
722 Request::EvalDiagnostic {
723 session,
724 source,
725 reply,
726 } => {
727 let result = broker_session_id(&session)
728 .map_err(RuntimeDiagnostic::message)
729 .and_then(|id| {
730 let runtime = kernel
731 .session_mut(&id)
732 .and_then(|session| session.runtime_mut())
733 .map_err(RuntimeDiagnostic::message)?;
734 let ((result, frames), exception) = runtime.eval_native_diagnostic(&source);
735 result.map_err(|message| RuntimeDiagnostic {
736 message,
737 exception: captured_exception(exception),
738 frames,
739 })
740 });
741 let _ = reply.send(result);
742 }
743 Request::Namespace { session, reply } => {
744 let result =
745 broker_session_id(&session).and_then(|id| kernel.session_namespace(&id));
746 let _ = reply.send(result);
747 }
748 Request::Complete {
749 session,
750 prefix,
751 reply,
752 } => {
753 let result = broker_session_id(&session).and_then(|id| {
754 kernel.session(&id)?.runtime().map(|runtime| {
755 let mut symbols = runtime
756 .visible_symbols()
757 .into_iter()
758 .filter(|symbol| symbol.starts_with(&prefix))
759 .collect::<Vec<_>>();
760 symbols.dedup();
761 symbols
762 })
763 });
764 let _ = reply.send(result);
765 }
766 Request::Doc {
767 session,
768 symbol,
769 reply,
770 } => {
771 let result = broker_session_id(&session)
772 .and_then(|id| documentation(kernel.session(&id)?.runtime()?, &symbol));
773 let _ = reply.send(result);
774 }
775 Request::Create { session, reply } => {
776 let result = SessionId::parse(&session)
777 .map_err(|_| format!("Session already exists or is invalid: {session}"))
778 .and_then(|id| {
779 kernel
780 .create_session(id)
781 .map_err(|_| format!("Session already exists or is invalid: {session}"))
782 })
783 .map(|_| session);
784 let _ = reply.send(result);
785 }
786 Request::Close { session, reply } => {
787 let result = broker_session_id(&session)
788 .and_then(|id| kernel.close_session(&id))
789 .map(|_| session)
790 .map_err(|error| match error.as_str() {
791 "ROOT_CANNOT_CLOSE" => "ROOT cannot be closed".into(),
792 _ if error.starts_with("NO_SESSION ") => {
793 format!("No session: {}", error.trim_start_matches("NO_SESSION "))
794 }
795 _ => error,
796 });
797 let _ = reply.send(result);
798 }
799 Request::List { reply } => {
800 let names = kernel
801 .session_names()
802 .into_iter()
803 .map(|id| id.to_string())
804 .collect();
805 let _ = reply.send(Ok(names));
806 }
807 Request::Info { session, reply } => {
808 let result = broker_session_id(&session)
809 .and_then(|id| kernel.session_namespace(&id))
810 .map(|namespace| format!("{session} {namespace}"));
811 let _ = reply.send(result);
812 }
813 Request::RegisterResource {
814 name,
815 source,
816 reply,
817 } => {
818 kernel.register_resource(&name, &source);
819 let _ = reply.send(Ok(()));
820 }
821 Request::RemoveResource { name, reply } => {
822 kernel.remove_resource(&name);
823 let _ = reply.send(Ok(()));
824 }
825 Request::ListResources { reply } => {
826 let _ = reply.send(Ok(kernel.resource_names()));
827 }
828 Request::InstallModule {
829 session,
830 manifest,
831 module,
832 reply,
833 } => {
834 let result = broker_session_id(&session).and_then(|id| {
835 let runtime = kernel.session_mut(&id)?.runtime_mut()?;
836 let provider = module.provider();
837 let parsed =
838 crate::extension::ExtensionManifest::parse(&manifest, "MODULE PUT")?;
839 let namespace = parsed.namespace.clone();
840 runtime.install_wasm_extension(&manifest, "MODULE PUT", provider)?;
841 Ok(namespace)
842 });
843 let _ = reply.send(result);
844 }
845 Request::InvokeModule {
846 session,
847 namespace,
848 export,
849 arguments,
850 reply,
851 } => {
852 let result = broker_session_id(&session).and_then(|id| {
853 let runtime = kernel.session_mut(&id)?.runtime_mut()?;
854 let arguments = crate::hta::decode(&arguments)?;
855 let arguments: Vec<crate::extension::Value> = match arguments {
856 crate::extension::Value::Vector(values) => values.iter().cloned().collect(),
857 crate::extension::Value::Tuple(values) => values.iter().cloned().collect(),
858 other => {
859 return Err(format!(
860 "hta/arguments: expected vector, got {}",
861 other.display()
862 ))
863 }
864 };
865 let result = runtime.invoke_wasm_extension(&namespace, &export, &arguments)?;
866 crate::hta::encode(&result)
867 });
868 let _ = reply.send(result);
869 }
870 Request::InvokeHta {
871 session,
872 qualified_var,
873 arguments,
874 reply,
875 } => {
876 let result = SessionId::parse(&session)
877 .map_err(|_| InvokeHtaError::SessionMissing(session.clone()))
878 .and_then(|id| {
879 kernel
880 .session_mut(&id)
881 .map_err(|_| InvokeHtaError::SessionMissing(session.clone()))?
882 .runtime_mut()
883 .map_err(InvokeHtaError::Execution)?
884 .invoke_hta(&qualified_var, &arguments)
885 });
886 let _ = reply.send(result);
887 }
888 Request::SandboxOpen { spec, reply } => {
889 let _ = reply.send(kernel.open_sandbox(spec).map_err(|error| error.to_string()));
890 }
891 Request::SandboxEval {
892 sandbox,
893 source,
894 started,
895 reply,
896 } => match kernel.sandbox_eval(sandbox, &source) {
897 Ok(pending) => {
898 let _ = started.send(Ok(pending.evaluation()));
899 std::thread::spawn(move || {
900 let _ = reply.send(pending.wait().map_err(|error| error.to_string()));
901 });
902 }
903 Err(error) => {
904 let error = error.to_string();
905 let _ = started.send(Err(error.clone()));
906 let _ = reply.send(Err(error));
907 }
908 },
909 Request::SandboxCall {
910 sandbox,
911 callable,
912 arguments,
913 started,
914 reply,
915 } => match kernel.sandbox_call(sandbox, &callable, &arguments) {
916 Ok(pending) => {
917 let _ = started.send(Ok(pending.evaluation()));
918 std::thread::spawn(move || {
919 let _ = reply.send(pending.wait().map_err(|error| error.to_string()));
920 });
921 }
922 Err(error) => {
923 let error = error.to_string();
924 let _ = started.send(Err(error.clone()));
925 let _ = reply.send(Err(error));
926 }
927 },
928 Request::SandboxCancel {
929 sandbox,
930 evaluation,
931 reply,
932 } => {
933 let result = match evaluation {
934 Some(evaluation) => kernel.cancel_sandbox_evaluation(sandbox, evaluation),
935 None => kernel.cancel_sandbox(sandbox),
936 };
937 let _ = reply.send(result.map_err(|error| error.to_string()));
938 }
939 Request::SandboxStatus { sandbox, reply } => {
940 let _ = reply.send(
941 kernel
942 .sandbox_status(sandbox)
943 .map_err(|error| error.to_string()),
944 );
945 }
946 Request::SandboxClose { sandbox, reply } => {
947 let _ = reply.send(
948 kernel
949 .close_sandbox(sandbox)
950 .map_err(|error| error.to_string()),
951 );
952 }
953 Request::Shutdown => break,
954 }
955 }
956}
957
958fn broker_session_id(session: &str) -> Result<SessionId, String> {
959 SessionId::parse(session).map_err(|_| format!("No session: {session}"))
960}
961
962fn documentation(runtime: &Runtime, symbol: &str) -> Result<Documentation, String> {
963 documentation::lookup(runtime, symbol)
964}
965
966pub fn install_native_kernel(runtime: &mut Runtime, broker: RuntimeBroker) {
970 runtime.install_native_kernel_provider(Rc::new(move |operation, arguments| {
971 kernel_call(&broker, &operation, &arguments)
972 }));
973}
974
975#[cfg(test)]
976mod tests;