Merge pull request 'feature/video-stream-manager' (#4) from feature/video-stream-manager into master

Reviewed-on: #4
2021-02-13 18:18:12 +00:00
parent e4c23c0fe5 11d1e9600a
commit 1ad7abb69c
10 changed files with 213 additions and 50 deletions
--- a/.gitignore
+++ b/.gitignore
@@ -2,3 +2,12 @@
 database/target
 *.db
 .env
+
+# Default ignored files
+.idea/shelf/
+.idea/workspace.xml
+# Datasource local storage ignored files
+.idea/dataSources*
+.idea/dataSources.local.xml
+# Editor-based HTTP Client requests
+.idea/httpRequests/
--- a/.idea/image-api.iml
+++ b/.idea/image-api.iml
@@ -0,0 +1,11 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<module type="CPP_MODULE" version="4">
+  <component name="NewModuleRootManager">
+    <content url="file://$MODULE_DIR$">
+      <sourceFolder url="file://$MODULE_DIR$/src" isTestSource="false" />
+      <excludeFolder url="file://$MODULE_DIR$/target" />
+    </content>
+    <orderEntry type="inheritedJdk" />
+    <orderEntry type="sourceFolder" forTests="false" />
+  </component>
+</module>
--- a/.idea/misc.xml
+++ b/.idea/misc.xml
@@ -0,0 +1,6 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project version="4">
+  <component name="JavaScriptSettings">
+    <option name="languageLevel" value="ES6" />
+  </component>
+</project>
--- a/.idea/modules.xml
+++ b/.idea/modules.xml
@@ -0,0 +1,8 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project version="4">
+  <component name="ProjectModuleManager">
+    <modules>
+      <module fileurl="file://$PROJECT_DIR$/.idea/image-api.iml" filepath="$PROJECT_DIR$/.idea/image-api.iml" />
+    </modules>
+  </component>
+</project>
--- a/.idea/vcs.xml
+++ b/.idea/vcs.xml
@@ -0,0 +1,6 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project version="4">
+  <component name="VcsDirectoryMappings">
+    <mapping directory="$PROJECT_DIR$" vcs="Git" />
+  </component>
+</project>
--- a/Cargo.lock
+++ b/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",
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -7,6 +7,7 @@ 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"
--- a/src/data/mod.rs
+++ b/src/data/mod.rs
@@ -1,9 +1,9 @@
 use std::str::FromStr;

-use actix_web::{dev, Error, FromRequest, http::header, HttpRequest};
 use actix_web::error::ErrorUnauthorized;
+use actix_web::{dev, http::header, Error, FromRequest, HttpRequest};
 use futures::future::{err, ok, Ready};
-use jsonwebtoken::{Algorithm, decode, DecodingKey, Validation};
+use jsonwebtoken::{decode, Algorithm, DecodingKey, Validation};
 use serde::{Deserialize, Serialize};

 #[derive(Serialize)]
--- a/src/main.rs
+++ b/src/main.rs
@@ -2,23 +2,27 @@
 extern crate diesel;
 extern crate rayon;

-use actix_files::NamedFile;
-use actix_multipart as mp;
-use actix_web::web::{HttpRequest, HttpResponse, Json};
-use actix_web::{get, post, web, App, HttpServer, Responder};
-use chrono::{Duration, Utc};
-use data::{AddFavoriteRequest, LoginRequest, ThumbnailRequest};
-use futures::stream::StreamExt;
-use jsonwebtoken::{encode, EncodingKey, Header};
-use notify::{watcher, DebouncedEvent, RecursiveMode, Watcher};
-use rayon::prelude::*;
-use serde::Serialize;
 use std::fs::File;
 use std::io::prelude::*;
 use std::path::{Path, PathBuf};
+use std::sync::{Arc};
 use std::sync::mpsc::channel;

-use crate::data::{secret_key, Claims, CreateAccountRequest, Token};
+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 chrono::{Duration, Utc};
+use futures::stream::StreamExt;
+use jsonwebtoken::{encode, EncodingKey, Header};
+use notify::{DebouncedEvent, RecursiveMode, watcher, Watcher};
+use rayon::prelude::*;
+use serde::Serialize;
+
+use data::{AddFavoriteRequest, LoginRequest, ThumbnailRequest};
+
+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::*;
@@ -171,14 +175,20 @@ async fn upload_image(_: Claims, mut payload: mp::Multipart) -> impl Responder {
 }

 #[post("/video/generate")]
-async fn generate_video(_claims: Claims, body: web::Json<ThumbnailRequest>) -> impl Responder {
+async fn generate_video(
+    _claims: Claims,
+    data: web::Data<AppState>,
+    body: web::Json<ThumbnailRequest>,
+) -> impl Responder {
    let filename = PathBuf::from(&body.path);

    if let Some(name) = filename.file_stem() {
        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) {
-            create_playlist(&path.to_str().unwrap(), &playlist);
+            if let Ok(child) = create_playlist(&path.to_str().unwrap(), &playlist) {
+                data.stream_manager.do_send(ProcessMessage(playlist.clone(), child));
+            }
        } else {
            return HttpResponse::BadRequest().finish();
        }
@@ -246,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);

@@ -301,11 +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();

-    tokio::spawn(async {
+    std::thread::spawn(|| {
        let (wtx, wrx) = channel();
        let mut watcher = watcher(wtx, std::time::Duration::from_secs(10)).unwrap();
        watcher
@@ -313,24 +322,25 @@ async fn main() -> std::io::Result<()> {
            .unwrap();

        loop {
-            let ev = wrx.recv_timeout(std::time::Duration::from_secs(5));
-            match ev {
-                Ok(event) => {
+            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,
                };
            }
-                Err(e) => {
-                    println!("Event: {:?}", e);
-                    // break;
-                }
-            }
        }
    });

-    HttpServer::new(|| {
+    let system = actix::System::new("Fileserver");
+    let act = StreamActor {}.start();
+
+    let app_data = web::Data::new(AppState {
+        stream_manager: Arc::new(act),
+    });
+
+    HttpServer::new(move || {
        App::new()
            .service(login)
            .service(list_photos)
@@ -341,9 +351,15 @@ async fn main() -> std::io::Result<()> {
            .service(get_video_part)
            .service(favorites)
            .service(post_add_favorite)
+            .app_data(app_data.clone())
    })
        .bind(dotenv::var("BIND_URL").unwrap())?
        .bind("localhost:8088")?
-    .run()
-    .await
+        .run();
+
+    system.run()
+}
+
+struct AppState {
+    stream_manager: Arc<Addr<StreamActor>>,
 }
--- a/src/video.rs
+++ b/src/video.rs
@@ -1,13 +1,47 @@
+use std::io::Result;
 use std::path::Path;
-use std::process::Command;
+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 fn create_playlist(video_path: &str, playlist_file: &str) {
+pub struct StreamActor;
+
+impl Actor for StreamActor {
+    type Context = Context<Self>;
+}
+
+pub struct ProcessMessage(pub String, pub Child);
+
+impl Message for ProcessMessage {
+    type Result = Result<ExitStatus>;
+}
+
+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
+    }
+}
+
+pub fn create_playlist(video_path: &str, playlist_file: &str) -> Result<Child> {
    if Path::new(playlist_file).exists() {
        println!("Playlist already exists: {}", playlist_file);
-        return;
+        return Err(std::io::Error::from(std::io::ErrorKind::AlreadyExists));
    }

    let result = Command::new("ffmpeg")
@@ -26,16 +60,22 @@ pub fn create_playlist(video_path: &str, playlist_file: &str) {
        .arg("-vf")
        .arg("scale=1080:-2,setsar=1:1")
        .arg(playlist_file)
+        .stdout(Stdio::null())
+        .stderr(Stdio::null())
        .spawn();

    let start_time = std::time::Instant::now();
    loop {
        std::thread::sleep(std::time::Duration::from_secs(1));

-        if Path::new(playlist_file).exists() || std::time::Instant::now() - start_time > std::time::Duration::from_secs(5) {
+        if Path::new(playlist_file).exists()
+            || std::time::Instant::now() - start_time > std::time::Duration::from_secs(5)
+        {
            break;
        }
    }
+
+    result
 }

 pub fn generate_video_thumbnail(path: &Path, destination: &Path) {