handle ffmpeg thread shutdown

Author: Fredolx

src-tauri/src/lib.rs

@@ -405,8 +405,12 @@ async fn on_start_check_epg(
 }
 
 #[tauri::command]
-async fn start_restream(state: State<'_, Mutex<AppState>>, channel: Channel) -> Result<(), String> {
-    crate::restream::start_restream(state, channel)
+async fn start_restream(
+    state: State<'_, Mutex<AppState>>,
+    app: AppHandle,
+    channel: Channel,
+) -> Result<(), String> {
+    crate::restream::start_restream(state, app, channel)
         .await
         .map_err(map_err_frontend)
 }

src-tauri/src/restream.rs

@@ -1,10 +1,11 @@
 use std::{
     path::{Path, PathBuf},
     process::{Child, Command, Stdio},
+    time::Duration,
 };
 
 use anyhow::{Context, Result};
-use tauri::State;
+use tauri::{AppHandle, Emitter, State};
 use tokio::{
     fs,
     sync::{
@@ -14,7 +15,6 @@ use tokio::{
 };
 
 use crate::{
-    log::log,
     mpv,
     settings::get_settings,
     sql,
@@ -83,31 +83,39 @@ async fn start_web_server(
     return Ok((tx, handle));
 }
 
-pub async fn start_restream(state: State<'_, Mutex<AppState>>, channel: Channel) -> Result<()> {
-    let mut state = state.lock().await;
+pub async fn start_restream(
+    state: State<'_, Mutex<AppState>>,
+    app: AppHandle,
+    channel: Channel,
+) -> Result<()> {
+    let stop = state.lock().await.restream_stop_signal.clone();
+    stop.store(false, std::sync::atomic::Ordering::Relaxed);
     let restream_dir = get_restream_folder()?;
     delete_old_segments(&restream_dir).await?;
-    state.ffmpeg_child = Some(start_ffmpeg_listening(channel, restream_dir.clone())?);
-    (state.web_server_tx, state.web_server_handle) = start_web_server(restream_dir)
-        .await
-        .map(|(tx, handle)| (Some(tx), Some(handle)))?;
+    let mut ffmpeg_child = start_ffmpeg_listening(channel, restream_dir.clone())?;
+    let (web_server_tx, web_server_handle) = start_web_server(restream_dir).await?;
+    let _ = app.emit("restream_started", true);
+    while !stop.load(std::sync::atomic::Ordering::Relaxed)
+        && ffmpeg_child
+            .try_wait()
+            .map(|option| option.is_none())
+            .unwrap_or(true)
+        && !web_server_handle.is_finished()
+    {
+        tokio::time::sleep(Duration::from_millis(500)).await
+    }
+    let _ = ffmpeg_child.kill();
+    let _ = web_server_tx.send(true);
+    let _ = ffmpeg_child.wait();
+    let _ = web_server_handle.await;
     Ok(())
 }
 
 pub async fn stop_restream(state: State<'_, Mutex<AppState>>) -> Result<()> {
-    let mut state = state.lock().await;
-    let mut ffmpeg_child = state.ffmpeg_child.take().context("no ffmpeg child")?;
-    let web_server_tx = state.web_server_tx.take().context("no web server tx")?;
-    let web_server_handle = state
-        .web_server_handle
-        .take()
-        .context("no web server handle")?;
-    let _ = ffmpeg_child.kill();
-    let _ = web_server_tx.send(true);
-    let _ = ffmpeg_child.wait();
-    let _ = web_server_handle
-        .await
-        .unwrap_or_else(|e| log(format!("{:?}", e)));
+    let state = state.lock().await;
+    state
+        .restream_stop_signal
+        .store(true, std::sync::atomic::Ordering::Relaxed);
     Ok(())
 }
 

src-tauri/src/types.rs

@@ -141,9 +141,7 @@ pub struct EPGNotify {
 pub struct AppState {
     pub notify_stop: Arc<AtomicBool>,
     pub thread_handle: Option<JoinHandle<Result<(), anyhow::Error>>>,
-    pub ffmpeg_child: Option<std::process::Child>,
-    pub web_server_tx: Option<tokio::sync::oneshot::Sender<bool>>,
-    pub web_server_handle: Option<tokio::task::JoinHandle<()>>,
+    pub restream_stop_signal: Arc<AtomicBool>,
 }
 
 #[derive(Clone, PartialEq, Debug, Deserialize, Serialize)]

src/app/restream-modal/restream-modal.component.ts

@@ -1,43 +1,48 @@
-import { Component, OnInit } from "@angular/core";
+import { Component, NgZone, OnDestroy, OnInit } from "@angular/core";
 import { Channel } from "../models/channel";
 import { invoke } from "@tauri-apps/api/core";
 import { ErrorService } from "../error.service";
 import { NetworkInfo } from "../models/networkInfo";
 import { NgbActiveModal } from "@ng-bootstrap/ng-bootstrap";
+import { UnlistenFn, listen } from "@tauri-apps/api/event";
 
 @Component({
   selector: "app-restream-modal",
   templateUrl: "./restream-modal.component.html",
   styleUrl: "./restream-modal.component.css",
 })
-export class RestreamModalComponent implements OnInit {
+export class RestreamModalComponent implements OnInit, OnDestroy {
   channel?: Channel;
   loading = false;
   watching = false;
   started = false;
-  address = "http://192.168.2.10/stream.m3u8";
-  wanAddress = "http://10.145.22.12/stream.m3u8";
   networkInfo?: NetworkInfo;
   selectedIP?: string;
-
+  toUnlisten: UnlistenFn[] = [];
   constructor(
     private error: ErrorService,
     public activeModal: NgbActiveModal,
+    private ngZone: NgZone,
   ) {}
 
   ngOnInit(): void {
     invoke("get_network_info").then((network) => {
       this.networkInfo = network as NetworkInfo;
       this.selectedIP = this.networkInfo.local_ips[0];
     });
+    listen<boolean>("restream_started", () => {
+      this.ngZone.run(() => {
+        this.started = true;
+        this.loading = false;
+      });
+    }).then((unlisten) => this.toUnlisten.push(unlisten));
   }
 
   async start() {
     this.loading = true;
     try {
       await invoke("start_restream", { channel: this.channel });
-      this.error.success("Successfully started service");
-      this.started = true;
+      this.started = false;
     } catch (e) {
       this.error.handleError(e);
     }
@@ -48,8 +53,6 @@ export class RestreamModalComponent implements OnInit {
     this.loading = true;
     try {
       await invoke("stop_restream");
-      this.error.success("Successfully stopped service");
-      this.started = false;
     } catch (e) {
       this.error.handleError(e);
     }
@@ -76,4 +79,8 @@ export class RestreamModalComponent implements OnInit {
       this.error.handleError(e);
     }
   }
+
+  ngOnDestroy(): void {
+    this.toUnlisten.forEach((x) => x());
+  }
 }