//! The `[cli_client]` is used to attach to a running server session //! and dispatch actions, that are specified through the command line. use std::collections::{BTreeMap, HashSet}; use std::io::{self, BufRead, Write}; use std::process; use std::str::FromStr; use std::{fs, path::PathBuf}; use crate::os_input_output::ClientOsApi; use uuid::Uuid; use zellij_utils::{ cli::{SubscribeCli, SubscribeFormat}, data::PaneId, errors::prelude::*, input::actions::Action, ipc::{ClientToServerMsg, ExitReason, ServerToClientMsg}, }; pub fn start_cli_client( mut os_input: Box, session_name: &str, actions: Vec, ) -> i32 { let zellij_ipc_pipe: PathBuf = { let mut sock_dir = zellij_utils::consts::ZELLIJ_SOCK_DIR.clone(); fs::create_dir_all(&sock_dir).unwrap(); zellij_utils::shared::set_permissions(&sock_dir, 0o700).unwrap(); sock_dir.push(session_name); sock_dir }; crate::check_ipc_pipe_length(&zellij_ipc_pipe); os_input.connect_to_server(&*zellij_ipc_pipe); let pane_id = os_input .env_variable("ZELLIJ_PANE_ID") .and_then(|e| e.trim().parse().ok()); for action in actions { match action { Action::CliPipe { pipe_id, name, payload, plugin, args, configuration, launch_new, skip_cache, floating, in_place, cwd, pane_title, } => { pipe_client( &mut os_input, pipe_id, name, payload, plugin, args, configuration, launch_new, skip_cache, floating, in_place, pane_id, cwd, pane_title, ); }, action => { if let Some(exit_status) = individual_messages_client(&mut os_input, action, pane_id) { return exit_status; } }, } } os_input.send_to_server(ClientToServerMsg::ClientExited); 0 } fn pipe_client( os_input: &mut Box, pipe_id: String, mut name: Option, mut payload: Option, plugin: Option, args: Option>, mut configuration: Option>, launch_new: bool, skip_cache: bool, floating: Option, in_place: Option, pane_id: Option, cwd: Option, pane_title: Option, ) { let mut stdin = os_input.get_stdin_reader(); let name = name // first we try to take the explicitly supplied message name .take() // then we use the plugin, to facilitate using aliases .or_else(|| plugin.clone()) // then we use a uuid to at least have some sort of identifier for this message .or_else(|| Some(Uuid::new_v4().to_string())); if launch_new { // we do this to make sure the plugin is unique (has a unique configuration parameter) so // that a new one would be launched, but we'll still send it to the same instance rather // than launching a new one in every iteration of the loop configuration .get_or_insert_with(BTreeMap::new) .insert("_zellij_id".to_owned(), Uuid::new_v4().to_string()); } let create_msg = |payload: Option| -> ClientToServerMsg { ClientToServerMsg::Action { action: Action::CliPipe { pipe_id: pipe_id.clone(), name: name.clone(), payload, args: args.clone(), plugin: plugin.clone(), configuration: configuration.clone(), floating, in_place, launch_new, skip_cache, cwd: cwd.clone(), pane_title: pane_title.clone(), }, terminal_id: pane_id, client_id: None, is_cli_client: true, } }; let is_piped = !os_input.stdin_is_terminal(); loop { if let Some(payload) = payload.take() { let msg = create_msg(Some(payload)); os_input.send_to_server(msg); } else if !is_piped { // here we send an empty message to trigger the plugin, because we don't have any more // data let msg = create_msg(None); os_input.send_to_server(msg); } else { // we didn't get payload from the command line, meaning we listen on STDIN because this // signifies the user is about to pipe more (eg. cat my-large-file | zellij pipe ...) let mut buffer = String::new(); let _ = stdin.read_line(&mut buffer); if buffer.is_empty() { let msg = create_msg(None); os_input.send_to_server(msg); break; } else { // we've got data! send it down the pipe (most common) let msg = create_msg(Some(buffer)); os_input.send_to_server(msg); } } loop { // wait for a response and act accordingly match os_input.recv_from_server() { Some((ServerToClientMsg::UnblockCliPipeInput { pipe_name }, _)) => { // unblock this pipe, meaning we need to stop waiting for a response and read // once more from STDIN if pipe_name == pipe_id { if !is_piped { // if this client is not piped, we need to exit the process completely // rather than wait for more data process::exit(0); } else { break; } } }, Some((ServerToClientMsg::CliPipeOutput { pipe_name, output }, _)) => { // send data to STDOUT, this *does not* mean we need to unblock the input let err_context = "Failed to write to stdout"; if pipe_name == pipe_id { let mut stdout = os_input.get_stdout_writer(); stdout .write_all(output.as_bytes()) .context(err_context) .non_fatal(); stdout.flush().context(err_context).non_fatal(); } }, Some((ServerToClientMsg::Log { lines: log_lines }, _)) => { log_lines.iter().for_each(|line| println!("{line}")); process::exit(0); }, Some((ServerToClientMsg::LogError { lines: log_lines }, _)) => { log_lines.iter().for_each(|line| eprintln!("{line}")); process::exit(2); }, Some((ServerToClientMsg::Exit { exit_reason }, _)) => match exit_reason { ExitReason::Error(e) => { eprintln!("{}", e); process::exit(2); }, _ => { process::exit(0); }, }, _ => {}, } } } } fn individual_messages_client( os_input: &mut Box, action: Action, pane_id: Option, ) -> Option { let is_blocking = matches!( &action, Action::NewBlockingPane { unblock_condition: Some(_), .. } ); let msg = ClientToServerMsg::Action { action, terminal_id: pane_id, client_id: None, is_cli_client: true, }; os_input.send_to_server(msg); loop { match os_input.recv_from_server() { Some((ServerToClientMsg::UnblockInputThread, _)) if !is_blocking => { return None; }, Some((ServerToClientMsg::Log { lines: log_lines }, _)) => { log_lines.iter().for_each(|line| println!("{line}")); return None; }, Some((ServerToClientMsg::LogError { lines: log_lines }, _)) => { log_lines.iter().for_each(|line| eprintln!("{line}")); return Some(2); }, Some((ServerToClientMsg::Exit { exit_reason }, _)) => match exit_reason { ExitReason::Error(e) => { eprintln!("{}", e); return Some(2); }, ExitReason::CustomExitStatus(exit_status) => { return Some(exit_status); }, _ => { return None; }, }, _ => {}, } } } pub fn start_subscribe_client( os_input: Box, session_name: &str, subscribe_cli: SubscribeCli, ) { let zellij_ipc_pipe: PathBuf = { let mut sock_dir = zellij_utils::consts::ZELLIJ_SOCK_DIR.clone(); fs::create_dir_all(&sock_dir).unwrap(); zellij_utils::shared::set_permissions(&sock_dir, 0o700).unwrap(); sock_dir.push(session_name); sock_dir }; crate::check_ipc_pipe_length(&zellij_ipc_pipe); os_input.connect_to_server(&*zellij_ipc_pipe); // Parse pane IDs let pane_ids: Vec = subscribe_cli .pane_id .iter() .map(|s| { PaneId::from_str(s).unwrap_or_else(|e| { eprintln!("Invalid pane ID '{}': {}", s, e); process::exit(2); }) }) .collect(); // Send subscribe message os_input.send_to_server(ClientToServerMsg::SubscribeToPaneRenders { pane_ids: pane_ids.clone(), scrollback: subscribe_cli.scrollback, ansi: subscribe_cli.ansi, }); // Track remaining panes for exit-on-all-closed let mut remaining_panes: HashSet = pane_ids.into_iter().collect(); // Streaming receive loop let stdout = io::stdout(); let mut stdout = stdout.lock(); loop { match os_input.recv_from_server() { Some(( ServerToClientMsg::PaneRenderUpdate { pane_id, viewport, scrollback, is_initial, }, _, )) => match subscribe_cli.format { SubscribeFormat::Raw => { if let Some(ref scrollback_lines) = scrollback { for line in scrollback_lines { let _ = writeln!(stdout, "{}", line); } } for line in &viewport { let _ = writeln!(stdout, "{}", line); } let _ = stdout.flush(); }, SubscribeFormat::Json => { let json = serde_json::json!({ "event": "pane_update", "pane_id": pane_id.to_string(), "viewport": viewport, "scrollback": scrollback, "is_initial": is_initial, }); let _ = writeln!(stdout, "{}", json); let _ = stdout.flush(); }, }, Some((ServerToClientMsg::SubscribedPaneClosed { pane_id }, _)) => { remaining_panes.remove(&pane_id); match subscribe_cli.format { SubscribeFormat::Raw => {}, SubscribeFormat::Json => { let json = serde_json::json!({ "event": "pane_closed", "pane_id": pane_id.to_string(), }); let _ = writeln!(stdout, "{}", json); let _ = stdout.flush(); }, } if remaining_panes.is_empty() { break; } }, Some((ServerToClientMsg::Exit { .. }, _)) => break, Some((ServerToClientMsg::LogError { lines }, _)) => { for line in lines { eprintln!("{}", line); } process::exit(2); }, None => break, _ => {}, } } os_input.send_to_server(ClientToServerMsg::ClientExited); }