feat/upload progress sse

Cargo.toml

@@ -49,7 +49,7 @@ opentelemetry-prometheus = { version = "0.16.0", optional = true }
 tikv-jemallocator = { version = "0.6.0", optional = true }
 tokio = { version = "1.48.0", features = ["full"] }
 tokio-rustls = { version = "0.26.2", optional = true }
-tokio-stream = "0.1.17"
+tokio-stream = { version = "0.1.17", features = ["sync"] }
 tokio-tungstenite = "0.27.0"
 tokio-util = { version = "0.7.15", features = ["compat"] }
 tracing = "0.1.41"

src/plugins.rs

@@ -65,6 +65,9 @@ pub mod metrics;
 /// Idempotency-Key based request de-duplication plugin.
 pub mod idempotency;
 
+/// Upload progress tracking plugin for monitoring file uploads.
+pub mod upload_progress;
+
 /// Trait for implementing Tako framework plugins.
 ///
 /// Plugins extend the framework's functionality by implementing this trait. They can

src/plugins/upload_progress.rs

@@ -0,0 +1,67 @@
+use std::sync::Arc;
+
+use dashmap::DashMap;
+use http::Method;
+use tokio::sync::broadcast;
+use uuid::Uuid;
+
+use crate::plugins::TakoPlugin;
+
+#[derive(Clone)]
+pub struct Config<'a> {
+  pub path: &'a str,
+}
+
+#[derive(Clone)]
+pub struct UploadProgress {
+  pub id: Uuid,
+  pub bytes: u64,
+  pub total: Option<u64>,
+}
+
+#[derive(Clone)]
+pub struct UploadProgressHub {
+  inner: Arc<DashMap<Uuid, broadcast::Sender<UploadProgress>>>,
+}
+
+impl UploadProgressHub {
+  pub fn new() -> Self {
+    Self {
+      inner: Arc::new(DashMap::new()),
+    }
+  }
+
+  pub fn register(&self, id: Uuid) -> broadcast::Receiver<UploadProgress> {
+    let (tx, rx) = broadcast::channel(100);
+    self.inner.insert(id, tx);
+    rx
+  }
+
+  pub fn subscribe(&self, id: &Uuid) -> Option<broadcast::Receiver<UploadProgress>> {
+    self.inner.get(&id).map(|e| e.value().subscribe())
+  }
+
+  pub fn notify(&self, msg: UploadProgress) {
+    if let Some(tx) = self.inner.get(&msg.id) {
+      let _ = tx.send(msg);
+    }
+  }
+}
+
+pub struct UploadProgressPlugin<'a>(Config<'a>);
+
+impl<'a> UploadProgressPlugin<'a> {
+  pub fn new(config: Config<'a>) -> Self {
+    Self(config)
+  }
+}
+
+impl TakoPlugin for UploadProgressPlugin<'static> {
+  fn name(&self) -> &'static str {
+    "UploadProgressPlugin"
+  }
+
+  fn setup(&self, router: &crate::router::Router) -> anyhow::Result<()> {
+    Ok(())
+  }
+}