mux: track list of clients

Define a way to compute a client ID and pass that through to the
mux server when verifying version compatibility.

Once associated, the session handler will keep some metadata
updated in the mux.

A new cli subcommand exposes the info:

```
; ./target/debug/wezterm cli list-clients
USER HOST              PID CONNECTED  IDLE      WORKSPACE
wez  mba.localdomain 52979 30.009225s 1.009225s
```

refs: #1531

Cargo.lock

@@ -533,6 +533,7 @@ dependencies = [
  "num-integer",
  "num-traits",
  "pure-rust-locales",
+ "serde",
  "time",
  "winapi 0.3.9",
 ]
@@ -2290,10 +2291,12 @@ dependencies = [
  "async-trait",
  "base64",
  "bintree",
+ "chrono",
  "config",
  "crossbeam",
  "downcast-rs",
  "filedescriptor",
+ "hostname",
  "k9",
  "lazy_static",
  "libc",
@@ -4614,6 +4617,7 @@ name = "wezterm"
 version = "0.1.0"
 dependencies = [
  "anyhow",
+ "chrono",
  "codec",
  "config",
  "env-bootstrap",

codec/src/lib.rs

@@ -12,7 +12,7 @@
 #![cfg_attr(feature = "cargo-clippy", allow(clippy::range_plus_one))]
 
 use anyhow::{bail, Context as _, Error};
-use leb128;
+use mux::client::{ClientId, ClientInfo};
 use mux::domain::DomainId;
 use mux::pane::PaneId;
 use mux::renderable::{RenderableDimensions, StableCursorPosition};
@@ -406,7 +406,7 @@ macro_rules! pdu {
 /// The overall version of the codec.
 /// This must be bumped when backwards incompatible changes
 /// are made to the types and protocol.
-pub const CODEC_VERSION: usize = 13;
+pub const CODEC_VERSION: usize = 14;
 
 // Defines the Pdu enum.
 // Each struct has an explicit identifying number.
@@ -445,6 +445,9 @@ pdu! {
     PaneRemoved: 37,
     SetPalette: 38,
     NotifyAlert: 39,
+    SetClientId: 40,
+    GetClientList: 41,
+    GetClientListResponse: 42,
 }
 
 impl Pdu {
@@ -697,6 +700,19 @@ pub struct NotifyAlert {
     pub alert: Alert,
 }
 
+#[derive(Deserialize, Serialize, PartialEq, Debug)]
+pub struct SetClientId {
+    pub client_id: ClientId,
+}
+
+#[derive(Deserialize, Serialize, PartialEq, Debug)]
+pub struct GetClientList;
+
+#[derive(Deserialize, Serialize, PartialEq, Debug)]
+pub struct GetClientListResponse {
+    pub clients: Vec<ClientInfo>,
+}
+
 #[derive(Deserialize, Serialize, PartialEq, Debug)]
 pub struct Resize {
     pub containing_tab_id: TabId,

mux/Cargo.toml

@@ -11,10 +11,12 @@ anyhow = "1.0"
 async-trait = "0.1"
 base64 = "0.13"
 bintree = { path = "../bintree" }
+chrono = { version = "0.4", features = ["serde"] }
 config = { path = "../config" }
 crossbeam = "0.8"
 downcast-rs = "1.0"
 filedescriptor = { version="0.8", path = "../filedescriptor" }
+hostname = "0.3"
 lazy_static = "1.4"
 libc = "0.2"
 log = "0.4"

mux/src/client.rs

@@ -0,0 +1,64 @@
+use chrono::serde::ts_seconds;
+use chrono::{DateTime, Utc};
+use serde::*;
+use std::sync::atomic::{AtomicUsize, Ordering};
+use std::time::SystemTime;
+
+static CLIENT_ID: AtomicUsize = AtomicUsize::new(0);
+lazy_static::lazy_static! {
+    static ref EPOCH: u64 = SystemTime::now()
+                                .duration_since(SystemTime::UNIX_EPOCH)
+                                .unwrap().as_secs();
+}
+
+#[derive(Serialize, Deserialize, Clone, Debug, PartialEq, Eq, Hash)]
+pub struct ClientId {
+    pub hostname: String,
+    pub username: String,
+    pub pid: u32,
+    pub epoch: u64,
+    pub id: usize,
+}
+
+impl ClientId {
+    pub fn new() -> Self {
+        let id = CLIENT_ID.fetch_add(1, Ordering::Relaxed);
+        Self {
+            hostname: hostname::get()
+                .map(|s| s.to_string_lossy().to_string())
+                .unwrap_or_else(|_| "localhost".to_string()),
+            username: config::username_from_env().unwrap_or_else(|_| "somebody".to_string()),
+            pid: unsafe { libc::getpid() as u32 },
+            epoch: *EPOCH,
+            id,
+        }
+    }
+}
+
+#[derive(Deserialize, Serialize, PartialEq, Debug, Clone)]
+pub struct ClientInfo {
+    pub client_id: ClientId,
+    /// The time this client last connected
+    #[serde(with = "ts_seconds")]
+    pub connected_at: DateTime<Utc>,
+    /// Which workspace is active
+    pub active_workspace: Option<String>,
+    /// The last time we received input from this client
+    #[serde(with = "ts_seconds")]
+    pub last_input: DateTime<Utc>,
+}
+
+impl ClientInfo {
+    pub fn new(client_id: &ClientId) -> Self {
+        Self {
+            client_id: client_id.clone(),
+            connected_at: Utc::now(),
+            active_workspace: None,
+            last_input: Utc::now(),
+        }
+    }
+
+    pub fn update_last_input(&mut self) {
+        self.last_input = Utc::now();
+    }
+}

mux/src/lib.rs

@@ -1,3 +1,4 @@
+use crate::client::{ClientId, ClientInfo};
 use crate::pane::{Pane, PaneId};
 use crate::tab::{Tab, TabId};
 use crate::window::{Window, WindowId};
@@ -25,6 +26,7 @@ use thiserror::*;
 use winapi::um::winsock2::{SOL_SOCKET, SO_RCVBUF, SO_SNDBUF};
 
 pub mod activity;
+pub mod client;
 pub mod connui;
 pub mod domain;
 pub mod localpane;
@@ -64,6 +66,7 @@ pub struct Mux {
     domains_by_name: RefCell<HashMap<String, Arc<dyn Domain>>>,
     subscribers: RefCell<HashMap<usize, Box<dyn Fn(MuxNotification) -> bool>>>,
     banner: RefCell<Option<String>>,
+    clients: RefCell<HashMap<ClientId, ClientInfo>>,
 }
 
 const BUFSIZE: usize = 1024 * 1024;
@@ -317,9 +320,34 @@ impl Mux {
             domains: RefCell::new(domains),
             subscribers: RefCell::new(HashMap::new()),
             banner: RefCell::new(None),
+            clients: RefCell::new(HashMap::new()),
         }
     }
 
+    pub fn client_had_input(&self, client_id: &ClientId) {
+        if let Some(info) = self.clients.borrow_mut().get_mut(client_id) {
+            info.update_last_input();
+        }
+    }
+
+    pub fn register_client(&self, client_id: &ClientId) {
+        self.clients
+            .borrow_mut()
+            .insert(client_id.clone(), ClientInfo::new(client_id));
+    }
+
+    pub fn iter_clients(&self) -> Vec<ClientInfo> {
+        self.clients
+            .borrow()
+            .values()
+            .map(|info| info.clone())
+            .collect()
+    }
+
+    pub fn unregister_client(&self, client_id: &ClientId) {
+        self.clients.borrow_mut().remove(client_id);
+    }
+
     pub fn subscribe<F>(&self, subscriber: F)
     where
         F: Fn(MuxNotification) -> bool + 'static,

wezterm-client/src/client.rs

@@ -8,6 +8,7 @@ use codec::*;
 use config::{configuration, SshDomain, TlsDomainClient, UnixDomain, UnixTarget};
 use filedescriptor::FileDescriptor;
 use futures::FutureExt;
+use mux::client::ClientId;
 use mux::connui::ConnectionUI;
 use mux::domain::DomainId;
 use mux::pane::PaneId;
@@ -41,6 +42,7 @@ enum ReaderMessage {
 pub struct Client {
     sender: Sender<ReaderMessage>,
     local_domain_id: Option<DomainId>,
+    client_id: ClientId,
     pub is_reconnectable: bool,
     pub is_local: bool,
 }
@@ -865,6 +867,7 @@ impl Client {
         let is_reconnectable = reconnectable.reconnectable();
         let is_local = reconnectable.is_local();
         let (sender, mut receiver) = unbounded();
+        let client_id = ClientId::new();
 
         thread::spawn(move || {
             const BASE_INTERVAL: Duration = Duration::from_secs(1);
@@ -957,6 +960,7 @@ impl Client {
             local_domain_id,
             is_reconnectable,
             is_local,
+            client_id,
         }
     }
 
@@ -971,6 +975,10 @@ impl Client {
                     info.version_string,
                     info.codec_vers
                 );
+                self.set_client_id(SetClientId {
+                    client_id: self.client_id.clone(),
+                })
+                .await?;
                 Ok(info)
             }
             Ok(info) => {
@@ -1117,4 +1125,6 @@ impl Client {
         SearchScrollbackResponse
     );
     rpc!(kill_pane, KillPane, UnitResponse);
+    rpc!(set_client_id, SetClientId, UnitResponse);
+    rpc!(list_clients, GetClientList, GetClientListResponse);
 }

wezterm-mux-server-impl/src/sessionhandler.rs

@@ -2,6 +2,7 @@ use crate::PKI;
 use anyhow::{anyhow, Context};
 use codec::*;
 use config::keyassignment::SpawnTabDomain;
+use mux::client::ClientId;
 use mux::pane::{Pane, PaneId};
 use mux::renderable::{RenderableDimensions, StableCursorPosition};
 use mux::tab::TabId;
@@ -196,6 +197,16 @@ fn maybe_push_pane_changes(
 pub struct SessionHandler {
     to_write_tx: PduSender,
     per_pane: HashMap<TabId, Arc<Mutex<PerPane>>>,
+    client_id: Option<ClientId>,
+}
+
+impl Drop for SessionHandler {
+    fn drop(&mut self) {
+        if let Some(client_id) = self.client_id.take() {
+            let mux = Mux::get().unwrap();
+            mux.unregister_client(&client_id);
+        }
+    }
 }
 
 impl SessionHandler {
@@ -214,6 +225,7 @@ impl SessionHandler {
         Self {
             to_write_tx,
             per_pane: HashMap::new(),
+            client_id: None,
         }
     }
 
@@ -244,6 +256,10 @@ impl SessionHandler {
         let sender = self.to_write_tx.clone();
         let serial = decoded.serial;
 
+        if let Some(client_id) = &self.client_id {
+            Mux::get().unwrap().client_had_input(client_id);
+        }
+
         let send_response = move |result: anyhow::Result<Pdu>| {
             let pdu = match result {
                 Ok(pdu) => pdu,
@@ -265,6 +281,30 @@ impl SessionHandler {
 
         match decoded.pdu {
             Pdu::Ping(Ping {}) => send_response(Ok(Pdu::Pong(Pong {}))),
+            Pdu::SetClientId(SetClientId { client_id }) => {
+                self.client_id.replace(client_id.clone());
+                spawn_into_main_thread(async move {
+                    let mux = Mux::get().unwrap();
+                    mux.register_client(&client_id);
+                })
+                .detach();
+                send_response(Ok(Pdu::UnitResponse(UnitResponse {})))
+            }
+            Pdu::GetClientList(GetClientList) => {
+                spawn_into_main_thread(async move {
+                    catch(
+                        move || {
+                            let mux = Mux::get().unwrap();
+                            let clients = mux.iter_clients();
+                            Ok(Pdu::GetClientListResponse(GetClientListResponse {
+                                clients,
+                            }))
+                        },
+                        send_response,
+                    )
+                })
+                .detach();
+            }
             Pdu::ListPanes(ListPanes {}) => {
                 spawn_into_main_thread(async move {
                     catch(
@@ -593,6 +633,7 @@ impl SessionHandler {
             | Pdu::GetLinesResponse { .. }
             | Pdu::GetCodecVersionResponse { .. }
             | Pdu::GetTlsCredsResponse { .. }
+            | Pdu::GetClientListResponse { .. }
             | Pdu::PaneRemoved { .. }
             | Pdu::ErrorResponse { .. } => {
                 send_response(Err(anyhow!("expected a request, got {:?}", decoded.pdu)))

wezterm/Cargo.toml

@@ -11,6 +11,7 @@ anyhow = "1.0"
 
 [dependencies]
 anyhow = "1.0"
+chrono = "0.4"
 codec = { path = "../codec" }
 config = { path = "../config" }
 env-bootstrap = { path = "../env-bootstrap" }

wezterm/src/main.rs

@@ -1,4 +1,5 @@
 use anyhow::{anyhow, Context};
+use chrono::{DateTime, Utc};
 use config::keyassignment::SpawnTabDomain;
 use config::wezterm_version;
 use mux::activity::Activity;
@@ -112,6 +113,9 @@ enum CliSubCommand {
     #[structopt(name = "list", about = "list windows, tabs and panes")]
     List,
 
+    #[structopt(name = "list-clients", about = "list clients")]
+    ListClients,
+
     #[structopt(name = "proxy", about = "start rpc proxy pipe")]
     Proxy,
 
@@ -400,6 +404,60 @@ async fn run_cli_async(config: config::ConfigHandle, cli: CliCommand) -> anyhow:
     )?;
 
     match cli.sub {
+        CliSubCommand::ListClients => {
+            let cols = vec![
+                Column {
+                    name: "USER".to_string(),
+                    alignment: Alignment::Left,
+                },
+                Column {
+                    name: "HOST".to_string(),
+                    alignment: Alignment::Left,
+                },
+                Column {
+                    name: "PID".to_string(),
+                    alignment: Alignment::Right,
+                },
+                Column {
+                    name: "CONNECTED".to_string(),
+                    alignment: Alignment::Left,
+                },
+                Column {
+                    name: "IDLE".to_string(),
+                    alignment: Alignment::Left,
+                },
+                Column {
+                    name: "WORKSPACE".to_string(),
+                    alignment: Alignment::Left,
+                },
+            ];
+            let mut data = vec![];
+            let clients = client.list_clients(codec::GetClientList).await?;
+            let now: DateTime<Utc> = Utc::now();
+
+            fn duration_string(d: chrono::Duration) -> String {
+                if let Ok(d) = d.to_std() {
+                    format!("{:?}", d)
+                } else {
+                    d.to_string()
+                }
+            }
+
+            for info in clients.clients {
+                let connected = now - info.connected_at;
+                let idle = now - info.last_input;
+                data.push(vec![
+                    info.client_id.username.to_string(),
+                    info.client_id.hostname.to_string(),
+                    info.client_id.pid.to_string(),
+                    duration_string(connected),
+                    duration_string(idle),
+                    info.active_workspace.as_deref().unwrap_or("").to_string(),
+                ]);
+            }
+
+            tabulate_output(&cols, &data, &mut std::io::stdout().lock())?;
+        }
         CliSubCommand::List => {
             let cols = vec![
                 Column {