work on process controller

Cargo.toml

@@ -6,13 +6,17 @@ authors = ["Jonathan Baecker jonbae77@gmail.com"]
 readme = "README.md"
 version = "0.9.6"
 edition = "2021"
+default-run = "ffplayout"
 
 [dependencies]
+anyhow = "1.0"
 chrono = { git = "https://github.com/sbrocket/chrono", branch = "parse-error-kind-public" }
 clap = { version = "3.1", features = ["derive"] }
 crossbeam-channel = "0.5"
 ffprobe = "0.3"
 file-rotate = { git = "https://github.com/Ploppz/file-rotate.git", branch = "timestamp-parse-fix" }
+horust = "0.1"
+itertools = "0.10"
 jsonrpc-http-server = "18.0"
 lettre = "0.10.0-rc.6"
 log = "0.4"
@@ -34,6 +38,10 @@ openssl = { version = "0.10", features = ["vendored"] }
 name = "ffplayout"
 path = "src/main.rs"
 
+[[bin]]
+name = "ffpc"
+path = "src/bin/ffpc.rs"
+
 [profile.release]
 opt-level = 3
 strip = true

assets/services/channel-01.toml

@@ -0,0 +1,17 @@
+command = "/usr/bin/ffplayout -o stream"
+start-delay = "0s"
+user = "${USER}"
+
+[restart]
+strategy = "always"
+backoff = "1s"
+attempts = 5
+
+[termination]
+signal = "TERM"
+wait = "0s"
+
+[environment]
+keep-env = true
+re-export = ["PATH"]
+additional = { FONTCONFIG_PATH = "/etc/fonts" }

src/bin/ffpc.rs

@@ -0,0 +1,81 @@
+use std::path::PathBuf;
+
+use anyhow::{Context, Result};
+use clap::Parser;
+use itertools::Itertools;
+
+use horust::horust::ExitStatus;
+use horust::horust::HorustConfig;
+use horust::Horust;
+
+#[derive(Parser, Debug)]
+#[clap(version,
+    name = "ffpc",
+    version = "0.1.0",
+    about = "ffplayout process control",
+    long_about = None)]
+pub struct Args {
+    #[clap(
+        short,
+        long,
+        help = "File path to ffpc.toml",
+        default_value = "/etc/ffplayout/ffpc.toml"
+    )]
+    config_path: PathBuf,
+
+    #[clap(
+        short,
+        long = "services-path",
+        help = "Play folder content",
+        default_value = "/etc/ffplayout/services"
+    )]
+    services_paths: Vec<PathBuf>,
+}
+
+fn main() -> Result<()> {
+    let opts = Args::parse();
+
+    let horust_cfg = HorustConfig {
+        unsuccessful_exit_finished_failed: true,
+    };
+
+    let config =
+        HorustConfig::load_and_merge(&horust_cfg, &opts.config_path).with_context(|| {
+            format!(
+                "Failed loading configuration: {}",
+                &opts.config_path.display()
+            )
+        })?;
+
+    println!(
+        "Loading services from {}",
+        display_directories(&opts.services_paths)
+    );
+
+    let mut horust = {
+        Horust::from_services_dirs(&opts.services_paths).with_context(|| {
+            format!(
+                "Failed loading services from {}",
+                display_directories(&opts.services_paths)
+            )
+        })?
+    };
+
+    if let ExitStatus::SomeServiceFailed = horust.run() {
+        if config.unsuccessful_exit_finished_failed {
+            println!("Some processes have failed.");
+            std::process::exit(101);
+        }
+    }
+    Ok(())
+}
+
+fn display_directories(dirs: &[PathBuf]) -> String {
+    match dirs.len() {
+        1 => format!("directory: {}", dirs.first().unwrap().display()),
+        _ => format!(
+            "directories:\n{}",
+            dirs.iter().map(|d| format!("* {}", d.display())).join("\n")
+        ),
+    }
+}

src/input/ingest.rs

@@ -40,7 +40,7 @@ pub fn ingest_server(
         server_cmd.join(" ")
     );
 
-    'ingest_iter: loop {
+    while !proc_control.is_terminated.load(Ordering::SeqCst) {
         let mut server_proc = match Command::new("ffmpeg")
             .args(server_cmd.clone())
             .stdout(Stdio::piped())
@@ -79,7 +79,7 @@ pub fn ingest_server(
                     error!("Ingest server write error: {e:?}");
 
                     proc_control.is_terminated.store(true, Ordering::SeqCst);
-                    break 'ingest_iter;
+                    break;
                 }
             } else {
                 break;
@@ -98,10 +98,6 @@ pub fn ingest_server(
         if let Err(e) = error_reader_thread.join() {
             error!("{e:?}");
         };
-
-        if proc_control.is_terminated.load(Ordering::SeqCst) {
-            break;
-        }
     }
 
     Ok(())

src/lib.rs

@@ -0,0 +1,11 @@
+extern crate log;
+extern crate simplelog;
+
+pub mod filter;
+pub mod input;
+pub mod macros;
+pub mod output;
+pub mod rpc;
+#[cfg(test)]
+mod tests;
+pub mod utils;

src/main.rs

@@ -1,6 +1,3 @@
-extern crate log;
-extern crate simplelog;
-
 use std::{
     fs::{self, File},
     path::PathBuf,
@@ -13,21 +10,14 @@ use serde::{Deserialize, Serialize};
 use serde_json::json;
 use simplelog::*;
 
-mod filter;
+use ffplayout_engine::{
-mod input;
+    output::{player, write_hls},
-mod macros;
+    rpc::json_rpc_server,
-mod output;
+    utils::{
-mod rpc;
-#[cfg(test)]
-mod tests;
-mod utils;
-
-use crate::output::{player, write_hls};
-use crate::utils::{
         generate_playlist, init_logging, send_mail, validate_ffmpeg, GlobalConfig, PlayerControl,
         PlayoutStatus, ProcessControl,
+    },
 };
-use rpc::json_rpc_server;
 
 #[derive(Serialize, Deserialize)]
 struct StatusData {

src/utils/config.rs

@@ -274,6 +274,12 @@ impl GlobalConfig {
     }
 }
 
+impl Default for GlobalConfig {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
 /// When add_loudnorm is False we use a different audio encoder,
 /// s302m has higher quality, but is experimental
 /// and works not well together with the loudnorm filter.

src/utils/controller.rs

@@ -60,6 +60,12 @@ impl ProcessControl {
     }
 }
 
+impl Default for ProcessControl {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
 impl ProcessControl {
     pub fn kill(&mut self, proc: ProcessUnit) -> Result<(), String> {
         match proc {
@@ -169,6 +175,12 @@ impl PlayerControl {
     }
 }
 
+impl Default for PlayerControl {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
 /// Global playout control, for move forward/backward clip, or resetting playlist/state.
 #[derive(Clone, Debug)]
 pub struct PlayoutStatus {
@@ -188,3 +200,9 @@ impl PlayoutStatus {
         }
     }
 }
+
+impl Default for PlayoutStatus {
+    fn default() -> Self {
+        Self::new()
+    }
+}