Add test to check webrtc task leak

Cargo.toml

@@ -49,6 +49,7 @@ v4l = "0.14.0"
 directories = "5.0.1"
 pnet = { version = "0.34.0", features = ["std"] }
 semver = "1.0"
+thirtyfour = "0.31"
 tracing = { version = "0.1.37", features = ["log", "async-await"] }
 tracing-subscriber = { version = "0.3.17", features = ["env-filter"] }
 tracing-log = "0.1.3"

src/cli/manager.rs

@@ -97,6 +97,14 @@ pub fn enable_thread_counter() -> bool {
         .is_present("enable-thread-counter");
 }
 
+pub fn enable_webrtc_task_test() -> Option<u32> {
+    return MANAGER
+        .as_ref()
+        .clap_matches
+        .value_of("enable-webrtc-task-test")
+        .and_then(|value| value.parse::<u32>().ok());
+}
+
 // Return the command line used to start this application
 pub fn command_line_string() -> String {
     std::env::args().collect::<Vec<String>>().join(" ")
@@ -230,6 +238,14 @@ fn get_clap_matches<'a>() -> clap::ArgMatches<'a> {
                 .long("enable-thread-counter")
                 .help("Enable a thread that prints the number of children processes.")
                 .takes_value(false),
+        )
+        .arg(
+            clap::Arg::with_name("enable-webrtc-task-test")
+                .long("enable-webrtc-task-test")
+                .help("Enable webrtc thread test with limit of child tasks (can use port for webdriver as parameter).")
+                .value_name("PORT")
+                .default_value("9515")
+                .empty_values(true)
         );
 
     matches.get_matches()

src/helper/develop.rs

@@ -0,0 +1,56 @@
+use crate::cli;
+use crate::helper;
+use anyhow::Result;
+use core::time::Duration;
+use std::thread;
+use thirtyfour::prelude::*;
+use tokio::runtime::Runtime;
+use tracing::*;
+
+async fn task(mut counter: i32) -> Result<()> {
+    info!("Started webrtc test..");
+
+    let mut caps = DesiredCapabilities::chrome();
+    let _ = caps.set_headless();
+
+    let port = cli::manager::enable_webrtc_task_test().unwrap();
+    let driver = WebDriver::new(&format!("http://localhost:{}", port), caps)
+        .await
+        .expect("Failed to create web driver.");
+
+    driver
+        .goto("http://0.0.0.0:6020/webrtc/index.html")
+        .await
+        .expect("Failed to connect to local webrtc page.");
+
+    loop {
+        for button in ["add-consumer", "add-session", "remove-all-consumers"] {
+            thread::sleep(Duration::from_secs(3));
+            driver.find(By::Id(button)).await?.click().await?;
+        }
+
+        counter += 1;
+
+        info!("Restarted webrtc {} times", counter);
+        if helper::threads::process_task_counter() > 100 {
+            error!("Thead leak detected!");
+            std::process::exit(-1);
+        }
+    }
+}
+
+pub fn start_check_tasks_on_webrtc_reconnects() {
+    let counter = 0;
+    thread::spawn(move || {
+        let rt = Runtime::new().unwrap();
+        rt.block_on(async move {
+            loop {
+                if let Err(error) = task(counter).await {
+                    error!("WebRTC Checker Task failed: {error:#?}");
+                }
+            }
+        });
+        error!("Webrtc test failed internally.");
+        std::process::exit(-1);
+    });
+}

src/helper/mod.rs

@@ -1,3 +1,4 @@
 #[macro_use]
+pub mod develop;
 pub mod macros;
 pub mod threads;

src/helper/threads.rs

@@ -1,17 +1,20 @@
+use cached::proc_macro::cached;
 use std::thread;
 use std::time::Duration;
 use sysinfo::{System, SystemExt};
 use tracing::*;
 
-pub fn start_thread_counter_thread() {
+#[cached(time = 1)]
+pub fn process_task_counter() -> usize {
     let mut system = System::new_all();
     let pid = sysinfo::get_current_pid().expect("Failed to get current PID.");
+    system.refresh_process(pid);
+    system.process(pid).unwrap().tasks.len()
+}
+
+pub fn start_thread_counter_thread() {
     thread::spawn(move || loop {
-        system.refresh_process(pid);
-        info!(
-            "Number of child processes: {}",
-            system.process(pid).unwrap().tasks.len()
-        );
+        info!("Number of child processes: {}", process_task_counter());
         thread::sleep(Duration::from_secs(1));
     });
 }

src/main.rs

@@ -38,6 +38,10 @@ async fn main() -> Result<(), std::io::Error> {
         helper::threads::start_thread_counter_thread();
     }
 
+    if cli::manager::enable_webrtc_task_test().is_some() {
+        helper::develop::start_check_tasks_on_webrtc_reconnects();
+    }
+
     stream::webrtc::signalling_server::SignallingServer::default();
 
     if let Err(error) = stream::manager::start_default() {

src/server/manager.rs

@@ -55,6 +55,7 @@ pub async fn run(server_address: &str) -> Result<(), std::io::Error> {
                 r"/{filename:.*(\.html|\.js|\.css)}",
                 web::get().to(pages::root),
             )
+            .route("/info", web::get().to(pages::info))
             .route("/delete_stream", web::delete().to(pages::remove_stream))
             .route("/reset_settings", web::post().to(pages::reset_settings))
             .route("/streams", web::get().to(pages::streams))

src/server/pages.rs

@@ -1,3 +1,4 @@
+use crate::helper;
 use crate::settings;
 use crate::stream::{manager as stream_manager, types::StreamInformation};
 use crate::video::{
@@ -12,7 +13,8 @@ use actix_web::{
     web::{self, Json},
     HttpRequest, HttpResponse,
 };
-use paperclip::actix::{api_v2_operation, Apiv2Schema};
+
+use paperclip::actix::{api_v2_operation, Apiv2Schema, CreatedJson};
 use serde::{Deserialize, Serialize};
 use tracing::*;
 use validator::Validate;
@@ -77,6 +79,41 @@ pub struct ThumbnailFileRequest {
     target_height: Option<u16>,
 }
 
+#[derive(Apiv2Schema, Serialize, Debug)]
+pub struct Development {
+    number_of_tasks: usize,
+}
+
+#[derive(Apiv2Schema, Serialize, Debug)]
+pub struct Info {
+    /// Name of the program
+    name: String,
+    /// Version/tag
+    version: String,
+    /// Git SHA
+    sha: String,
+    build_date: String,
+    /// Authors name
+    authors: String,
+    /// Unstable field for custom development
+    development: Development,
+}
+
+impl Info {
+    pub fn new() -> Self {
+        Self {
+            name: env!("CARGO_PKG_NAME").into(),
+            version: env!("VERGEN_GIT_SEMVER").into(),
+            sha: env!("VERGEN_GIT_SHA_SHORT").into(),
+            build_date: env!("VERGEN_BUILD_DATE").into(),
+            authors: env!("CARGO_PKG_AUTHORS").into(),
+            development: Development {
+                number_of_tasks: helper::threads::process_task_counter(),
+            },
+        }
+    }
+}
+
 use std::{ffi::OsStr, path::Path};
 
 use include_dir::{include_dir, Dir};
@@ -136,6 +173,13 @@ pub fn root(req: HttpRequest) -> HttpResponse {
     HttpResponse::Ok().content_type(mime).body(content)
 }
 
+#[api_v2_operation]
+/// Provide information about the running service
+/// There is no stable API guarantee for the development field
+pub async fn info() -> CreatedJson<Info> {
+    CreatedJson(Info::new())
+}
+
 //TODO: change endpoint name to sources
 #[api_v2_operation]
 /// Provides list of all video sources, with controls and formats

src/stream/webrtc/frontend/src/App.vue

@@ -41,10 +41,18 @@ const manager = reactive(new Manager(ip, 6021, rtc_configuration));
       <p>Consumers: {{ manager.consumers.size }}</p>
     </div>
     <div>
-      <button type="button" v-on:click="manager.addConsumer()">
+      <button
+        id="add-consumer"
+        type="button"
+        v-on:click="manager.addConsumer()"
+      >
         Add consumer
       </button>
-      <button type="button" v-on:click="manager.removeAllConsumers()">
+      <button
+        id="remove-all-consumers"
+        type="button"
+        v-on:click="manager.removeAllConsumers()"
+      >
         Remove all consumers
       </button>
     </div>
@@ -96,6 +104,7 @@ const manager = reactive(new Manager(ip, 6021, rtc_configuration));
             {{ stream.encode }}
           </p>
           <button
+            id="add-session"
             type="button"
             v-on:click="manager.addSession(consumer.id, stream.id)"
           >