djsundog/ffplayout
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

switch back from global config to local config

Browse Source
This commit is contained in:
jb-alvarado 2022-05-12 21:17:26 +02:00
parent 2a17bd4690
commit cb0850a624
23 changed files with 182 additions and 231 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

19
Cargo.lock generated
View File

@ -94,9 +94,9 @@ dependencies = [
[[package]]
name = "clap"
version = "3.1.17"
version = "3.1.18"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "47582c09be7c8b32c0ab3a6181825ababb713fde6fff20fc573a3870dd45c6a0"
checksum = "d2dbdf4bdacb33466e854ce889eee8dfd5729abf7ccd7664d0a2d60cd384440b"
dependencies = [
"atty",
"bitflags",
@ -111,9 +111,9 @@ dependencies = [
[[package]]
name = "clap_derive"
version = "3.1.7"
version = "3.1.18"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "a3aab4734e083b809aaf5794e14e756d1c798d2c69c7f7de7a09a2f5214993c1"
checksum = "25320346e922cffe59c0bbc5410c8d8784509efb321488971081313cb1e1a33c"
dependencies = [
"heck",
"proc-macro-error",
@ -207,7 +207,6 @@ dependencies = [
"lettre",
"log",
"notify",
"once_cell",
"openssl",
"rand",
"regex",
@ -1262,12 +1261,12 @@ checksum = "f2dd574626839106c320a323308629dcb1acfc96e32a8cba364ddc61ac23ee83"
[[package]]
name = "socket2"
version = "0.4.5"
version = "0.4.4"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "ca642ba17f8b2995138b1d7711829c92e98c0a25ea019de790f4f09279c4e296"
checksum = "66d72b759436ae32898a2af0a14218dbf55efde3feeb170eb623637db85ee1e0"
dependencies = [
"libc",
"windows-sys",
"winapi 0.3.9",
]
[[package]]
@ -1278,9 +1277,9 @@ checksum = "73473c0e59e6d5812c5dfe2a064a6444949f089e20eec9a2e5506596494e4623"
[[package]]
name = "syn"
version = "1.0.92"
version = "1.0.93"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "7ff7c592601f11445996a06f8ad0c27f094a58857c2f89e97974ab9235b92c52"
checksum = "04066589568b72ec65f42d65a1a52436e954b168773148893c020269563decf2"
dependencies = [
"proc-macro2",
"quote",

1
Cargo.toml
View File

@ -17,7 +17,6 @@ jsonrpc-http-server = "18.0"
lettre = "0.10.0-rc.6"
log = "0.4"
notify = "4.0"
once_cell = "1.10"
rand = "0.8"
regex = "1"
serde = { version = "1.0", features = ["derive"] }

4
src/filter/ingest_filter.rs
View File

@ -21,9 +21,7 @@ fn audio_filter(config: &GlobalConfig) -> String {
}
/// Create filter nodes for ingest live stream.
pub fn filter_cmd() -> Vec<String> {
let config = GlobalConfig::global();
pub fn filter_cmd(config: &GlobalConfig) -> Vec<String> {
let mut filter = format!(
"[0:v]fps={},scale={}:{},setdar=dar={},fade=in:st=0:d=0.5",
config.processing.fps,

8
src/filter/mod.rs
View File

@ -182,7 +182,7 @@ fn extend_video(node: &mut Media, chain: &mut Filters) {
/// add drawtext filter for lower thirds messages
fn add_text(node: &mut Media, chain: &mut Filters, config: &GlobalConfig) {
if config.text.add_text && config.text.over_pre {
let filter = v_drawtext::filter_node(node);
let filter = v_drawtext::filter_node(config, node);
chain.add_filter(&filter, "video");
@ -277,7 +277,7 @@ fn realtime_filter(node: &mut Media, chain: &mut Filters, config: &GlobalConfig,
if &config.out.mode.to_lowercase() == "hls" {
let mut speed_filter = format!("{t}realtime=speed=1");
let (delta, _) = get_delta(&node.begin.unwrap());
let (delta, _) = get_delta(config, &node.begin.unwrap());
let duration = node.out - node.seek;
if delta < 0.0 {
@ -292,9 +292,7 @@ fn realtime_filter(node: &mut Media, chain: &mut Filters, config: &GlobalConfig,
}
}
pub fn filter_chains(node: &mut Media) -> Vec<String> {
let config = GlobalConfig::global();
pub fn filter_chains(config: &GlobalConfig, node: &mut Media) -> Vec<String> {
let mut filters = Filters::new();
let mut audio_map = "1:a".to_string();
filters.audio_map = Some(audio_map);

3
src/filter/v_drawtext.rs
View File

@ -4,8 +4,7 @@ use regex::Regex;
use crate::utils::{GlobalConfig, Media};
pub fn filter_node(node: &mut Media) -> String {
let config = GlobalConfig::global();
pub fn filter_node(config: &GlobalConfig, node: &mut Media) -> String {
let mut filter = String::new();
let mut font = String::new();

16
src/input/folder.rs
View File

@ -33,8 +33,11 @@ pub struct FolderSource {
}
impl FolderSource {
pub fn new(current_list: Arc<Mutex<Vec<Media>>>, global_index: Arc<AtomicUsize>) -> Self {
let config = GlobalConfig::global();
pub fn new(
config: &GlobalConfig,
current_list: Arc<Mutex<Vec<Media>>>,
global_index: Arc<AtomicUsize>,
) -> Self {
let mut media_list = vec![];
let mut index: usize = 0;
@ -120,7 +123,7 @@ impl Iterator for FolderSource {
let i = self.index.load(Ordering::SeqCst);
self.current_node = self.nodes.lock().unwrap()[i].clone();
self.current_node.add_probe();
self.current_node.add_filter();
self.current_node.add_filter(&self.config);
self.current_node.begin = Some(get_sec());
self.index.fetch_add(1, Ordering::SeqCst);
@ -143,7 +146,7 @@ impl Iterator for FolderSource {
self.current_node = self.nodes.lock().unwrap()[0].clone();
self.current_node.add_probe();
self.current_node.add_filter();
self.current_node.add_filter(&self.config);
self.current_node.begin = Some(get_sec());
self.index.store(1, Ordering::SeqCst);
@ -160,11 +163,10 @@ fn file_extension(filename: &Path) -> Option<&str> {
/// Create a watcher, which monitor file changes.
/// When a change is register, update the current file list.
/// This makes it possible, to play infinitely and and always new files to it.
pub fn watchman(sources: Arc<Mutex<Vec<Media>>>) {
let config = GlobalConfig::global();
pub fn watchman(config: GlobalConfig, sources: Arc<Mutex<Vec<Media>>>) {
let (tx, rx) = channel();
let path = config.storage.path.clone();
let path = config.storage.path;
if !Path::new(&path).exists() {
error!("Folder path not exists: '{path}'");

6
src/input/ingest.rs
View File

@ -16,18 +16,18 @@ use crate::vec_strings;
///
/// Start ffmpeg in listen mode, and wait for input.
pub fn ingest_server(
config: GlobalConfig,
log_format: String,
ingest_sender: Sender<(usize, [u8; 65088])>,
mut proc_control: ProcessControl,
) -> Result<(), Error> {
let config = GlobalConfig::global();
let mut buffer: [u8; 65088] = [0; 65088];
let mut server_cmd = vec_strings!["-hide_banner", "-nostats", "-v", log_format];
let stream_input = config.ingest.input_cmd.clone().unwrap();
server_cmd.append(&mut stream_input.clone());
server_cmd.append(&mut filter_cmd());
server_cmd.append(&mut config.processing.settings.clone().unwrap());
server_cmd.append(&mut filter_cmd(&config));
server_cmd.append(&mut config.processing.settings.unwrap());
let mut is_running;

8
src/input/mod.rs
View File

@ -35,17 +35,19 @@ pub fn source_generator(
&config.storage.path
);
let folder_source = FolderSource::new(current_list, index);
let config_clone = config.clone();
let folder_source = FolderSource::new(&config, current_list, index);
let node_clone = folder_source.nodes.clone();
// Spawn a thread to monitor folder for file changes.
thread::spawn(move || watchman(node_clone));
thread::spawn(move || watchman(config_clone, node_clone));
Box::new(folder_source) as Box<dyn Iterator<Item = Media>>
}
"playlist" => {
info!("Playout in playlist mode");
let program = CurrentProgram::new(playout_stat, is_terminated, current_list, index);
let program =
CurrentProgram::new(&config, playout_stat, is_terminated, current_list, index);
Box::new(program) as Box<dyn Iterator<Item = Media>>
}

52
src/input/playlist.rs
View File

@ -34,13 +34,13 @@ pub struct CurrentProgram {
impl CurrentProgram {
pub fn new(
config: &GlobalConfig,
playout_stat: PlayoutStatus,
is_terminated: Arc<AtomicBool>,
current_list: Arc<Mutex<Vec<Media>>>,
global_index: Arc<AtomicUsize>,
) -> Self {
let config = GlobalConfig::global();
let json = read_json(None, is_terminated.clone(), true, 0.0);
let json = read_json(config, None, is_terminated.clone(), true, 0.0);
*current_list.lock().unwrap() = json.program;
*playout_stat.current_date.lock().unwrap() = json.date.clone();
@ -72,7 +72,7 @@ impl CurrentProgram {
// Check if playlist file got updated, and when yes we reload it and setup everything in place.
fn check_update(&mut self, seek: bool) {
if self.json_path.is_none() {
let json = read_json(None, self.is_terminated.clone(), seek, 0.0);
let json = read_json(&self.config, None, self.is_terminated.clone(), seek, 0.0);
self.json_path = json.current_file;
self.json_mod = json.modified;
@ -92,6 +92,7 @@ impl CurrentProgram {
);
let json = read_json(
&self.config,
self.json_path.clone(),
self.is_terminated.clone(),
false,
@ -127,7 +128,7 @@ impl CurrentProgram {
let current_time = get_sec();
let start_sec = self.config.playlist.start_sec.unwrap();
let target_length = self.config.playlist.length_sec.unwrap();
let (delta, total_delta) = get_delta(&current_time);
let (delta, total_delta) = get_delta(&self.config, &current_time);
let mut duration = self.current_node.out;
if self.current_node.duration > self.current_node.out {
@ -144,7 +145,13 @@ impl CurrentProgram {
|| is_close(total_delta, 0.0, 2.0)
|| is_close(total_delta, target_length, 2.0)
{
let json = read_json(None, self.is_terminated.clone(), false, next_start);
let json = read_json(
&self.config,
None,
self.is_terminated.clone(),
false,
next_start,
);
let data = json!({
"time_shift": 0.0,
@ -236,7 +243,7 @@ impl CurrentProgram {
let mut node_clone = self.nodes.lock().unwrap()[index].clone();
node_clone.seek = time_sec - node_clone.begin.unwrap();
self.current_node = handle_list_init(node_clone);
self.current_node = handle_list_init(&self.config, node_clone);
}
}
}
@ -271,7 +278,7 @@ impl Iterator for CurrentProgram {
self.init_clip();
} else {
let mut current_time = get_sec();
let (_, total_delta) = get_delta(&current_time);
let (_, total_delta) = get_delta(&self.config, &current_time);
let mut duration = DUMMY_LEN;
if DUMMY_LEN > total_delta {
@ -288,7 +295,7 @@ impl Iterator for CurrentProgram {
media.duration = duration;
media.out = duration;
self.current_node = gen_source(media);
self.current_node = gen_source(&self.config, media);
self.nodes.lock().unwrap().push(self.current_node.clone());
self.index
.store(self.nodes.lock().unwrap().len(), Ordering::SeqCst);
@ -326,7 +333,8 @@ impl Iterator for CurrentProgram {
let last_playlist = self.json_path.clone();
let last_ad = self.current_node.last_ad;
self.check_for_next_playlist();
let (_, total_delta) = get_delta(&self.config.playlist.start_sec.unwrap());
let (_, total_delta) =
get_delta(&self.config, &self.config.playlist.start_sec.unwrap());
if last_playlist == self.json_path
&& total_delta.abs() > self.config.general.stop_threshold
@ -343,12 +351,12 @@ impl Iterator for CurrentProgram {
}
self.current_node.duration = duration;
self.current_node.out = duration;
self.current_node = gen_source(self.current_node.clone());
self.current_node = gen_source(&self.config, self.current_node.clone());
self.nodes.lock().unwrap().push(self.current_node.clone());
self.last_next_ad();
self.current_node.last_ad = last_ad;
self.current_node.add_filter();
self.current_node.add_filter(&self.config);
self.index.fetch_add(1, Ordering::SeqCst);
@ -356,7 +364,7 @@ impl Iterator for CurrentProgram {
}
self.index.store(0, Ordering::SeqCst);
self.current_node = gen_source(self.nodes.lock().unwrap()[0].clone());
self.current_node = gen_source(&self.config, self.nodes.lock().unwrap()[0].clone());
self.last_next_ad();
self.current_node.last_ad = last_ad;
@ -377,7 +385,7 @@ fn timed_source(
last: bool,
playout_stat: &PlayoutStatus,
) -> Media {
let (delta, total_delta) = get_delta(&node.begin.unwrap());
let (delta, total_delta) = get_delta(config, &node.begin.unwrap());
let mut shifted_delta = delta;
let mut new_node = node.clone();
new_node.process = Some(false);
@ -397,7 +405,7 @@ fn timed_source(
debug!("Total time remaining: <yellow>{total_delta:.3}</>");
let sync = check_sync(shifted_delta);
let sync = check_sync(config, shifted_delta);
if !sync {
new_node.cmd = None;
@ -411,7 +419,7 @@ fn timed_source(
|| !config.playlist.length.contains(':')
{
// when we are in the 24 hour range, get the clip
new_node = gen_source(node);
new_node = gen_source(config, node);
new_node.process = Some(true);
} else if total_delta <= 0.0 {
info!("Begin is over play time, skip: {}", node.source);
@ -423,7 +431,7 @@ fn timed_source(
}
/// Generate the source CMD, or when clip not exist, get a dummy.
fn gen_source(mut node: Media) -> Media {
fn gen_source(config: &GlobalConfig, mut node: Media) -> Media {
if Path::new(&node.source).is_file() {
node.add_probe();
node.cmd = Some(seek_and_length(
@ -432,7 +440,7 @@ fn gen_source(mut node: Media) -> Media {
node.out,
node.duration,
));
node.add_filter();
node.add_filter(config);
} else {
if node.source.chars().count() == 0 {
warn!(
@ -442,10 +450,10 @@ fn gen_source(mut node: Media) -> Media {
} else {
error!("File not found: <b><magenta>{}</></b>", node.source);
}
let (source, cmd) = gen_dummy(node.out - node.seek);
let (source, cmd) = gen_dummy(config, node.out - node.seek);
node.source = source;
node.cmd = Some(cmd);
node.add_filter();
node.add_filter(config);
}
node
@ -453,9 +461,9 @@ fn gen_source(mut node: Media) -> Media {
/// Handle init clip, but this clip can be the last one in playlist,
/// this we have to figure out and calculate the right length.
fn handle_list_init(mut node: Media) -> Media {
fn handle_list_init(config: &GlobalConfig, mut node: Media) -> Media {
debug!("Playlist init");
let (_, total_delta) = get_delta(&node.begin.unwrap());
let (_, total_delta) = get_delta(config, &node.begin.unwrap());
let mut out = node.out;
if node.out - node.seek > total_delta {
@ -463,7 +471,7 @@ fn handle_list_init(mut node: Media) -> Media {
}
node.out = out;
gen_source(node)
gen_source(config, node)
}
/// when we come to last clip in playlist,

21
src/main.rs
View File

@ -23,8 +23,8 @@ mod utils;
use crate::output::{player, write_hls};
use crate::utils::{
generate_playlist, init_config, init_logging, validate_ffmpeg, GlobalConfig, PlayerControl,
PlayoutStatus, ProcessControl,
generate_playlist, init_logging, validate_ffmpeg, GlobalConfig, PlayerControl, PlayoutStatus,
ProcessControl,
};
use rpc::json_rpc_server;
@ -65,22 +65,21 @@ fn status_file(stat_file: &str, playout_stat: &PlayoutStatus) {
}
fn main() {
// Init the config, set process controller, create logging.
init_config(None);
let config = GlobalConfig::global();
let config = GlobalConfig::new();
let config_clone = config.clone();
let play_control = PlayerControl::new();
let playout_stat = PlayoutStatus::new();
let proc_control = ProcessControl::new();
let logging = init_logging();
let logging = init_logging(&config);
CombinedLogger::init(logging).unwrap();
validate_ffmpeg();
validate_ffmpeg(&config);
status_file(&config.general.stat_file, &playout_stat);
if let Some(range) = config.general.generate.clone() {
// run a simple playlist generator and save them to disk
generate_playlist(range);
generate_playlist(&config, range);
exit(0);
}
@ -91,15 +90,15 @@ fn main() {
if config.rpc_server.enable {
// If RPC server is enable we also fire up a JSON RPC server.
thread::spawn(move || json_rpc_server(play_ctl, play_stat, proc_ctl));
thread::spawn(move || json_rpc_server(config_clone, play_ctl, play_stat, proc_ctl));
}
if &config.out.mode.to_lowercase() == "hls" {
// write files/playlist to HLS m3u8 playlist
write_hls(play_control, playout_stat, proc_control);
write_hls(&config, play_control, playout_stat, proc_control);
} else {
// play on desktop or stream to a remote target
player(play_control, playout_stat, proc_control);
player(&config, play_control, playout_stat, proc_control);
}
info!("Playout done...");

8
src/output/desktop.rs
View File

@ -9,9 +9,7 @@ use crate::vec_strings;
/// Desktop Output
///
/// Instead of streaming, we run a ffplay instance and play on desktop.
pub fn output(log_format: &str) -> process::Child {
let config = GlobalConfig::global();
pub fn output(config: &GlobalConfig, log_format: &str) -> process::Child {
let mut enc_filter: Vec<String> = vec![];
let mut enc_cmd = vec_strings!["-hide_banner", "-nostats", "-v", log_format, "-i", "pipe:0"];
@ -23,7 +21,9 @@ pub fn output(log_format: &str) -> process::Child {
);
let mut filter: String = "null,".to_string();
filter.push_str(v_drawtext::filter_node(&mut Media::new(0, String::new(), false)).as_str());
filter.push_str(
v_drawtext::filter_node(config, &mut Media::new(0, String::new(), false)).as_str(),
);
enc_filter = vec!["-vf".to_string(), filter];
}

9
src/output/hls.rs
View File

@ -41,17 +41,17 @@ fn format_line(line: String, level: &str) -> String {
/// Ingest Server for HLS
fn ingest_to_hls_server(
config: GlobalConfig,
playout_stat: PlayoutStatus,
mut proc_control: ProcessControl,
) -> Result<(), Error> {
let config = GlobalConfig::global();
let playlist_init = playout_stat.list_init;
let mut server_cmd = vec_strings!["-hide_banner", "-nostats", "-v", "level+info"];
let stream_input = config.ingest.input_cmd.clone().unwrap();
server_cmd.append(&mut stream_input.clone());
server_cmd.append(&mut filter_cmd());
server_cmd.append(&mut filter_cmd(&config));
server_cmd.append(&mut config.out.clone().output_cmd.unwrap());
let mut is_running;
@ -130,11 +130,12 @@ fn ingest_to_hls_server(
///
/// Write with single ffmpeg instance directly to a HLS playlist.
pub fn write_hls(
config: &GlobalConfig,
play_control: PlayerControl,
playout_stat: PlayoutStatus,
mut proc_control: ProcessControl,
) {
let config = GlobalConfig::global();
let config_clone = config.clone();
let ff_log_format = format!("level+{}", config.logging.ffmpeg_level.to_lowercase());
let play_stat = playout_stat.clone();
let proc_control_c = proc_control.clone();
@ -149,7 +150,7 @@ pub fn write_hls(
// spawn a thread for ffmpeg ingest server and create a channel for package sending
if config.ingest.enable {
thread::spawn(move || ingest_to_hls_server(play_stat, proc_control_c));
thread::spawn(move || ingest_to_hls_server(config_clone, play_stat, proc_control_c));
}
for node in get_source {

11
src/output/mod.rs
View File

@ -32,11 +32,12 @@ use crate::vec_strings;
/// When a live ingest arrive, it stops the current playing and switch to the live source.
/// When ingest stops, it switch back to playlist/folder mode.
pub fn player(
config: &GlobalConfig,
play_control: PlayerControl,
playout_stat: PlayoutStatus,
mut proc_control: ProcessControl,
) {
let config = GlobalConfig::global();
let config_clone = config.clone();
let ff_log_format = format!("level+{}", config.logging.ffmpeg_level.to_lowercase());
let mut buffer = [0; 65088];
let mut live_on = false;
@ -53,8 +54,8 @@ pub fn player(
// get ffmpeg output instance
let mut enc_proc = match config.out.mode.as_str() {
"desktop" => desktop::output(&ff_log_format),
"stream" => stream::output(&ff_log_format),
"desktop" => desktop::output(config, &ff_log_format),
"stream" => stream::output(config, &ff_log_format),
_ => panic!("Output mode doesn't exists!"),
};
@ -74,7 +75,9 @@ pub fn player(
if config.ingest.enable {
let (ingest_sender, rx) = bounded(96);
ingest_receiver = Some(rx);
thread::spawn(move || ingest_server(ff_log_format_c, ingest_sender, proc_control_c));
thread::spawn(move || {
ingest_server(config_clone, ff_log_format_c, ingest_sender, proc_control_c)
});
}
'source_iter: for node in get_source {

7
src/output/stream.rs
View File

@ -12,8 +12,7 @@ use crate::vec_strings;
/// Streaming Output
///
/// Prepare the ffmpeg command for streaming output
pub fn output(log_format: &str) -> process::Child {
let config = GlobalConfig::global();
pub fn output(config: &GlobalConfig, log_format: &str) -> process::Child {
let mut enc_filter: Vec<String> = vec![];
let mut preview: Vec<String> = vec_strings![];
let mut preview_cmd = config.out.preview_cmd.as_ref().unwrap().clone();
@ -36,7 +35,9 @@ pub fn output(log_format: &str) -> process::Child {
);
let mut filter = "[0:v]null,".to_string();
filter.push_str(v_drawtext::filter_node(&mut Media::new(0, String::new(), false)).as_str());
filter.push_str(
v_drawtext::filter_node(config, &mut Media::new(0, String::new(), false)).as_str(),
);
if config.out.preview {
filter.push_str(",split=2[v_out1][v_out2]");

26
src/rpc/mod.rs
View File

@ -56,11 +56,13 @@ fn get_data_map(config: &GlobalConfig, media: Media) -> Map<String, Value> {
/// - get last clip
/// - reset player state to original clip
pub fn json_rpc_server(
config: GlobalConfig,
play_control: PlayerControl,
playout_stat: PlayoutStatus,
proc_control: ProcessControl,
) {
let config = GlobalConfig::global();
let addr = config.rpc_server.address.clone();
let auth = config.rpc_server.authorization.clone();
let mut io = IoHandler::default();
let proc = proc_control.clone();
@ -90,10 +92,10 @@ pub fn json_rpc_server(
let mut media = play_control.current_list.lock().unwrap()[index].clone();
media.add_probe();
let (delta, _) = get_delta(&media.begin.unwrap_or(0.0));
let (delta, _) = get_delta(&config, &media.begin.unwrap_or(0.0));
*time_shift = delta;
*date = current_date.clone();
write_status(&current_date, delta);
write_status(&config, &current_date, delta);
data_map.insert("operation".to_string(), json!("move_to_next"));
data_map.insert("shifted_seconds".to_string(), json!(delta));
@ -129,10 +131,10 @@ pub fn json_rpc_server(
play_control.index.fetch_sub(2, Ordering::SeqCst);
media.add_probe();
let (delta, _) = get_delta(&media.begin.unwrap_or(0.0));
let (delta, _) = get_delta(&config, &media.begin.unwrap_or(0.0));
*time_shift = delta;
*date = current_date.clone();
write_status(&current_date, delta);
write_status(&config, &current_date, delta);
data_map.insert("operation".to_string(), json!("move_to_last"));
data_map.insert("shifted_seconds".to_string(), json!(delta));
@ -164,7 +166,7 @@ pub fn json_rpc_server(
*date = current_date.clone();
playout_stat.list_init.store(true, Ordering::SeqCst);
write_status(&current_date, 0.0);
write_status(&config, &current_date, 0.0);
data_map.insert("operation".to_string(), json!("reset_playout_state"));
@ -177,7 +179,7 @@ pub fn json_rpc_server(
// get infos about current clip
if map.contains_key("media") && &map["media"] == "current" {
if let Some(media) = play_control.current_media.lock().unwrap().clone() {
let data_map = get_data_map(config, media);
let data_map = get_data_map(&config, media);
return Ok(Value::Object(data_map));
};
@ -190,7 +192,7 @@ pub fn json_rpc_server(
if index < play_control.current_list.lock().unwrap().len() {
let media = play_control.current_list.lock().unwrap()[index].clone();
let data_map = get_data_map(config, media);
let data_map = get_data_map(&config, media);
return Ok(Value::Object(data_map));
}
@ -205,7 +207,7 @@ pub fn json_rpc_server(
if index > 1 && index - 2 < play_control.current_list.lock().unwrap().len() {
let media = play_control.current_list.lock().unwrap()[index - 2].clone();
let data_map = get_data_map(config, media);
let data_map = get_data_map(&config, media);
return Ok(Value::Object(data_map));
}
@ -223,9 +225,9 @@ pub fn json_rpc_server(
AccessControlAllowOrigin::Null,
]))
// add middleware, for authentication
.request_middleware(|request: hyper::Request<hyper::Body>| {
.request_middleware(move |request: hyper::Request<hyper::Body>| {
if request.headers().contains_key("authorization")
&& request.headers()["authorization"] == config.rpc_server.authorization
&& request.headers()["authorization"] == auth
{
if request.uri() == "/status" {
println!("{:?}", request.headers().contains_key("authorization"));
@ -238,7 +240,7 @@ pub fn json_rpc_server(
}
})
.rest_api(RestApi::Secure)
.start_http(&config.rpc_server.address.parse().unwrap())
.start_http(&addr.parse().unwrap())
.expect("Unable to start RPC server");
*proc_control.rpc_handle.lock().unwrap() = Some(server.close_handle());

60
src/tests/mod.rs
View File

@ -21,55 +21,49 @@ fn timed_kill(sec: u64, mut proc_ctl: ProcessControl) {
#[test]
#[ignore]
fn playlist_change_at_midnight() {
let config = TestConfig {
mode: "playlist".into(),
start: "00:00:00".into(),
length: "24:00:00".into(),
log_to_file: false,
mail_recipient: "".into(),
};
init_config(Some(config));
let mut config = GlobalConfig::new();
config.mail.recipient = "".into();
config.processing.mode = "playlist".into();
config.playlist.day_start = "00:00:00".into();
config.playlist.length = "24:00:00".into();
config.logging.log_to_file = false;
let play_control = PlayerControl::new();
let playout_stat = PlayoutStatus::new();
let proc_control = ProcessControl::new();
let proc_ctl = proc_control.clone();
let logging = init_logging();
let logging = init_logging(&config);
CombinedLogger::init(logging).unwrap();
mock_time::set_mock_time("2022-05-09T23:59:45");
thread::spawn(move || timed_kill(30, proc_ctl));
player(play_control, playout_stat, proc_control);
player(&config, play_control, playout_stat, proc_control);
}
// #[test]
// #[ignore]
// fn playlist_change_at_six() {
// let config = TestConfig {
// mode: "playlist".into(),
// start: "06:00:00".into(),
// length: "24:00:00".into(),
// log_to_file: false,
// mail_recipient: "".into(),
// };
#[test]
#[ignore]
fn playlist_change_at_six() {
let mut config = GlobalConfig::new();
config.mail.recipient = "".into();
config.processing.mode = "playlist".into();
config.playlist.day_start = "06:00:00".into();
config.playlist.length = "24:00:00".into();
config.logging.log_to_file = false;
// init_config(Some(config));
let play_control = PlayerControl::new();
let playout_stat = PlayoutStatus::new();
let proc_control = ProcessControl::new();
let proc_ctl = proc_control.clone();
// let play_control = PlayerControl::new();
// let playout_stat = PlayoutStatus::new();
// let proc_control = ProcessControl::new();
// let proc_ctl = proc_control.clone();
let logging = init_logging(&config);
CombinedLogger::init(logging).unwrap();
// let logging = init_logging();
// CombinedLogger::init(logging).unwrap();
mock_time::set_mock_time("2022-05-09T05:59:45");
// mock_time::set_mock_time("2022-05-09T05:59:45");
thread::spawn(move || timed_kill(30, proc_ctl));
// thread::spawn(move || timed_kill(30, proc_ctl));
// player(play_control, playout_stat, proc_control);
// }
player(&config, play_control, playout_stat, proc_control);
}

16
src/tests/utils/mod.rs
View File

@ -38,17 +38,15 @@ fn get_date_tomorrow() {
#[test]
fn test_delta() {
let config = TestConfig {
mode: "playlist".into(),
start: "00:00:00".into(),
length: "24:00:00".into(),
log_to_file: false,
mail_recipient: "".into(),
};
let mut config = GlobalConfig::new();
config.mail.recipient = "".into();
config.processing.mode = "playlist".into();
config.playlist.day_start = "00:00:00".into();
config.playlist.length = "24:00:00".into();
config.logging.log_to_file = false;
init_config(Some(config));
mock_time::set_mock_time("2022-05-09T23:59:59");
let (delta, _) = get_delta(&86401.0);
let (delta, _) = get_delta(&config, &86401.0);
assert!(delta < 2.0);
}

34
src/utils/config.rs
View File

@ -5,12 +5,10 @@ use std::{
process,
};
use once_cell::sync::OnceCell;
use serde::{Deserialize, Serialize};
use serde_yaml::{self};
use shlex::split;
use crate::utils::{get_args, time_to_sec, TestConfig};
use crate::utils::{get_args, time_to_sec};
use crate::vec_strings;
/// Global Config
@ -138,7 +136,7 @@ pub struct Out {
impl GlobalConfig {
/// Read config from YAML file, and set some extra config values.
fn new() -> Self {
pub fn new() -> Self {
let args = get_args();
let mut config_path = match env::current_exe() {
Ok(path) => path.parent().unwrap().join("ffplayout.yml"),
@ -270,14 +268,8 @@ impl GlobalConfig {
config
}
pub fn global() -> &'static GlobalConfig {
INSTANCE.get_or_init(GlobalConfig::new)
}
}
static INSTANCE: OnceCell<GlobalConfig> = OnceCell::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.
@ -290,25 +282,3 @@ fn pre_audio_codec(add_loudnorm: bool) -> Vec<String> {
codec
}
#[cfg(not(test))]
pub fn init_config(_: Option<TestConfig>) {
let config = GlobalConfig::new();
INSTANCE.set(config).unwrap();
}
#[cfg(test)]
pub fn init_config(test_config: Option<TestConfig>) {
let mut config = GlobalConfig::new();
config.out.mode = "desktop".into();
if let Some(cfg) = test_config {
config.logging.log_to_file = cfg.log_to_file;
config.mail.recipient = cfg.mail_recipient;
config.playlist.day_start = cfg.start.clone();
config.playlist.start_sec = Some(time_to_sec(&cfg.start));
config.playlist.length = cfg.length.clone();
config.playlist.length_sec = Some(time_to_sec(&cfg.length));
};
INSTANCE.set(config).unwrap();
}

4
src/utils/controller.rs
View File

@ -136,7 +136,9 @@ impl ProcessControl {
for unit in [Encoder, Decoder, Ingest] {
if let Err(e) = self.kill(unit) {
error!("{e}")
if !e.contains("exited process") {
error!("{e}")
}
}
}
}

5
src/utils/generator.rs
View File

@ -50,8 +50,7 @@ fn get_date_range(date_range: &[String]) -> Vec<String> {
}
/// Generate playlists
pub fn generate_playlist(mut date_range: Vec<String>) {
let config = GlobalConfig::global();
pub fn generate_playlist(config: &GlobalConfig, mut date_range: Vec<String>) {
let total_length = config.playlist.length_sec.unwrap();
let current_list = Arc::new(Mutex::new(vec![Media::new(0, "".to_string(), false)]));
let index = Arc::new(AtomicUsize::new(0));
@ -70,7 +69,7 @@ pub fn generate_playlist(mut date_range: Vec<String>) {
date_range = get_date_range(&date_range)
}
let media_list = FolderSource::new(current_list, index);
let media_list = FolderSource::new(config, current_list, index);
let list_length = media_list.nodes.lock().unwrap().len();
for date in date_range {

6
src/utils/json_serializer.rs
View File

@ -48,13 +48,13 @@ impl Playlist {
/// Read json playlist file, fills Playlist struct and set some extra values,
/// which we need to process.
pub fn read_json(
config: &GlobalConfig,
path: Option<String>,
is_terminated: Arc<AtomicBool>,
seek: bool,
next_start: f64,
) -> Playlist {
let config = GlobalConfig::global();
let config_clone = config.clone();
let mut playlist_path = Path::new(&config.playlist.path).to_owned();
let mut start_sec = config.playlist.start_sec.unwrap();
let date = get_date(seek, start_sec, next_start);
@ -113,7 +113,7 @@ pub fn read_json(
let list_clone = playlist.clone();
thread::spawn(move || validate_playlist(list_clone, is_terminated, config.clone()));
thread::spawn(move || validate_playlist(list_clone, is_terminated, config_clone));
playlist
}

57
src/utils/logging.rs
View File

@ -25,26 +25,21 @@ use simplelog::*;
use crate::utils::GlobalConfig;
/// send log messages to mail recipient
fn send_mail(msg: String) {
let config = GlobalConfig::global();
fn send_mail(cfg: &GlobalConfig, msg: String) {
let email = Message::builder()
.from(config.mail.sender_addr.parse().unwrap())
.to(config.mail.recipient.parse().unwrap())
.subject(config.mail.subject.clone())
.from(cfg.mail.sender_addr.parse().unwrap())
.to(cfg.mail.recipient.parse().unwrap())
.subject(cfg.mail.subject.clone())
.header(header::ContentType::TEXT_PLAIN)
.body(clean_string(&msg))
.unwrap();
let credentials = Credentials::new(
config.mail.sender_addr.clone(),
config.mail.sender_pass.clone(),
);
let credentials = Credentials::new(cfg.mail.sender_addr.clone(), cfg.mail.sender_pass.clone());
let mut transporter = SmtpTransport::relay(config.mail.smtp_server.clone().as_str());
let mut transporter = SmtpTransport::relay(cfg.mail.smtp_server.clone().as_str());
if config.mail.starttls {
transporter = SmtpTransport::starttls_relay(config.mail.smtp_server.clone().as_str())
if cfg.mail.starttls {
transporter = SmtpTransport::starttls_relay(cfg.mail.smtp_server.clone().as_str())
}
let mailer = transporter.unwrap().credentials(credentials).build();
@ -59,11 +54,11 @@ fn send_mail(msg: String) {
/// Basic Mail Queue
///
/// Check every give seconds for messages and send them.
fn mail_queue(messages: Arc<Mutex<Vec<String>>>, interval: u64) {
fn mail_queue(cfg: GlobalConfig, messages: Arc<Mutex<Vec<String>>>, interval: u64) {
loop {
if messages.lock().unwrap().len() > 0 {
let msg = messages.lock().unwrap().join("\n");
send_mail(msg);
send_mail(&cfg, msg);
messages.lock().unwrap().clear();
}
@ -141,8 +136,8 @@ fn clean_string(text: &str) -> String {
/// - console logger
/// - file logger
/// - mail logger
pub fn init_logging() -> Vec<Box<dyn SharedLogger>> {
let config = GlobalConfig::global();
pub fn init_logging(config: &GlobalConfig) -> Vec<Box<dyn SharedLogger>> {
let config_clone = config.clone();
let app_config = config.logging.clone();
let mut time_level = LevelFilter::Off;
let mut app_logger: Vec<Box<dyn SharedLogger>> = vec![];
@ -169,7 +164,7 @@ pub fn init_logging() -> Vec<Box<dyn SharedLogger>> {
let file_config = log_config
.clone()
.set_time_format_custom(format_description!(
"[[year]-[month]-[day] [hour]:[minute]:[second].[subsecond digits:5]]"
"[[[year]-[month]-[day] [hour]:[minute]:[second].[subsecond digits:5]]"
))
.build();
let mut log_path = "logs/ffplayout.log".to_string();
@ -185,20 +180,18 @@ pub fn init_logging() -> Vec<Box<dyn SharedLogger>> {
println!("Logging path not exists!")
}
let log = || {
FileRotate::new(
log_path,
AppendTimestamp::with_format(
"%Y-%m-%d",
FileLimit::MaxFiles(app_config.backup_count),
DateFrom::DateYesterday,
),
ContentLimit::Time(TimeFrequency::Daily),
Compression::None,
)
};
let log_file = FileRotate::new(
log_path,
AppendTimestamp::with_format(
"%Y-%m-%d",
FileLimit::MaxFiles(app_config.backup_count),
DateFrom::DateYesterday,
),
ContentLimit::Time(TimeFrequency::Daily),
Compression::None,
);
app_logger.push(WriteLogger::new(LevelFilter::Debug, file_config, log()));
app_logger.push(WriteLogger::new(LevelFilter::Debug, file_config, log_file));
} else {
let term_config = log_config
.clone()
@ -225,7 +218,7 @@ pub fn init_logging() -> Vec<Box<dyn SharedLogger>> {
let messages_clone = messages.clone();
let interval = config.mail.interval;
thread::spawn(move || mail_queue(messages_clone, interval));
thread::spawn(move || mail_queue(config_clone, messages_clone, interval));
let mail_config = log_config.build();

32
src/utils/mod.rs
View File

@ -26,7 +26,7 @@ mod json_validate;
mod logging;
pub use arg_parse::get_args;
pub use config::{init_config, GlobalConfig};
pub use config::GlobalConfig;
pub use controller::{PlayerControl, PlayoutStatus, ProcessControl, ProcessUnit::*};
pub use generator::generate_playlist;
pub use json_serializer::{read_json, Playlist, DUMMY_LEN};
@ -119,9 +119,9 @@ impl Media {
}
}
pub fn add_filter(&mut self) {
pub fn add_filter(&mut self, config: &GlobalConfig) {
let mut node = self.clone();
self.filter = Some(filter_chains(&mut node))
self.filter = Some(filter_chains(config, &mut node))
}
}
@ -190,8 +190,7 @@ impl MediaProbe {
/// Write current status to status file in temp folder.
///
/// The status file is init in main function and mostly modified in RPC server.
pub fn write_status(date: &str, shift: f64) {
let config = GlobalConfig::global();
pub fn write_status(config: &GlobalConfig, date: &str, shift: f64) {
let stat_file = config.general.stat_file.clone();
let data = json!({
@ -286,8 +285,7 @@ pub fn is_close(a: f64, b: f64, to: f64) -> bool {
/// if we still in sync.
///
/// We also get here the global delta between clip start and time when a new playlist should start.
pub fn get_delta(begin: &f64) -> (f64, f64) {
let config = GlobalConfig::global();
pub fn get_delta(config: &GlobalConfig, begin: &f64) -> (f64, f64) {
let mut current_time = get_sec();
let start = config.playlist.start_sec.unwrap();
let length = time_to_sec(&config.playlist.length);
@ -318,9 +316,7 @@ pub fn get_delta(begin: &f64) -> (f64, f64) {
}
/// Check if clip in playlist is in sync with global time.
pub fn check_sync(delta: f64) -> bool {
let config = GlobalConfig::global();
pub fn check_sync(config: &GlobalConfig, delta: f64) -> bool {
if delta.abs() > config.general.stop_threshold && config.general.stop_threshold > 0.0 {
error!("Clip begin out of sync for <yellow>{}</> seconds", delta);
return false;
@ -330,8 +326,7 @@ pub fn check_sync(delta: f64) -> bool {
}
/// Create a dummy clip as a placeholder for missing video files.
pub fn gen_dummy(duration: f64) -> (String, Vec<String>) {
let config = GlobalConfig::global();
pub fn gen_dummy(config: &GlobalConfig, duration: f64) -> (String, Vec<String>) {
let color = "#121212";
let source = format!(
"color=c={color}:s={}x{}:d={duration}",
@ -475,9 +470,7 @@ fn ffmpeg_libs_and_filter() -> (Vec<String>, Vec<String>) {
/// Validate ffmpeg/ffprobe/ffplay.
///
/// Check if they are in system and has all filters and codecs we need.
pub fn validate_ffmpeg() {
let config = GlobalConfig::global();
pub fn validate_ffmpeg(config: &GlobalConfig) {
is_in_system("ffmpeg");
is_in_system("ffprobe");
@ -506,15 +499,6 @@ pub fn validate_ffmpeg() {
}
}
/// In test cases we override some configuration values to fit the needs.
pub struct TestConfig {
pub mode: String,
pub start: String,
pub length: String,
pub log_to_file: bool,
pub mail_recipient: String,
}
/// Get system time, in non test case.
#[cfg(not(test))]
pub fn time_now() -> DateTime<Local> {