zellij_utils/
ipc.rs

1//! IPC stuff for starting to split things into a client and server model.
2use crate::{
3    data::{ClientId, ConnectToSession, HostTerminalThemeMode, KeyWithModifier, PaneId, Style},
4    errors::{prelude::*, ErrorContext},
5    input::{actions::Action, cli_assets::CliAssets},
6    pane_size::{Size, SizeInPixels},
7};
8use interprocess::local_socket::Stream as LocalSocketStream;
9use log::warn;
10use serde::{Deserialize, Serialize};
11use std::{
12    fmt::{Display, Error, Formatter},
13    io::{self, Read, Write},
14    marker::PhantomData,
15};
16
17// Protobuf imports
18use crate::client_server_contract::client_server_contract::{
19    ClientToServerMsg as ProtoClientToServerMsg, ServerToClientMsg as ProtoServerToClientMsg,
20};
21use prost::Message;
22
23mod enum_conversions;
24mod protobuf_conversion;
25
26#[cfg(test)]
27mod tests;
28
29type SessionId = u64;
30
31/// A bidirectional byte stream that supports cloning for simultaneous read/write.
32pub trait IpcStream: Read + Write + Send + 'static {
33    fn try_clone_stream(&self) -> io::Result<Box<dyn IpcStream>>;
34}
35
36impl IpcStream for LocalSocketStream {
37    fn try_clone_stream(&self) -> io::Result<Box<dyn IpcStream>> {
38        use interprocess::TryClone;
39        Ok(Box::new(self.try_clone()?))
40    }
41}
42
43#[derive(PartialEq, Eq, Serialize, Deserialize, Hash)]
44pub struct Session {
45    // Unique ID for this session
46    id: SessionId,
47    // Identifier for the underlying IPC primitive (socket, pipe)
48    conn_name: String,
49    // User configured alias for the session
50    alias: String,
51}
52
53// How do we want to connect to a session?
54#[derive(Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
55pub enum ClientType {
56    Reader,
57    Writer,
58}
59
60#[derive(Default, Serialize, Deserialize, Debug, Clone)]
61pub struct ClientAttributes {
62    pub size: Size,
63    pub style: Style,
64}
65
66#[derive(Default, Serialize, Deserialize, Debug, Clone, Copy, PartialEq, Eq)]
67pub struct PixelDimensions {
68    pub text_area_size: Option<SizeInPixels>,
69    pub character_cell_size: Option<SizeInPixels>,
70}
71
72#[derive(Serialize, Deserialize, Debug, Clone, Default, PartialEq, Eq)]
73pub struct PaneReference {
74    pub pane_id: u32,
75    pub is_plugin: bool,
76}
77
78#[derive(Serialize, Deserialize, Debug, Clone, Default, PartialEq, Eq)]
79pub struct ColorRegister {
80    pub index: usize,
81    pub color: String,
82}
83
84impl PixelDimensions {
85    pub fn merge(&mut self, other: PixelDimensions) {
86        if let Some(text_area_size) = other.text_area_size {
87            self.text_area_size = Some(text_area_size);
88        }
89        if let Some(character_cell_size) = other.character_cell_size {
90            self.character_cell_size = Some(character_cell_size);
91        }
92    }
93}
94
95#[derive(Serialize, Deserialize, Debug, Clone, PartialEq)]
96pub struct MobileSizePayload {
97    pub cols: usize,
98    pub rows: usize,
99}
100
101#[derive(Serialize, Deserialize, Debug, Clone, PartialEq)]
102pub struct MobileActivePanePayload {
103    pub pane_id: u32,
104    pub is_plugin: bool,
105    pub tab_position: usize,
106}
107
108#[derive(Serialize, Deserialize, Debug, Clone, PartialEq)]
109pub struct MobileTabPayload {
110    pub position: usize,
111    pub name: String,
112    pub active: bool,
113}
114
115#[derive(Serialize, Deserialize, Debug, Clone, PartialEq)]
116pub struct MobilePanePayload {
117    pub tab_position: usize,
118    pub pane_id: u32,
119    pub is_plugin: bool,
120    pub title: String,
121    pub is_floating: bool,
122    pub last_activity_secs_ago: u64,
123}
124
125#[derive(Serialize, Deserialize, Debug, Clone, PartialEq)]
126pub struct MobileSessionPayload {
127    pub name: String,
128    pub web_clients_allowed: bool,
129    pub tab_count: usize,
130    pub pane_count: usize,
131    pub connected_clients: usize,
132    pub creation_secs_ago: u64,
133}
134
135#[derive(Serialize, Deserialize, Debug, Clone, PartialEq)]
136pub struct MobileRenderPrefsPayload {
137    pub single_pane: bool,
138    pub fit: bool,
139    pub active_pane_is_fullscreen: bool,
140}
141
142#[derive(Serialize, Deserialize, Debug, Clone, PartialEq)]
143pub struct MobileStatePayload {
144    pub session_name: String,
145    pub now_secs: u64,
146    pub is_welcome_screen: bool,
147    pub desktop_client_connected: bool,
148    pub desktop_size: Option<MobileSizePayload>,
149    pub active_pane: Option<MobileActivePanePayload>,
150    pub tabs: Vec<MobileTabPayload>,
151    pub panes: Vec<MobilePanePayload>,
152    pub sessions: Vec<MobileSessionPayload>,
153    pub render_prefs: MobileRenderPrefsPayload,
154}
155
156// Types of messages sent from the client to the server
157#[allow(clippy::large_enum_variant)]
158#[derive(Serialize, Deserialize, Debug, Clone, PartialEq)]
159pub enum ClientToServerMsg {
160    DetachSession {
161        client_ids: Vec<ClientId>,
162    },
163    TerminalPixelDimensions {
164        pixel_dimensions: PixelDimensions,
165    },
166    BackgroundColor {
167        color: String,
168    },
169    ForegroundColor {
170        color: String,
171    },
172    ColorRegisters {
173        color_registers: Vec<ColorRegister>,
174    },
175    TerminalResize {
176        new_size: Size,
177    },
178    FirstClientConnected {
179        cli_assets: CliAssets,
180        is_web_client: bool,
181    },
182    AttachClient {
183        cli_assets: CliAssets,
184        tab_position_to_focus: Option<usize>,
185        pane_to_focus: Option<PaneReference>,
186        is_web_client: bool,
187    },
188    AttachWatcherClient {
189        terminal_size: Size,
190        is_web_client: bool,
191    },
192    Action {
193        action: Action,
194        terminal_id: Option<u32>,
195        client_id: Option<ClientId>,
196        is_cli_client: bool,
197    },
198    Key {
199        key: KeyWithModifier,
200        raw_bytes: Vec<u8>,
201        is_kitty_keyboard_protocol: bool,
202    },
203    ClientExited,
204    KillSession,
205    ConnStatus,
206    WebServerStarted {
207        base_url: String,
208    },
209    FailedToStartWebServer {
210        error: String,
211    },
212    SubscribeToPaneRenders {
213        pane_ids: Vec<PaneId>,
214        scrollback: Option<usize>,
215        ansi: bool,
216    },
217    DesktopNotificationResponse {
218        raw_bytes: Vec<u8>,
219    },
220    ForwardedReplyFromHost {
221        token: u32,
222        reply_bytes: Vec<u8>,
223    },
224    HostTerminalThemeChanged {
225        mode: HostTerminalThemeMode,
226    },
227    SoftKeyboardVisibilityChanged {
228        visible: bool,
229    },
230    NestedSessionFrameFromHost {
231        payload_bytes: Vec<u8>,
232    },
233    KittyGraphicsSupport {
234        supported: bool,
235    },
236    SixelSupport {
237        supported: bool,
238    },
239    RequestSessionList,
240    SetMobileRenderPreferences {
241        single_pane: bool,
242        fit: bool,
243    },
244    HostTerminalFocusChanged {
245        focused: bool,
246    },
247}
248
249// Types of messages sent from the server to the client
250#[derive(Serialize, Deserialize, Debug, Clone, PartialEq)]
251pub enum ServerToClientMsg {
252    Render {
253        content: String,
254    },
255    UnblockInputThread,
256    Exit {
257        exit_reason: ExitReason,
258    },
259    Connected,
260    Log {
261        lines: Vec<String>,
262    },
263    LogError {
264        lines: Vec<String>,
265    },
266    SwitchSession {
267        connect_to_session: ConnectToSession,
268    },
269    UnblockCliPipeInput {
270        pipe_name: String,
271    },
272    CliPipeOutput {
273        pipe_name: String,
274        output: String,
275    },
276    QueryTerminalSize,
277    SetSoftKeyboard {
278        on: bool,
279    },
280    StartWebServer,
281    RenamedSession {
282        name: String,
283    },
284    ConfigFileUpdated,
285    PaneRenderUpdate {
286        pane_id: PaneId,
287        viewport: Vec<String>,
288        scrollback: Option<Vec<String>>,
289        is_initial: bool,
290    },
291    SubscribedPaneClosed {
292        pane_id: PaneId,
293    },
294    ForwardQueryToHost {
295        token: u32,
296        query_bytes: Vec<u8>,
297        resolve_async: bool,
298    },
299    EmitNestedSessionFrame {
300        payload_bytes: Vec<u8>,
301    },
302    MobileState {
303        payload: MobileStatePayload,
304    },
305}
306
307#[derive(Serialize, Deserialize, Debug, Clone, PartialEq)]
308pub enum ExitReason {
309    Normal,
310    NormalDetached,
311    ForceDetached,
312    CannotAttach,
313    Disconnect,
314    WebClientsForbidden,
315    KickedByHost,
316    CustomExitStatus(i32),
317    Error(String),
318}
319
320impl Display for ExitReason {
321    fn fmt(&self, f: &mut Formatter) -> Result<(), Error> {
322        match self {
323            Self::Normal => write!(f, "Bye from Zellij!"),
324            Self::NormalDetached => write!(f, "Session detached"),
325            Self::ForceDetached => write!(
326                f,
327                "Session was detached from this client (possibly because another client connected)"
328            ),
329            Self::CannotAttach => write!(
330                f,
331                "Session attached to another client. Use --force flag to force connect."
332            ),
333            Self::WebClientsForbidden => write!(
334                f,
335                "Web clients are not allowed in this session - cannot attach"
336            ),
337            Self::Disconnect => {
338                let session_tip = match crate::envs::get_session_name() {
339                    Ok(name) => format!("`zellij attach {}`", name),
340                    Err(_) => "see `zellij ls` and `zellij attach`".to_string(),
341                };
342                write!(
343                    f,
344                    "
345Your zellij client lost connection to the zellij server.
346
347As a safety measure, you have been disconnected from the current zellij session.
348However, the session should still exist and none of your data should be lost.
349
350This usually means that your terminal didn't process server messages quick
351enough. Maybe your system is currently under high load, or your terminal
352isn't performant enough.
353
354There are a few things you can try now:
355    - Reattach to your previous session and see if it works out better this
356      time: {session_tip}
357    - Try using a faster (maybe GPU-accelerated) terminal emulator
358    "
359                )
360            },
361            Self::KickedByHost => write!(f, "Disconnected by host"),
362            Self::CustomExitStatus(exit_status) => write!(f, "Exit {}", exit_status),
363            Self::Error(e) => write!(f, "Error occurred in server:\n{}", e),
364        }
365    }
366}
367
368/// Sends messages on a stream socket, along with an [`ErrorContext`].
369pub struct IpcSenderWithContext<T: Serialize> {
370    sender: io::BufWriter<Box<dyn IpcStream>>,
371    _phantom: PhantomData<T>,
372}
373
374impl<T: Serialize> IpcSenderWithContext<T> {
375    /// Returns a sender to the given [LocalSocketStream](interprocess::local_socket::LocalSocketStream).
376    pub fn new(sender: LocalSocketStream) -> Self {
377        Self {
378            sender: io::BufWriter::new(Box::new(sender)),
379            _phantom: PhantomData,
380        }
381    }
382
383    fn from_boxed(sender: Box<dyn IpcStream>) -> Self {
384        Self {
385            sender: io::BufWriter::new(sender),
386            _phantom: PhantomData,
387        }
388    }
389
390    pub fn send_client_msg(&mut self, msg: ClientToServerMsg) -> Result<()> {
391        let proto_msg: ProtoClientToServerMsg = msg.into();
392        write_protobuf_message(&mut self.sender, &proto_msg)?;
393        let _ = self.sender.flush();
394        Ok(())
395    }
396
397    pub fn send_server_msg(&mut self, msg: ServerToClientMsg) -> Result<()> {
398        let proto_msg: ProtoServerToClientMsg = msg.into();
399        write_protobuf_message(&mut self.sender, &proto_msg)?;
400        let _ = self.sender.flush();
401        Ok(())
402    }
403
404    /// Returns an [`IpcReceiverWithContext`] with the same socket as this sender.
405    pub fn get_receiver<F>(&self) -> IpcReceiverWithContext<F>
406    where
407        F: for<'de> Deserialize<'de> + Serialize,
408    {
409        let socket = self.sender.get_ref().try_clone_stream().unwrap();
410        IpcReceiverWithContext::from_boxed(socket)
411    }
412}
413
414/// Receives messages on a stream socket, along with an [`ErrorContext`].
415pub struct IpcReceiverWithContext<T> {
416    receiver: io::BufReader<Box<dyn IpcStream>>,
417    _phantom: PhantomData<T>,
418}
419
420#[derive(Debug, Clone, Copy, PartialEq, Eq)]
421pub enum IpcReceiveError {
422    Disconnected,
423    Undecodable,
424}
425
426impl Display for IpcReceiveError {
427    fn fmt(&self, f: &mut Formatter<'_>) -> std::result::Result<(), Error> {
428        match self {
429            IpcReceiveError::Disconnected => write!(f, "the peer closed the connection"),
430            IpcReceiveError::Undecodable => write!(f, "received a message that could not be read"),
431        }
432    }
433}
434
435impl<T> IpcReceiverWithContext<T>
436where
437    T: for<'de> Deserialize<'de> + Serialize,
438{
439    /// Returns a receiver to the given [LocalSocketStream](interprocess::local_socket::LocalSocketStream).
440    pub fn new(receiver: LocalSocketStream) -> Self {
441        Self {
442            receiver: io::BufReader::new(Box::new(receiver)),
443            _phantom: PhantomData,
444        }
445    }
446
447    fn from_boxed(receiver: Box<dyn IpcStream>) -> Self {
448        Self {
449            receiver: io::BufReader::new(receiver),
450            _phantom: PhantomData,
451        }
452    }
453
454    pub fn recv_client_msg(&mut self) -> Option<(ClientToServerMsg, ErrorContext)> {
455        self.try_recv_client_msg().ok()
456    }
457
458    pub fn recv_server_msg(&mut self) -> Option<(ServerToClientMsg, ErrorContext)> {
459        self.try_recv_server_msg().ok()
460    }
461
462    pub fn try_recv_client_msg(
463        &mut self,
464    ) -> std::result::Result<(ClientToServerMsg, ErrorContext), IpcReceiveError> {
465        let proto_msg = read_protobuf_message::<ProtoClientToServerMsg>(&mut self.receiver)?;
466        match proto_msg.try_into() {
467            Ok(rust_msg) => Ok((rust_msg, ErrorContext::default())),
468            Err(e) => {
469                warn!("Error converting protobuf to ClientToServerMsg: {:?}", e);
470                Err(IpcReceiveError::Undecodable)
471            },
472        }
473    }
474
475    pub fn try_recv_server_msg(
476        &mut self,
477    ) -> std::result::Result<(ServerToClientMsg, ErrorContext), IpcReceiveError> {
478        let proto_msg = read_protobuf_message::<ProtoServerToClientMsg>(&mut self.receiver)?;
479        match proto_msg.try_into() {
480            Ok(rust_msg) => Ok((rust_msg, ErrorContext::default())),
481            Err(e) => {
482                warn!("Error converting protobuf to ServerToClientMsg: {:?}", e);
483                Err(IpcReceiveError::Undecodable)
484            },
485        }
486    }
487
488    /// Returns an [`IpcSenderWithContext`] with the same socket as this receiver.
489    pub fn get_sender<F: Serialize>(&self) -> IpcSenderWithContext<F> {
490        let socket = self.receiver.get_ref().try_clone_stream().unwrap();
491        IpcSenderWithContext::from_boxed(socket)
492    }
493}
494
495// Protobuf wire format utilities
496fn read_protobuf_message<T: Message + Default>(
497    reader: &mut impl Read,
498) -> std::result::Result<T, IpcReceiveError> {
499    // Read length-prefixed protobuf message
500    let mut len_bytes = [0u8; 4];
501    reader
502        .read_exact(&mut len_bytes)
503        .map_err(|_| IpcReceiveError::Disconnected)?;
504    let len = u32::from_le_bytes(len_bytes) as usize;
505
506    let mut buf = vec![0u8; len];
507    reader
508        .read_exact(&mut buf)
509        .map_err(|_| IpcReceiveError::Disconnected)?;
510
511    T::decode(&buf[..]).map_err(|_| IpcReceiveError::Undecodable)
512}
513
514fn write_protobuf_message<T: Message>(writer: &mut impl Write, msg: &T) -> Result<()> {
515    let encoded = msg.encode_to_vec();
516    let len = encoded.len() as u32;
517
518    // we measure the length of the message and transmit it first so that the reader will be able
519    // to first read exactly 4 bytes (representing this length) and then read that amount of bytes
520    // as the actual message - this is so that we are able to distinct whole messages over the wire
521    // stream
522    writer.write_all(&len.to_le_bytes())?;
523    writer.write_all(&encoded)?;
524    Ok(())
525}
526
527// Protobuf helper functions
528pub fn send_protobuf_client_to_server(
529    sender: &mut IpcSenderWithContext<ClientToServerMsg>,
530    msg: ClientToServerMsg,
531) -> Result<()> {
532    let proto_msg: ProtoClientToServerMsg = msg.into();
533    write_protobuf_message(&mut sender.sender, &proto_msg)?;
534    let _ = sender.sender.flush();
535    Ok(())
536}
537
538pub fn send_protobuf_server_to_client(
539    sender: &mut IpcSenderWithContext<ServerToClientMsg>,
540    msg: ServerToClientMsg,
541) -> Result<()> {
542    let proto_msg: ProtoServerToClientMsg = msg.into();
543    write_protobuf_message(&mut sender.sender, &proto_msg)?;
544    let _ = sender.sender.flush();
545    Ok(())
546}
547
548pub fn recv_protobuf_client_to_server(
549    receiver: &mut IpcReceiverWithContext<ClientToServerMsg>,
550) -> Option<(ClientToServerMsg, ErrorContext)> {
551    receiver.try_recv_client_msg().ok()
552}
553
554pub fn recv_protobuf_server_to_client(
555    receiver: &mut IpcReceiverWithContext<ServerToClientMsg>,
556) -> Option<(ServerToClientMsg, ErrorContext)> {
557    receiver.try_recv_server_msg().ok()
558}
559
560/// Asynchronously send `ClientToServerMsg::KillSession` to the peer at `path`
561/// and wait until the peer's existing shutdown path replies (or its socket
562/// closes). Either of those outcomes confirms the kill landed; the caller
563/// wraps this in `tokio::time::timeout` to bound the wait against a wedged
564/// peer.
565///
566/// On Unix the local socket is bidirectional, so the same async stream is
567/// used for both send and receive. On Windows the named pipe is half-duplex
568/// and the existing sync `ipc_connect` / `ipc_connect_reply` flow is
569/// dispatched onto a blocking task.
570#[cfg(unix)]
571pub async fn async_send_kill_and_await(path: &std::path::Path) -> io::Result<()> {
572    use interprocess::local_socket::traits::tokio::Stream as _;
573    use interprocess::local_socket::{prelude::*, GenericFilePath};
574    use tokio::io::{AsyncReadExt, AsyncWriteExt};
575
576    let fs_name = path.to_fs_name::<GenericFilePath>()?;
577    let mut stream: interprocess::local_socket::tokio::Stream =
578        interprocess::local_socket::tokio::Stream::connect(fs_name).await?;
579
580    let proto_msg: ProtoClientToServerMsg = crate::ipc::ClientToServerMsg::KillSession.into();
581    let encoded = proto_msg.encode_to_vec();
582    let len_bytes = (encoded.len() as u32).to_le_bytes();
583
584    stream.write_all(&len_bytes).await?;
585    stream.write_all(&encoded).await?;
586    // Best-effort flush; failing here doesn't mean the kill failed.
587    let _ = stream.flush().await;
588
589    // The peer's shutdown path sends `ServerToClientMsg::Exit { Normal }`
590    // (zellij-server/src/lib.rs ServerInstruction::KillSession) over this
591    // same socket before exiting; if it dies without ACKing, the stream
592    // closes. Either outcome -- a successful 4-byte length-prefix read or a
593    // read error/EOF -- confirms the kill is no longer in flight.
594    let mut len_buf = [0u8; 4];
595    let _ = stream.read_exact(&mut len_buf).await;
596    Ok(())
597}
598
599#[cfg(windows)]
600pub async fn async_send_kill_and_await(path: &std::path::Path) -> io::Result<()> {
601    let path = path.to_path_buf();
602    tokio::task::spawn_blocking(move || {
603        use crate::consts::{ipc_connect, ipc_connect_reply};
604        let stream = ipc_connect(&path)?;
605        let reply = ipc_connect_reply(&path);
606        let mut sender = IpcSenderWithContext::<ClientToServerMsg>::new(stream);
607        sender
608            .send_client_msg(ClientToServerMsg::KillSession)
609            .map_err(|e| io::Error::new(io::ErrorKind::Other, e.to_string()))?;
610        if let Ok(reply_stream) = reply {
611            let mut receiver: IpcReceiverWithContext<ServerToClientMsg> =
612                IpcReceiverWithContext::new(reply_stream);
613            let _ = receiver.recv_server_msg();
614        }
615        Ok::<(), io::Error>(())
616    })
617    .await
618    .map_err(|e| io::Error::new(io::ErrorKind::Other, e.to_string()))?
619}
zellij_utils/ipc.rs

zellij_utils/
ipc.rs
