feat(api): add sse

libs/api/src/path.rs

@@ -22,3 +22,7 @@ pub fn streams(stream: &str) -> String {
 pub fn cascade(stream: &str) -> String {
     format!("/api/cascade/{}", stream)
 }
+
+pub fn streams_sse() -> String {
+    "/api/sse/streams".to_string()
+}

libs/api/src/request.rs

@@ -28,3 +28,9 @@ pub struct Cascade {
     // push mode ,value : whip_url
     pub target_url: Option<String>,
 }
+
+#[derive(Serialize, Deserialize, Clone)]
+pub struct StreamSSE {
+    #[serde(default)]
+    pub streams: Vec<String>,
+}

liveion/Cargo.toml

@@ -20,6 +20,8 @@ anyhow = { workspace = true, features = ["backtrace"] }
 clap = { workspace = true, features = ["derive"] }
 serde = { workspace = true, features = ["serde_derive"] }
 tokio = { workspace = true, features = ["full"] }
+tokio-stream = "0.1.15"
+async-stream = "0.3.5"
 tracing = { workspace = true }
 webrtc = { workspace = true }
 
@@ -40,8 +42,9 @@ serde_json = "1.0.114"
 toml = "0.8.10"
 tower-http = { version = "0.5.2", features = ["fs", "auth", "trace", "cors"] }
 
-reqwest = { version = "0.12", optional = true, features = ["rustls-tls"], default-features = false }
+reqwest = { version = "0.12", optional = true, features = [
+    "rustls-tls",
+], default-features = false }
 
 [features]
 webhook = ["dep:reqwest"]
-

liveion/src/forward/mod.rs

@@ -36,13 +36,15 @@ pub(crate) fn get_peer_id(peer: &Arc<RTCPeerConnection>) -> String {
 
 #[derive(Clone)]
 pub struct PeerForward {
+    pub(crate) stream: String,
     publish_lock: Arc<Mutex<()>>,
     internal: Arc<PeerForwardInternal>,
 }
 
 impl PeerForward {
     pub fn new(stream: impl ToString, ice_server: Vec<RTCIceServer>) -> Self {
         PeerForward {
+            stream: stream.to_string(),
             publish_lock: Arc::new(Mutex::new(())),
             internal: Arc::new(PeerForwardInternal::new(stream, ice_server)),
         }

liveion/src/lib.rs

@@ -12,7 +12,7 @@ use tokio::net::TcpListener;
 use tower_http::cors::CorsLayer;
 use tower_http::trace::TraceLayer;
 use tower_http::validate_request::ValidateRequestHeaderLayer;
-use tracing::{error, info_span};
+use tracing::{error, info_span, Level};
 
 use crate::auth::ManyValidate;
 use crate::config::Config;
@@ -63,21 +63,23 @@ where
         } else {
             CorsLayer::new()
         })
-        .layer(axum::middleware::from_fn(http_log::print_request_response))
         .layer(
-            TraceLayer::new_for_http().make_span_with(|request: &Request<_>| {
-                let span = info_span!(
-                    "http_request",
-                    uri = ?request.uri(),
-                    method = ?request.method(),
-                    span_id = tracing::field::Empty,
-                );
-                span.record(
-                    "span_id",
-                    span.id().unwrap_or(tracing::Id::from_u64(42)).into_u64(),
-                );
-                span
-            }),
+            TraceLayer::new_for_http()
+                .make_span_with(|request: &Request<_>| {
+                    let span = info_span!(
+                        "http_request",
+                        uri = ?request.uri(),
+                        method = ?request.method(),
+                        span_id = tracing::field::Empty,
+                    );
+                    span.record(
+                        "span_id",
+                        span.id().unwrap_or(tracing::Id::from_u64(42)).into_u64(),
+                    );
+                    span
+                })
+                .on_response(tower_http::trace::DefaultOnResponse::new().level(Level::INFO))
+                .on_failure(tower_http::trace::DefaultOnFailure::new().level(Level::INFO)),
         );
 
     app = app.fallback(static_handler);

liveion/src/route/stream.rs

@@ -1,19 +1,24 @@
+use std::convert::Infallible;
+
+use crate::error::AppError;
+use crate::AppState;
 use axum::extract::{Path, State};
-use axum::response::Response;
+use axum::response::sse::{Event, KeepAlive};
+use axum::response::{Response, Sse};
 use axum::routing::{delete, get, post};
 use axum::{Json, Router};
 use axum_extra::extract::Query;
 use http::StatusCode;
-
-use crate::error::AppError;
-use crate::AppState;
+use tokio_stream::wrappers::ReceiverStream;
+use tokio_stream::StreamExt;
 
 pub fn route() -> Router<AppState> {
     Router::new()
         .route(&api::path::streams(""), get(index))
         .route(&api::path::streams(":stream"), get(show))
         .route(&api::path::streams(":stream"), post(create))
         .route(&api::path::streams(":stream"), delete(destroy))
+        .route(&api::path::streams_sse(), get(sse))
 }
 
 async fn index(
@@ -72,3 +77,27 @@ async fn destroy(
         Err(e) => Err(AppError::StreamNotFound(e.to_string())),
     }
 }
+
+async fn sse(
+    State(state): State<AppState>,
+    Query(req): Query<api::request::StreamSSE>,
+) -> crate::result::Result<
+    Sse<impl tokio_stream::Stream<Item = Result<axum::response::sse::Event, Infallible>>>,
+> {
+    let recv = state
+        .stream_manager
+        .sse_handler(req.streams.clone())
+        .await?;
+    let stream = ReceiverStream::new(recv).map(|forward_infos| {
+        Ok(Event::default()
+            .json_data(
+                forward_infos
+                    .into_iter()
+                    .map(api::response::Stream::from)
+                    .collect::<Vec<_>>(),
+            )
+            .unwrap())
+    });
+    let resp = Sse::new(stream).keep_alive(KeepAlive::default());
+    Ok(resp)
+}

liveion/src/stream/manager.rs

@@ -460,4 +460,34 @@ impl Manager {
 
         Ok(())
     }
+
+    pub async fn sse_handler(
+        &self,
+        streams: Vec<String>,
+    ) -> Result<tokio::sync::mpsc::Receiver<Vec<ForwardInfo>>> {
+        let (send, recv) = tokio::sync::mpsc::channel(64);
+        let mut evnet_recv = self.event_sender.subscribe();
+        let stream_map = self.stream_map.clone();
+        tokio::spawn(async move {
+            while let Ok(event) = evnet_recv.recv().await {
+                let stream = match event {
+                    Event::Node(_) => continue,
+                    Event::Stream(val) => val.stream.stream,
+                    Event::Forward(val) => val.stream_info.id,
+                };
+                if streams.is_empty() || streams.contains(&stream) {
+                    let stream_map = stream_map.read().await;
+                    let mut infos = vec![];
+                    for (_, forward) in stream_map.iter() {
+                        if !streams.is_empty() && !streams.contains(&forward.stream) {
+                            continue;
+                        }
+                        infos.push(forward.info().await);
+                    }
+                    let _ = send.send(infos).await;
+                }
+            }
+        });
+        Ok(recv)
+    }
 }