Use an Actor for the Stream watching

Cargo.lock

@@ -1,5 +1,30 @@
 # This file is automatically @generated by Cargo.
 # It is not intended for manual editing.
+[[package]]
+name = "actix"
+version = "0.10.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "1be241f88f3b1e7e9a3fbe3b5a8a0f6915b5a1d7ee0d9a248d3376d01068cc60"
+dependencies = [
+ "actix-rt",
+ "actix_derive",
+ "bitflags",
+ "bytes 0.5.6",
+ "crossbeam-channel 0.4.4",
+ "derive_more",
+ "futures-channel",
+ "futures-util",
+ "log",
+ "once_cell",
+ "parking_lot",
+ "pin-project 0.4.27",
+ "smallvec",
+ "tokio",
+ "tokio-util",
+ "trust-dns-proto",
+ "trust-dns-resolver",
+]
+
 [[package]]
 name = "actix-codec"
 version = "0.3.0"
@@ -313,6 +338,17 @@ dependencies = [
  "syn",
 ]
 
+[[package]]
+name = "actix_derive"
+version = "0.5.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "b95aceadaf327f18f0df5962fedc1bde2f870566a0b9f65c89508a3b1f79334c"
+dependencies = [
+ "proc-macro2",
+ "quote",
+ "syn",
+]
+
 [[package]]
 name = "addr2line"
 version = "0.14.0"
@@ -642,6 +678,16 @@ dependencies = [
  "cfg-if 1.0.0",
 ]
 
+[[package]]
+name = "crossbeam-channel"
+version = "0.4.4"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "b153fe7cbef478c567df0f972e02e6d736db11affe43dfc9c56a9374d1adfb87"
+dependencies = [
+ "crossbeam-utils 0.7.2",
+ "maybe-uninit",
+]
+
 [[package]]
 name = "crossbeam-channel"
 version = "0.5.0"
@@ -649,7 +695,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "dca26ee1f8d361640700bde38b2c37d8c22b3ce2d360e1fc1c74ea4b0aa7d775"
 dependencies = [
  "cfg-if 1.0.0",
- "crossbeam-utils",
+ "crossbeam-utils 0.8.0",
 ]
 
 [[package]]
@@ -660,7 +706,7 @@ checksum = "94af6efb46fef72616855b036a624cf27ba656ffc9be1b9a3c931cfc7749a9a9"
 dependencies = [
  "cfg-if 1.0.0",
  "crossbeam-epoch",
- "crossbeam-utils",
+ "crossbeam-utils 0.8.0",
 ]
 
 [[package]]
@@ -671,12 +717,23 @@ checksum = "ec0f606a85340376eef0d6d8fec399e6d4a544d648386c6645eb6d0653b27d9f"
 dependencies = [
  "cfg-if 1.0.0",
  "const_fn",
- "crossbeam-utils",
+ "crossbeam-utils 0.8.0",
  "lazy_static",
  "memoffset",
  "scopeguard",
 ]
 
+[[package]]
+name = "crossbeam-utils"
+version = "0.7.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "c3c7c73a2d1e9fc0886a08b93e98eb643461230d5f1925e4036204d5f2e261a8"
+dependencies = [
+ "autocfg",
+ "cfg-if 0.1.10",
+ "lazy_static",
+]
+
 [[package]]
 name = "crossbeam-utils"
 version = "0.8.0"
