diff options
Diffstat (limited to 'zellij-server/src/plugins/plugin_map.rs')
| -rw-r--r-- | zellij-server/src/plugins/plugin_map.rs | 124 |
1 files changed, 96 insertions, 28 deletions
diff --git a/zellij-server/src/plugins/plugin_map.rs b/zellij-server/src/plugins/plugin_map.rs index 1d400288a..fb41c89e3 100644 --- a/zellij-server/src/plugins/plugin_map.rs +++ b/zellij-server/src/plugins/plugin_map.rs @@ -1,12 +1,18 @@ use crate::plugins::plugin_worker::MessageToWorker; use crate::plugins::PluginId; +use bytes::Bytes; +use std::io::Write; use std::{ - collections::{HashMap, HashSet}, + collections::{HashMap, HashSet, VecDeque}, path::PathBuf, sync::{Arc, Mutex}, }; -use wasmer::{Instance, Store}; -use wasmer_wasi::WasiEnv; +use wasmtime::{Instance, Store}; +use wasmtime_wasi::preview1::WasiP1Ctx; +use wasmtime_wasi::{ + HostInputStream, HostOutputStream, StdinStream, StdoutStream, StreamError, StreamResult, + Subscribe, +}; use crate::{thread_bus::ThreadSenders, ClientId}; @@ -165,9 +171,9 @@ impl PluginMap { .iter() .filter(|(_, (running_plugin, _subscriptions, _workers))| { let running_plugin = running_plugin.lock().unwrap(); - let running_plugin_location = &running_plugin.plugin_env.plugin.location; - let running_plugin_configuration = - &running_plugin.plugin_env.plugin.userspace_configuration; + let plugin_config = &running_plugin.store.data().plugin; + let running_plugin_location = &plugin_config.location; + let running_plugin_configuration = &plugin_config.userspace_configuration; running_plugin_location == plugin_location && running_plugin_configuration == plugin_configuration }) @@ -188,9 +194,9 @@ impl PluginMap { > = HashMap::new(); for ((plugin_id, client_id), (running_plugin, _, _)) in self.plugin_assets.iter() { let running_plugin = running_plugin.lock().unwrap(); - let running_plugin_location = &running_plugin.plugin_env.plugin.location; - let running_plugin_configuration = - &running_plugin.plugin_env.plugin.userspace_configuration; + let plugin_config = &running_plugin.store.data().plugin; + let running_plugin_location = &plugin_config.location; + let running_plugin_configuration = &plugin_config.userspace_configuration; match cloned_plugin_assets.get_mut(running_plugin_location) { Some(location_map) => match location_map.get_mut(running_plugin_configuration) { Some(plugin_instances_info) => { @@ -240,13 +246,10 @@ impl PluginMap { .find_map(|((p_id, _), (running_plugin, _, _))| { if *p_id == plugin_id { let running_plugin = running_plugin.lock().unwrap(); - let run_plugin_location = running_plugin.plugin_env.plugin.location.clone(); - let run_plugin_configuration = running_plugin - .plugin_env - .plugin - .userspace_configuration - .clone(); - let initial_cwd = running_plugin.plugin_env.plugin.initial_cwd.clone(); + let plugin_config = &running_plugin.store.data().plugin; + let run_plugin_location = plugin_config.location.clone(); + let run_plugin_configuration = plugin_config.userspace_configuration.clone(); + let initial_cwd = plugin_config.initial_cwd.clone(); Some(RunPlugin { _allow_exec_host_cmd: false, location: run_plugin_location, @@ -262,13 +265,12 @@ impl PluginMap { pub type Subscriptions = HashSet<EventType>; -#[derive(Clone)] pub struct PluginEnv { pub plugin_id: PluginId, pub plugin: PluginConfig, pub permissions: Arc<Mutex<Option<HashSet<PermissionType>>>>, pub senders: ThreadSenders, - pub wasi_env: WasiEnv, + pub wasi_ctx: WasiP1Ctx, pub tab_index: Option<usize>, pub client_id: ClientId, #[allow(dead_code)] @@ -282,6 +284,80 @@ pub struct PluginEnv { pub plugin_cwd: PathBuf, pub input_pipes_to_unblock: Arc<Mutex<HashSet<String>>>, pub input_pipes_to_block: Arc<Mutex<HashSet<String>>>, + pub subscriptions: Arc<Mutex<Subscriptions>>, + pub stdin_pipe: Arc<Mutex<VecDeque<u8>>>, + pub stdout_pipe: Arc<Mutex<VecDeque<u8>>>, +} + +#[derive(Clone)] +pub struct VecDequeInputStream(pub Arc<Mutex<VecDeque<u8>>>); + +impl StdinStream for VecDequeInputStream { + fn stream(&self) -> Box<dyn wasmtime_wasi::HostInputStream> { + Box::new(self.clone()) + } + + fn isatty(&self) -> bool { + false + } +} + +impl HostInputStream for VecDequeInputStream { + fn read(&mut self, size: usize) -> StreamResult<Bytes> { + let mut inner = self.0.lock().unwrap(); + let len = std::cmp::min(size, inner.len()); + Ok(Bytes::from_iter(inner.drain(0..len))) + } +} + +#[async_trait::async_trait] +impl Subscribe for VecDequeInputStream { + async fn ready(&mut self) {} +} + +pub struct WriteOutputStream<T>(pub Arc<Mutex<T>>); + +impl<T> Clone for WriteOutputStream<T> { + fn clone(&self) -> Self { + Self(self.0.clone()) + } +} + +impl<T: Write + Send + 'static> StdoutStream for WriteOutputStream<T> { + fn stream(&self) -> Box<dyn HostOutputStream> { + Box::new((*self).clone()) + } + + fn isatty(&self) -> bool { + false + } +} + +impl<T: Write + Send + 'static> HostOutputStream for WriteOutputStream<T> { + fn write(&mut self, bytes: Bytes) -> StreamResult<()> { + self.0 + .lock() + .unwrap() + .write_all(&*bytes) + .map_err(|e| StreamError::LastOperationFailed(e.into())) + } + + fn flush(&mut self) -> StreamResult<()> { + self.0 + .lock() + .unwrap() + .flush() + .map_err(|e| StreamError::LastOperationFailed(e.into())) + } + + fn check_write(&mut self) -> StreamResult<usize> { + Ok(usize::MAX) + } +} + +#[async_trait::async_trait] +impl<T: Send + 'static> Subscribe for WriteOutputStream<T> { + async fn ready(&mut self) {} } impl PluginEnv { @@ -305,9 +381,8 @@ pub enum AtomicEvent { } pub struct RunningPlugin { - pub store: Store, + pub store: Store<PluginEnv>, pub instance: Instance, - pub plugin_env: PluginEnv, pub rows: usize, pub columns: usize, next_event_ids: HashMap<AtomicEvent, usize>, @@ -315,17 +390,10 @@ pub struct RunningPlugin { } impl RunningPlugin { - pub fn new( - store: Store, - instance: Instance, - plugin_env: PluginEnv, - rows: usize, - columns: usize, - ) -> Self { + pub fn new(store: Store<PluginEnv>, instance: Instance, rows: usize, columns: usize) -> Self { RunningPlugin { store, instance, - plugin_env, rows, columns, next_event_ids: HashMap::new(), |