test the stats ws connection with monitor client

lib/monitor_client/Cargo.toml

@@ -12,7 +12,10 @@ license = "GPL v3.0"
 # monitor_types = "0.1.0"
 monitor_types = { path = "../types" }
 reqwest = { version = "0.11", features = ["json"] }
+tokio-tungstenite = { version = "0.18", features=["native-tls"] }
+tokio = "1.23"
 anyhow = "1.0"
 serde = "1.0"
 serde_json = "1.0"
-serde_derive = "1.0"
+serde_derive = "1.0"
+futures-util = "0.3"

lib/monitor_client/src/group.rs

@@ -1,5 +1,5 @@
 use anyhow::Context;
-use monitor_types::{Group, Update};
+use monitor_types::Group;
 use serde_json::{json, Value};
 
 use crate::MonitorClient;

lib/monitor_client/src/lib.rs

@@ -1,11 +1,14 @@
-#![allow(unused)]
+// #![allow(unused)]
 
 use anyhow::{anyhow, Context};
 use reqwest::StatusCode;
 use serde::{de::DeserializeOwned, Serialize};
+use serde_json::json;
+
+pub use futures_util;
+pub use tokio_tungstenite;
 
 pub use monitor_types as types;
-use serde_json::json;
 
 mod build;
 mod deployment;
@@ -229,7 +232,7 @@ impl MonitorClient {
         }
     }
 
-    async fn patch_string<B: Serialize>(
+    async fn _patch_string<B: Serialize>(
         &self,
         endpoint: &str,
         body: impl Into<Option<B>>,
@@ -287,7 +290,7 @@ impl MonitorClient {
         }
     }
 
-    async fn delete_string<B: Serialize>(
+    async fn _delete_string<B: Serialize>(
         &self,
         endpoint: &str,
         body: impl Into<Option<B>>,

lib/monitor_client/src/server.rs

@@ -1,9 +1,12 @@
-use anyhow::Context;
+use anyhow::{Context, anyhow};
+use futures_util::{SinkExt, StreamExt};
 use monitor_types::{
     BasicContainerInfo, ImageSummary, Log, Network, Server, ServerActionState, ServerWithStatus,
     SystemStats, SystemStatsQuery,
 };
 use serde_json::{json, Value};
+use tokio::net::TcpStream;
+use tokio_tungstenite::{connect_async, MaybeTlsStream, WebSocketStream, tungstenite::Message};
 
 use crate::MonitorClient;
 
@@ -88,6 +91,35 @@ impl MonitorClient {
             .context(format!("failed to get server stats at id {server_id}"))
     }
 
+    pub async fn subscribe_to_stats_ws(
+        &self,
+        server_id: &str,
+        query: impl Into<Option<SystemStatsQuery>>,
+    ) -> anyhow::Result<WebSocketStream<MaybeTlsStream<TcpStream>>> {
+        let query = query.into().unwrap_or_default();
+        let endpoint = format!(
+            "{}/ws/stats/{server_id}?networks={}&components={}&processes={}",
+            self.url.replace("http", "ws"),
+            query.networks,
+            query.components,
+            query.processes
+        );
+        let (mut socket, _) = connect_async(endpoint).await?;
+        socket.send(Message::Text(self.token.clone())).await?;
+        let msg = socket.next().await;
+        if let Some(Ok(Message::Text(msg))) = &msg {
+            if msg.as_str() == "LOGGED_IN" {
+                Ok(socket)
+            } else {
+                Err(anyhow!("failed to log in"))
+            }
+        } else if let Some(Err(e)) = &msg {
+            Err(anyhow!("error on connection: {e:?}"))
+        } else {
+            Err(anyhow!("some other failure"))
+        }
+    }
+
     pub async fn get_docker_networks(&self, server_id: &str) -> anyhow::Result<Vec<Network>> {
         self.get(
             &format!("/api/server/{server_id}/networks"),

lib/types/src/server.rs

@@ -147,6 +147,10 @@ impl SystemStatsQuery {
             processes: true,
         }
     }
+
+    pub fn none() -> SystemStatsQuery {
+        Default::default()
+    }
 }
 
 #[typeshare]