Add log information via web api

Cargo.toml

@@ -35,6 +35,7 @@ async-trait = "0.1.41"
 actix-files = "0.6.5"
 actix-web = { version = "4.6.0", features = ["rustls"] }
 actix-web-validator = "5.0.1"
+actix-ws = "0.3.0"
 actix-service = "2.0.2"
 actix-extensible-rate-limit = "0.3.1"
 actix-cors = "0.7.0"
@@ -62,6 +63,7 @@ tracing-log = "0.2.0"
 tracing-actix-web = "0.7.10"
 tracing-gstreamer = "0.7.0"
 tracing-tracy = { version = "0.10.5", features = ["ondemand", "broadcast"] } # Note that these features defines the tracy client behavior.
+ringbuffer = "0.15.0"
 
 # This is a custom revision adding file name suffix on top of tracing-appender-0.2.2" "0.2.2", but compatible with the current tracing version.
 # Reference: https://github.com/tokio-rs/tracing/issues/2441

src/lib/logger/manager.rs

@@ -1,9 +1,87 @@
+use std::{
+    io::{self, Write},
+    sync::{Arc, Mutex},
+};
+
+use ringbuffer::{AllocRingBuffer, RingBuffer};
+use tokio::sync::broadcast::{Receiver, Sender};
 use tracing::{metadata::LevelFilter, *};
 use tracing_log::LogTracer;
-use tracing_subscriber::{fmt, layer::SubscriberExt, EnvFilter, Layer};
+use tracing_subscriber::{
+    fmt::{self, MakeWriter},
+    layer::SubscriberExt,
+    EnvFilter, Layer,
+};
 
 use crate::cli;
 
+struct BroadcastWriter {
+    sender: Sender<String>,
+}
+
+impl Write for BroadcastWriter {
+    fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> {
+        let message = String::from_utf8_lossy(buf).to_string();
+        let _ = self.sender.send(message);
+        Ok(buf.len())
+    }
+
+    fn flush(&mut self) -> io::Result<()> {
+        Ok(())
+    }
+}
+
+struct BroadcastMakeWriter {
+    sender: Sender<String>,
+}
+
+impl<'a> MakeWriter<'a> for BroadcastMakeWriter {
+    type Writer = BroadcastWriter;
+
+    fn make_writer(&'a self) -> Self::Writer {
+        BroadcastWriter {
+            sender: self.sender.clone(),
+        }
+    }
+}
+
+#[derive(Debug, Default)]
+pub struct Manager {
+    pub process: Option<tokio::task::JoinHandle<()>>,
+}
+
+pub struct History {
+    pub history: AllocRingBuffer<String>,
+    pub sender: Sender<String>,
+}
+
+impl Default for History {
+    fn default() -> Self {
+        let (sender, _receiver) = tokio::sync::broadcast::channel(100);
+        Self {
+            history: AllocRingBuffer::new(10 * 1024),
+            sender,
+        }
+    }
+}
+
+impl History {
+    pub fn push(&mut self, message: String) {
+        self.history.push(message.clone());
+        let _ = self.sender.send(message);
+    }
+
+    pub fn subscribe(&self) -> (Receiver<String>, Vec<String>) {
+        let reader = self.sender.subscribe();
+        (reader, self.history.to_vec())
+    }
+}
+
+lazy_static! {
+    static ref MANAGER: Arc<Mutex<Manager>> = Default::default();
+    pub static ref HISTORY: Arc<Mutex<History>> = Default::default();
+}
+
 // Start logger, should be done inside main
 pub fn init() {
     // Redirect all logs from libs using "Log"
@@ -56,21 +134,55 @@ pub fn init() {
         .with_thread_names(true)
         .with_filter(file_env_filter);
 
+    // Configure the server log
+    let server_env_filter = if cli::manager::is_tracing() {
+        EnvFilter::new(LevelFilter::TRACE.to_string())
+    } else {
+        EnvFilter::new(LevelFilter::DEBUG.to_string())
+    };
+    let (tx, mut rx) = tokio::sync::broadcast::channel(100);
+    let server_layer = fmt::Layer::new()
+        .with_writer(BroadcastMakeWriter { sender: tx.clone() })
+        .with_ansi(false)
+        .with_file(true)
+        .with_line_number(true)
+        .with_span_events(fmt::format::FmtSpan::NONE)
+        .with_target(false)
+        .with_thread_ids(true)
+        .with_thread_names(true)
+        .with_filter(server_env_filter);
+
+    let history = HISTORY.clone();
+    MANAGER.lock().unwrap().process = Some(tokio::spawn(async move {
+        loop {
+            match rx.recv().await {
+                Ok(message) => {
+                    history.lock().unwrap().push(message);
+                }
+                Err(error) => {
+                    tokio::time::sleep(tokio::time::Duration::from_millis(10)).await;
+                }
+            }
+        }
+    }));
+
     // Configure the default subscriber
     match cli::manager::is_tracy() {
         true => {
             let tracy_layer = tracing_tracy::TracyLayer::default();
             let subscriber = tracing_subscriber::registry()
                 .with(console_layer)
                 .with(file_layer)
+                .with(server_layer)
                 .with(tracy_layer);
             tracing::subscriber::set_global_default(subscriber)
                 .expect("Unable to set a global subscriber");
         }
         false => {
             let subscriber = tracing_subscriber::registry()
                 .with(console_layer)
-                .with(file_layer);
+                .with(file_layer)
+                .with(server_layer);
             tracing::subscriber::set_global_default(subscriber)
                 .expect("Unable to set a global subscriber");
         }

src/lib/server/manager.rs

@@ -76,6 +76,7 @@ pub async fn run(server_address: &str) -> Result<(), std::io::Error> {
             )
             .route("/xml", web::get().to(pages::xml))
             .route("/sdp", web::get().to(pages::sdp))
+            .route("/log", web::get().to(pages::log))
             .service(
                 web::scope("/thumbnail")
                     // Add a rate limitter to prevent flood

src/lib/server/pages.rs

@@ -2,8 +2,9 @@ use std::io::prelude::*;
 
 use actix_web::{
     http::header,
+    rt,
     web::{self, Json},
-    HttpRequest, HttpResponse,
+    Error, HttpRequest, HttpResponse,
 };
 use paperclip::actix::{api_v2_operation, Apiv2Schema, CreatedJson};
 use serde::{Deserialize, Serialize};
@@ -465,3 +466,26 @@ pub async fn gst_info() -> HttpResponse {
             .body(format!("{error:#?}")),
     }
 }
+
+#[api_v2_operation]
+/// Provides a access point for the service log
+pub async fn log(req: HttpRequest, stream: web::Payload) -> Result<HttpResponse, Error> {
+    let (response, mut session, _stream) = actix_ws::handle(&req, stream)?;
+    rt::spawn(async move {
+        let (mut receiver, history) = crate::logger::manager::HISTORY.lock().unwrap().subscribe();
+
+        for message in history {
+            if session.text(message).await.is_err() {
+                return;
+            }
+        }
+
+        while let Ok(message) = receiver.recv().await {
+            if session.text(message).await.is_err() {
+                return;
+            }
+        }
+    });
+
+    Ok(response)
+}