chore: persistent connection

Cargo.toml

@@ -8,6 +8,7 @@ actix = "0.13.5"
 actix-web = "4.9.0"
 clap = { version = "4.5.20", features = ["derive"] }
 env_logger = "0.11.5"
+futures-util = { version = "0.3.31", features = ["tokio-io"] }
 log = "0.4.22"
 reqwest = { version = "0.12.9", features = ["json", "stream"] }
 serde = { version = "1.0.215", features = ["derive"] }

src/agent/agent.rs

@@ -10,10 +10,7 @@ pub struct Agent {
 }
 
 impl Agent {
-    pub fn new(
-        llamacpp_client: LlamacppClient,
-        name: Option<String>,
-    ) -> Self {
+    pub fn new(llamacpp_client: LlamacppClient, name: Option<String>) -> Self {
         Self {
             id: uuid::Uuid::new_v4(),
             name,

src/agent/state_reporter.rs

@@ -1,69 +1,64 @@
-use actix::{
-    AsyncContext,
-    fut::future::WrapFuture,
-};
+use actix::{fut::future::WrapFuture, AsyncContext};
 use log::error;
 use std::sync::Arc;
-use tokio::time;
 use tokio::sync::broadcast;
+use tokio::time;
 
 use crate::balancer::status_update::StatusUpdate;
+use crate::errors::result::Result;
 
 #[allow(dead_code)]
 pub struct StateReporter {
-    management_addr: url::Url,
+    stats_endpoint_url: String,
     status_update_rx: broadcast::Receiver<actix_web::web::Bytes>,
     status_update_tx: Arc<broadcast::Sender<actix_web::web::Bytes>>,
 }
 
 impl StateReporter {
-    pub fn new(management_addr: url::Url) -> Self {
+    pub fn new(management_addr: url::Url) -> Result<Self> {
         let (tx, rx) = broadcast::channel(1);
 
-        Self {
-            management_addr,
+        Ok(Self {
+            stats_endpoint_url: management_addr.join("/stream")?.to_string(),
             status_update_rx: rx,
             status_update_tx: Arc::new(tx),
-        }
+        })
     }
 }
 
 impl actix::Actor for StateReporter {
     type Context = actix::Context<Self>;
 
     fn started(&mut self, ctx: &mut actix::Context<Self>) {
-        let management_addr = self.management_addr.clone();
+        let stats_endpoint_url = self.stats_endpoint_url.clone();
         let status_update_tx = self.status_update_tx.clone();
 
         ctx.spawn(
             async move {
-                let mut interval = time::interval(time::Duration::from_secs(1));
-
                 loop {
-                    interval.tick().await;
-
                     let rx = status_update_tx.subscribe();
                     let stream = tokio_stream::wrappers::BroadcastStream::new(rx);
                     let reqwest_body = reqwest::Body::wrap_stream(stream);
 
                     let result = reqwest::Client::new()
-                        .post(management_addr.clone())
+                        .post(stats_endpoint_url.clone())
                         .body(reqwest_body)
                         .send()
-                        .await
-                    ;
+                        .await;
 
                     match result {
                         Ok(_) => {
                             error!("Mangement server connection closed");
-                        },
+                        }
                         Err(err) => {
                             error!("Management server error: {}", err);
-                        },
+                        }
                     }
+
+                    time::sleep(time::Duration::from_secs(1)).await;
                 }
             }
-            .into_actor(self)
+            .into_actor(self),
         );
     }
 }
@@ -72,8 +67,6 @@ impl actix::Handler<StatusUpdate> for StateReporter {
     type Result = ();
 
     fn handle(&mut self, msg: StatusUpdate, _ctx: &mut actix::Context<Self>) {
-        println!("Received status update: {:?}", msg);
-
         serde_json::to_vec(&msg)
             .map(|bytes| actix_web::web::Bytes::from(bytes))
             .map_err(|err| {

src/balancer/http_route/mod.rs

@@ -0,0 +1 @@
+pub mod receive_status_update;

src/balancer/http_route/receive_status_update.rs

@@ -0,0 +1,17 @@
+use actix_web::{post, web, Error, HttpResponse};
+use futures_util::StreamExt as _;
+
+pub fn register(cfg: &mut web::ServiceConfig) {
+    cfg.service(respond);
+}
+
+#[post("/stream")]
+async fn respond(mut payload: web::Payload) -> Result<HttpResponse, Error> {
+    while let Some(chunk) = payload.next().await {
+        println!("Chunk: {:?}", chunk);
+    }
+
+    println!("Stream ended");
+
+    Ok(HttpResponse::Ok().finish())
+}

src/balancer/mod.rs

@@ -1 +1,2 @@
+pub mod http_route;
 pub mod status_update;

src/balancer/status_update.rs

@@ -1,4 +1,4 @@
-use serde::{Serialize, Deserialize};
+use serde::{Deserialize, Serialize};
 
 use crate::llamacpp::slot::Slot;
 
@@ -10,11 +10,7 @@ pub struct StatusUpdate {
 }
 
 impl StatusUpdate {
-    pub fn new(
-        agent_id: uuid::Uuid,
-        agent_name: Option<String>,
-        slots: Vec<Slot>
-    ) -> Self {
+    pub fn new(agent_id: uuid::Uuid, agent_name: Option<String>, slots: Vec<Slot>) -> Self {
         Self {
             agent_id,
             agent_name,