feat(plugins): introduce 'pipes', allowing users to pipe data to and control plugins from the command line (#3066)

* prototype - working with message from the cli

* prototype - pipe from the CLI to plugins

* prototype - pipe from the CLI to plugins and back again

* prototype - working with better cli interface

* prototype - working after removing unused stuff

* prototype - working with launching plugin if it is not launched, also fixed event ordering

* refactor: change message to cli-message

* prototype - allow plugins to send messages to each other

* fix: allow cli messages to send plugin parameters (and implement backpressure)

* fix: use input_pipe_id to identify cli pipes instead of their message name

* fix: come cleanups and add skip_cache parameter

* fix: pipe/client-server communication robustness

* fix: leaking messages between plugins while loading

* feat: allow plugins to specify how a new plugin instance is launched when sending messages

* fix: add permissions

* refactor: adjust cli api

* fix: improve cli plugin loading error messages

* docs: cli pipe

* fix: take plugin configuration into account when messaging between plugins

* refactor: pipe message protobuf interface

* refactor: update(event) -> pipe

* refactor - rename CliMessage to CliPipe

* fix: add is_private to pipes and change some naming

* refactor - cli client

* refactor: various cleanups

* style(fmt): rustfmt

* fix(pipes): backpressure across multiple plugins

* style: some cleanups

* style(fmt): rustfmt

* style: fix merge conflict mistake

* style(wording): clarify pipe permission

1 files changed, 54 insertions, 3 deletions

diff --git a/zellij-server/src/plugins/plugin_map.rs b/zellij-server/src/plugins/plugin_map.rs
index 067c178f..e03cf172 100644
--- a/zellij-server/src/plugins/plugin_map.rs
+++ b/zellij-server/src/plugins/plugin_map.rs
@@ -15,7 +15,7 @@ use zellij_utils::{
     data::EventType,
     data::PluginCapabilities,
     input::command::TerminalAction,
-    input::layout::{Layout, RunPlugin, RunPluginLocation},
+    input::layout::{Layout, PluginUserConfiguration, RunPlugin, RunPluginLocation},
     input::plugins::PluginConfig,
     ipc::ClientAttributes,
 };
@@ -157,13 +157,19 @@ impl PluginMap {
     pub fn all_plugin_ids_for_plugin_location(
         &self,
         plugin_location: &RunPluginLocation,
+        plugin_configuration: &PluginUserConfiguration,
     ) -> Result<Vec<PluginId>> {
         let err_context = || format!("Failed to get plugin ids for location {plugin_location}");
         let plugin_ids: Vec<PluginId> = self
             .plugin_assets
             .iter()
             .filter(|(_, (running_plugin, _subscriptions, _workers))| {
-                &running_plugin.lock().unwrap().plugin_env.plugin.location == plugin_location
+                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;
+                running_plugin_location == plugin_location
+                    && running_plugin_configuration == plugin_configuration
             })
             .map(|((plugin_id, _client_id), _)| *plugin_id)
             .collect();
@@ -172,6 +178,49 @@ impl PluginMap {
         }
         Ok(plugin_ids)
     }
+    pub fn clone_plugin_assets(
+        &self,
+    ) -> HashMap<RunPluginLocation, HashMap<PluginUserConfiguration, Vec<(PluginId, ClientId)>>>
+    {
+        let mut cloned_plugin_assets: HashMap<
+            RunPluginLocation,
+            HashMap<PluginUserConfiguration, Vec<(PluginId, ClientId)>>,
+        > = 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;
+            match cloned_plugin_assets.get_mut(running_plugin_location) {
+                Some(location_map) => match location_map.get_mut(running_plugin_configuration) {
+                    Some(plugin_instances_info) => {
+                        plugin_instances_info.push((*plugin_id, *client_id));
+                    },
+                    None => {
+                        location_map.insert(
+                            running_plugin_configuration.clone(),
+                            vec![(*plugin_id, *client_id)],
+                        );
+                    },
+                },
+                None => {
+                    let mut location_map = HashMap::new();
+                    location_map.insert(
+                        running_plugin_configuration.clone(),
+                        vec![(*plugin_id, *client_id)],
+                    );
+                    cloned_plugin_assets.insert(running_plugin_location.clone(), location_map);
+                },
+            }
+        }
+        cloned_plugin_assets
+    }
+    pub fn all_plugin_ids(&self) -> Vec<(PluginId, ClientId)> {
+        self.plugin_assets
+            .iter()
+            .map(|((plugin_id, client_id), _)| (*plugin_id, *client_id))
+            .collect()
+    }
     pub fn insert(
         &mut self,
         plugin_id: PluginId,
@@ -218,7 +267,7 @@ pub struct PluginEnv {
     pub permissions: Arc<Mutex<Option<HashSet<PermissionType>>>>,
     pub senders: ThreadSenders,
     pub wasi_env: WasiEnv,
-    pub tab_index: usize,
+    pub tab_index: Option<usize>,
     pub client_id: ClientId,
     #[allow(dead_code)]
     pub plugin_own_data_dir: PathBuf,
@@ -228,6 +277,8 @@ pub struct PluginEnv {
     pub default_shell: Option<TerminalAction>,
     pub default_layout: Box<Layout>,
     pub plugin_cwd: PathBuf,
+    pub input_pipes_to_unblock: Arc<Mutex<HashSet<String>>>,
+    pub input_pipes_to_block: Arc<Mutex<HashSet<String>>>,
 }
 
 impl PluginEnv {