@@ -1146,6 +1203,7 @@ dependencies = [
 name = "image-api"
 version = "0.1.0"
 dependencies = [
+ "actix",
  "actix-cors",
  "actix-files",
  "actix-multipart",
@@ -1357,6 +1415,12 @@ version = "0.1.8"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "7ffc5c5338469d4d3ea17d269fa8ea3512ad247247c30bd2df69e68309ed0a08"
 
+[[package]]
+name = "maybe-uninit"
+version = "2.0.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "60302e4db3a61da70c0cb7991976248362f30319e88850c487b9b95bbf059e00"
+
 [[package]]
 name = "memchr"
 version = "2.3.4"
@@ -1824,9 +1888,9 @@ version = "1.9.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "9ab346ac5921dc62ffa9f89b7a773907511cdfa5490c572ae9be1be33e8afa4a"
 dependencies = [
- "crossbeam-channel",
+ "crossbeam-channel 0.5.0",
  "crossbeam-deque",
- "crossbeam-utils",
+ "crossbeam-utils 0.8.0",
  "lazy_static",
  "num_cpus",
 ]
@@ -2256,6 +2320,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "a6d7ad61edd59bfcc7e80dababf0f4aed2e6d5e0ba1659356ae889752dfc12ff"
 dependencies = [
  "bytes 0.5.6",
+ "fnv",
  "futures-core",
  "iovec",
  "lazy_static",
@@ -2277,6 +2342,7 @@ checksum = "be8242891f2b6cbef26a2d7e8605133c2c554cd35b3e4948ea892d6d68436499"
 dependencies = [
  "bytes 0.5.6",
  "futures-core",
+ "futures-io",
  "futures-sink",
  "log",
  "pin-project-lite 0.1.11",

Cargo.toml

@@ -7,11 +7,12 @@ edition = "2018"
 # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
 
 [dependencies]
+actix = "0.10"
 actix-web = "3"
 actix-rt = "1"
 actix-files = "0.4"
 actix-multipart = "0.3.0"
-actix-cors="0.5"
+actix-cors = "0.5"
 futures = "0.3.5"
 jsonwebtoken = "7.2.0"
 serde = "1"

src/main.rs

@@ -5,23 +5,24 @@ extern crate rayon;
 use std::fs::File;
 use std::io::prelude::*;
 use std::path::{Path, PathBuf};
+use std::sync::{Arc};
 use std::sync::mpsc::channel;
-use std::sync::{Arc, Mutex};
 
+use actix::{Actor, Addr};
 use actix_files::NamedFile;
 use actix_multipart as mp;
+use actix_web::{App, get, HttpServer, post, Responder, web};
 use actix_web::web::{HttpRequest, HttpResponse, Json};
-use actix_web::{get, post, web, App, HttpServer, Responder};
 use chrono::{Duration, Utc};
 use futures::stream::StreamExt;
 use jsonwebtoken::{encode, EncodingKey, Header};
-use notify::{watcher, DebouncedEvent, RecursiveMode, Watcher};
+use notify::{DebouncedEvent, RecursiveMode, watcher, Watcher};
 use rayon::prelude::*;
 use serde::Serialize;
 
 use data::{AddFavoriteRequest, LoginRequest, ThumbnailRequest};
 
-use crate::data::{secret_key, Claims, CreateAccountRequest, Token};
+use crate::data::{Claims, CreateAccountRequest, secret_key, Token};
 use crate::database::{add_favorite, create_user, get_favorites, get_user, user_exists};
 use crate::files::{is_valid_path, list_files};
 use crate::video::*;
@@ -185,12 +186,8 @@ async fn generate_video(
         let filename = name.to_str().expect("Filename should convert to string");
         let playlist = format!("tmp/{}.m3u8", filename);
         if let Some(path) = is_valid_path(&body.path) {
-            if let Ok(mut stream) = data.stream_manager.lock() {
             if let Ok(child) = create_playlist(&path.to_str().unwrap(), &playlist) {
-                    stream.track_stream(&playlist, child)
-                }
-            } else {
-                let _ = create_playlist(&path.to_str().unwrap(), &playlist);
+                data.stream_manager.do_send(ProcessMessage(playlist.clone(), child));
             }
         } else {
             return HttpResponse::BadRequest().finish();
@@ -259,7 +256,7 @@ async fn post_add_favorite(claims: Claims, body: web::Json<AddFavoriteRequest>)
     }
 }
 
-async fn create_thumbnails() {
+fn create_thumbnails() {
     let thumbs = &dotenv::var("THUMBNAILS").expect("THUMBNAILS not defined");
     let thumbnail_directory: &Path = Path::new(thumbs);
 
@@ -314,13 +311,10 @@ async fn create_thumbnails() {
     println!("Finished");
 }
 
-#[actix_rt::main]
-async fn main() -> std::io::Result<()> {
-    create_thumbnails().await;
+fn main() -> std::io::Result<()> {
+    create_thumbnails();
 
-    let stream_manager = Arc::new(Mutex::new(VideoStreamManager::new()));
-    let stream_manager_copy = stream_manager.clone();
-    tokio::spawn(async move {
+    std::thread::spawn(|| {
         let (wtx, wrx) = channel();
         let mut watcher = watcher(wtx, std::time::Duration::from_secs(10)).unwrap();
         watcher
@@ -328,24 +322,24 @@ async fn main() -> std::io::Result<()> {
             .unwrap();
 
         loop {
-            if let Ok(mut manager) = stream_manager_copy.lock() {
-                manager.check_for_finished_streams();
-            }
-
             let ev = wrx.recv_timeout(std::time::Duration::from_secs(10));
             if let Ok(event) = ev {
                 match event {
-                    DebouncedEvent::Create(_) => create_thumbnails().await,
-                    DebouncedEvent::Rename(_, _) => create_thumbnails().await,
+                    DebouncedEvent::Create(_) => create_thumbnails(),
+                    DebouncedEvent::Rename(_, _) => create_thumbnails(),
                     _ => continue,
                 };
             }
         }
     });
 
+    let system = actix::System::new("Fileserver");
+    let act = StreamActor {}.start();
+
     let app_data = web::Data::new(AppState {
-        stream_manager: stream_manager.clone(),
+        stream_manager: Arc::new(act),
     });
+
     HttpServer::new(move || {
         App::new()
             .service(login)
@@ -361,10 +355,11 @@ async fn main() -> std::io::Result<()> {
     })
         .bind(dotenv::var("BIND_URL").unwrap())?
         .bind("localhost:8088")?
-    .run()
-    .await
+        .run();
+
+    system.run()
 }
 
 struct AppState {
-    stream_manager: Arc<Mutex<VideoStreamManager>>,
+    stream_manager: Arc<Addr<StreamActor>>,
 }

src/video.rs

@@ -1,36 +1,40 @@
-use std::collections::HashMap;
 use std::io::Result;
 use std::path::Path;
-use std::process::{Child, Command, Stdio};
+use std::process::{Child, Command, ExitStatus, Stdio};
+
+use actix::prelude::*;
 
 // ffmpeg -i test.mp4 -c:v h264 -flags +cgop -g 30 -hls_time 3 out.m3u8
 // ffmpeg -i "filename.mp4" -preset veryfast -c:v libx264 -f hls -hls_list_size 100 -hls_time 2 -crf 24 -vf scale=1080:-2,setsar=1:1 attempt/vid_out.m3u8
 
-pub(crate) struct VideoStreamManager {
-    streams: HashMap<String, Child>,
+pub struct StreamActor;
+
+impl Actor for StreamActor {
+    type Context = Context<Self>;
 }
 
-impl VideoStreamManager {
-    pub(crate) fn new() -> VideoStreamManager {
-        VideoStreamManager {
-            streams: HashMap::new(),
-        }
-    }
+pub struct ProcessMessage(pub String, pub Child);
 
-    pub fn track_stream(&mut self, playlist_path: &dyn ToString, child_process: Child) {
-        println!("Tracking process for: {:?}", playlist_path.to_string());
-        self.streams
-            .insert(playlist_path.to_string(), child_process);
-    }
+impl Message for ProcessMessage {
+    type Result = Result<ExitStatus>;
+}
 
-    pub fn check_for_finished_streams(&mut self) {
-        self.streams.retain(|playlist_path, process| {
-            let is_done = process.try_wait().map_or(false, |status| status.is_some());
-            if is_done {
-                println!("Removing process: {:?}", playlist_path)
-            }
-            !is_done
-        });
+impl Handler<ProcessMessage> for StreamActor {
+    type Result = Result<ExitStatus>;
+
+    fn handle(&mut self, msg: ProcessMessage, _ctx: &mut Self::Context) -> Self::Result {
+        println!("Message received");
+        let mut process = msg.1;
+        let result = process.wait();
+
+        println!(
+            "Finished waiting for: {:?}. Code: {:?}",
+            msg.0,
+            result
+                .as_ref()
+                .map_or(-1, |status| status.code().unwrap_or(-1))
+        );
+        result
     }
 }