display number of watchers per stream in the watch ui

3 files changed, 51 insertions, 7 deletions

diff --git a/src/cmd/watch.rs b/src/cmd/watch.rs
index d400cfc..2a7676e 100644
--- a/src/cmd/watch.rs
+++ b/src/cmd/watch.rs
@@ -440,6 +440,8 @@ impl<S: tokio::io::AsyncRead + tokio::io::AsyncWrite + Send + 'static>
         let idle_width = format_time(max_idle_time).len();
         let idle_width = if idle_width < 4 { 4 } else { idle_width };
 
+        let watch_width = 5;
+
         let max_title_width = (sessions.size().cols as usize)
             - char_width
             - 3
@@ -448,6 +450,8 @@ impl<S: tokio::io::AsyncRead + tokio::io::AsyncWrite + Send + 'static>
             - size_width
             - 3
             - idle_width
+            - 3
+            - watch_width
             - 3;
 
         crossterm::terminal()
@@ -457,15 +461,17 @@ impl<S: tokio::io::AsyncRead + tokio::io::AsyncWrite + Send + 'static>
         println!("available sessions:\r");
         println!("\r");
         println!(
-            "{:4$} | {:5$} | {:6$} | {:7$} | title\r",
+            "{:5$} | {:6$} | {:7$} | {:8$} | {:9$} | title\r",
             "",
             "name",
             "size",
             "idle",
+            "watch",
             char_width,
             name_width,
             size_width,
-            idle_width
+            idle_width,
+            watch_width,
         );
         println!("{}\r", "-".repeat(sessions.size().cols as usize));
 
@@ -497,19 +503,22 @@ impl<S: tokio::io::AsyncRead + tokio::io::AsyncWrite + Send + 'static>
             };
             let display_idle = format_time(session.idle_time);
             let display_title = truncate(&session.title, max_title_width);
+            let display_watch = session.watchers;
 
             println!(
-                "{:5$} | {:6$} | {:7$} | {:8$} | {}\r",
+                "{:6$} | {:7$} | {:8$} | {:9$} | {:10$} | {}\r",
                 display_char,
                 display_name,
                 display_size_full,
                 display_idle,
+                display_watch,
                 display_title,
                 char_width,
                 name_width,
                 size_width
                     + (display_size_full.len() - display_size_plain.len()),
-                idle_width
+                idle_width,
+                watch_width,
             );
 
             prev_name = Some(&session.username);
diff --git a/src/protocol.rs b/src/protocol.rs
index 1e09f0a..381883e 100644
--- a/src/protocol.rs
+++ b/src/protocol.rs
@@ -12,6 +12,7 @@ pub struct Session {
     pub size: crate::term::Size,
     pub idle_time: u32,
     pub title: String,
+    pub watchers: u32,
 }
 
 pub struct FramedReader<T: tokio::io::AsyncRead>(
@@ -375,6 +376,7 @@ impl From<&Message> for Packet {
             write_size(&val.size, data);
             write_u32(val.idle_time, data);
             write_str(&val.title, data);
+            write_u32(val.watchers, data);
         }
         fn write_sessions(val: &[Session], data: &mut Vec<u8>) {
             write_u32(u32_from_usize(val.len()), data);
@@ -589,6 +591,7 @@ impl std::convert::TryFrom<Packet> for Message {
             let (size, data) = read_size(data)?;
             let (idle_time, data) = read_u32(data)?;
             let (title, data) = read_str(data)?;
+            let (watchers, data) = read_u32(data)?;
             Ok((
                 Session {
                     id,
@@ -597,6 +600,7 @@ impl std::convert::TryFrom<Packet> for Message {
                     size,
                     idle_time,
                     title,
+                    watchers,
                 },
                 data,
             ))
@@ -822,6 +826,7 @@ mod test {
                 size: crate::term::Size { rows: 24, cols: 80 },
                 idle_time: 123,
                 title: "it's my terminal title".to_string(),
+                watchers: 0,
             }]),
             Message::sessions(&[
                 Session {
@@ -831,6 +836,7 @@ mod test {
                     size: crate::term::Size { rows: 24, cols: 80 },
                     idle_time: 123,
                     title: "it's my terminal title".to_string(),
+                    watchers: 0,
                 },
                 Session {
                     id: "some-other-session-id".to_string(),
@@ -839,6 +845,7 @@ mod test {
                     size: crate::term::Size { rows: 24, cols: 80 },
                     idle_time: 68,
                     title: "some other terminal title".to_string(),
+                    watchers: 0,
                 },
             ]),
             Message::disconnected(),
diff --git a/src/server.rs b/src/server.rs
index 42b4c66..be37b82 100644
--- a/src/server.rs
+++ b/src/server.rs
@@ -264,7 +264,7 @@ impl<S: tokio::io::AsyncRead + tokio::io::AsyncWrite + Send + 'static>
         }
     }
 
-    fn session(&self) -> Option<crate::protocol::Session> {
+    fn session(&self, watchers: u32) -> Option<crate::protocol::Session> {
         let (username, term_info) = match &self.state {
             ConnectionState::Accepted => return None,
             ConnectionState::LoggingIn { .. } => return None,
@@ -300,6 +300,7 @@ impl<S: tokio::io::AsyncRead + tokio::io::AsyncWrite + Send + 'static>
                 .duration_since(self.last_activity)
                 .as_secs() as u32,
             title: title.to_string(),
+            watchers,
         })
     }
 
@@ -561,8 +562,28 @@ impl<S: tokio::io::AsyncRead + tokio::io::AsyncWrite + Send + 'static>
         &mut self,
         conn: &mut Connection<S>,
     ) -> Result<()> {
-        let sessions: Vec<_> =
-            self.streamers().flat_map(Connection::session).collect();
+        let mut watcher_counts = std::collections::HashMap::new();
+        for watcher in self.watchers() {
+            let watch_id =
+                if let ConnectionState::Watching { watch_id, .. } =
+                    &watcher.state
+                {
+                    watch_id
+                } else {
+                    unreachable!()
+                };
+            watcher_counts.insert(
+                watch_id,
+                *watcher_counts.get(&watch_id).unwrap_or(&0) + 1,
+            );
+        }
+        let sessions: Vec<_> = self
+            .streamers()
+            .flat_map(|streamer| {
+                streamer
+                    .session(*watcher_counts.get(&streamer.id).unwrap_or(&0))
+            })
+            .collect();
         conn.send_message(crate::protocol::Message::sessions(&sessions));
 
         Ok(())
@@ -889,6 +910,13 @@ impl<S: tokio::io::AsyncRead + tokio::io::AsyncWrite + Send + 'static>
         })
     }
 
+    fn watchers(&self) -> impl Iterator<Item = &Connection<S>> {
+        self.connections.values().filter(|conn| match conn.state {
+            ConnectionState::Watching { .. } => true,
+            _ => false,
+        })
+    }
+
     fn watchers_mut(&mut self) -> impl Iterator<Item = &mut Connection<S>> {
         self.connections
             .values_mut()