mirror of
https://github.com/v0l/zap-stream-core.git
synced 2025-06-16 01:28:08 +00:00
Compare commits
2 Commits
e400e969fd
...
1c651108ea
| Author | SHA1 | Date | |
|---|---|---|---|
1c651108ea
|
|||
|
5d7da09801
|
115
Cargo.lock
generated
115
Cargo.lock
generated
@ -857,9 +857,9 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "data-encoding"
|
||||
version = "2.7.0"
|
||||
version = "2.9.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "0e60eed09d8c01d3cee5b7d30acb059b76614c918fa0f992e0dd6eeb10daad6f"
|
||||
checksum = "2a2330da5de22e8a3cb63252ce2abb30116bf5265e89c0e01bc17015ce30a476"
|
||||
|
||||
[[package]]
|
||||
name = "data-url"
|
||||
@ -964,7 +964,7 @@ checksum = "4cd405aab171cb85d6735e5c8d9db038c17d3ca007a4d2c25f337935c3d90580"
|
||||
dependencies = [
|
||||
"humantime",
|
||||
"is-terminal",
|
||||
"log",
|
||||
"log 0.4.25",
|
||||
"regex",
|
||||
"termcolor",
|
||||
]
|
||||
@ -1045,12 +1045,12 @@ dependencies = [
|
||||
[[package]]
|
||||
name = "ffmpeg-rs-raw"
|
||||
version = "0.1.0"
|
||||
source = "git+https://git.v0l.io/Kieran/ffmpeg-rs-raw.git?rev=29ab0547478256c574766b4acc6fcda8ebf4cae6#29ab0547478256c574766b4acc6fcda8ebf4cae6"
|
||||
source = "git+https://github.com/v0l/ffmpeg-rs-raw.git?rev=d79693ddb0bee2e94c1db07f789523e87bf1b0fc#d79693ddb0bee2e94c1db07f789523e87bf1b0fc"
|
||||
dependencies = [
|
||||
"anyhow",
|
||||
"ffmpeg-sys-the-third",
|
||||
"libc",
|
||||
"log",
|
||||
"log 0.4.25",
|
||||
"slimbox",
|
||||
]
|
||||
|
||||
@ -1123,16 +1123,16 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "fontdb"
|
||||
version = "0.22.0"
|
||||
version = "0.23.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "a3a6f9af55fb97ad673fb7a69533eb2f967648a06fa21f8c9bb2cd6d33975716"
|
||||
checksum = "457e789b3d1202543297a350643cf459f836cade38934e7a4cf6a39e7cde2905"
|
||||
dependencies = [
|
||||
"fontconfig-parser",
|
||||
"log",
|
||||
"log 0.4.25",
|
||||
"memmap2",
|
||||
"slotmap",
|
||||
"tinyvec",
|
||||
"ttf-parser 0.24.1",
|
||||
"ttf-parser 0.25.1",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
@ -1857,9 +1857,9 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "image-webp"
|
||||
version = "0.1.3"
|
||||
version = "0.2.1"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "f79afb8cbee2ef20f59ccd477a218c12a93943d075b492015ecb1bb81f8ee904"
|
||||
checksum = "b77d01e822461baa8409e156015a1d91735549f0f2c17691bd2d996bef238f7f"
|
||||
dependencies = [
|
||||
"byteorder-lite",
|
||||
"quick-error",
|
||||
@ -2069,6 +2069,15 @@ dependencies = [
|
||||
"scopeguard",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "log"
|
||||
version = "0.3.9"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "e19e8d5c34a3e0e2223db8e060f9e8264aeeb5c5fc64a4ee9965c062211c024b"
|
||||
dependencies = [
|
||||
"log 0.4.25",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "log"
|
||||
version = "0.4.25"
|
||||
@ -2161,6 +2170,16 @@ version = "0.10.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "defc4c55412d89136f966bbb339008b474350e5e6e78d2714439c386b3137a03"
|
||||
|
||||
[[package]]
|
||||
name = "mustache"
|
||||
version = "0.9.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "51956ef1c5d20a1384524d91e616fb44dfc7d8f249bf696d49c97dd3289ecab5"
|
||||
dependencies = [
|
||||
"log 0.3.9",
|
||||
"serde",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "native-tls"
|
||||
version = "0.2.13"
|
||||
@ -2168,7 +2187,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "0dab59f8e050d5df8e4dd87d9206fb6f65a483e20ac9fda365ade4fab353196c"
|
||||
dependencies = [
|
||||
"libc",
|
||||
"log",
|
||||
"log 0.4.25",
|
||||
"openssl",
|
||||
"openssl-probe",
|
||||
"openssl-sys",
|
||||
@ -2627,7 +2646,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "865724d4dbe39d9f3dd3b52b88d859d66bcb2d6a0acfd5ea68a65fb66d4bdc1c"
|
||||
dependencies = [
|
||||
"env_logger",
|
||||
"log",
|
||||
"log 0.4.25",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
@ -2668,7 +2687,7 @@ dependencies = [
|
||||
"bytes",
|
||||
"heck",
|
||||
"itertools 0.12.1",
|
||||
"log",
|
||||
"log 0.4.25",
|
||||
"multimap",
|
||||
"once_cell",
|
||||
"petgraph",
|
||||
@ -2806,7 +2825,7 @@ dependencies = [
|
||||
"hyper-util",
|
||||
"ipnet",
|
||||
"js-sys",
|
||||
"log",
|
||||
"log 0.4.25",
|
||||
"mime",
|
||||
"native-tls",
|
||||
"once_cell",
|
||||
@ -2833,13 +2852,13 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "resvg"
|
||||
version = "0.44.0"
|
||||
version = "0.45.1"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "4a325d5e8d1cebddd070b13f44cec8071594ab67d1012797c121f27a669b7958"
|
||||
checksum = "a8928798c0a55e03c9ca6c4c6846f76377427d2c1e1f7e6de3c06ae57942df43"
|
||||
dependencies = [
|
||||
"gif",
|
||||
"image-webp",
|
||||
"log",
|
||||
"log 0.4.25",
|
||||
"pico-args",
|
||||
"rgb",
|
||||
"svgtypes",
|
||||
@ -2996,7 +3015,7 @@ version = "0.21.12"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "3f56a14d1f48b391359b22f731fd4bd7e43c97f3c50eee276f3aa09c94784d3e"
|
||||
dependencies = [
|
||||
"log",
|
||||
"log 0.4.25",
|
||||
"ring",
|
||||
"rustls-webpki 0.101.7",
|
||||
"sct",
|
||||
@ -3069,16 +3088,16 @@ checksum = "f7c45b9784283f1b2e7fb61b42047c2fd678ef0960d4f6f1eba131594cc369d4"
|
||||
|
||||
[[package]]
|
||||
name = "rustybuzz"
|
||||
version = "0.18.0"
|
||||
version = "0.20.1"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "c85d1ccd519e61834798eb52c4e886e8c2d7d698dd3d6ce0b1b47eb8557f1181"
|
||||
checksum = "fd3c7c96f8a08ee34eff8857b11b49b07d71d1c3f4e88f8a88d4c9e9f90b1702"
|
||||
dependencies = [
|
||||
"bitflags 2.8.0",
|
||||
"bytemuck",
|
||||
"core_maths",
|
||||
"log",
|
||||
"log 0.4.25",
|
||||
"smallvec",
|
||||
"ttf-parser 0.24.1",
|
||||
"ttf-parser 0.25.1",
|
||||
"unicode-bidi-mirroring",
|
||||
"unicode-ccc",
|
||||
"unicode-properties",
|
||||
@ -3315,7 +3334,7 @@ version = "0.2.2"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "7a9c6883ca9c3c7c90e888de77b7a5c849c779d25d74a1269b0218b14e8b136c"
|
||||
dependencies = [
|
||||
"log",
|
||||
"log 0.4.25",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
@ -3418,7 +3437,7 @@ dependencies = [
|
||||
"hashbrown 0.15.2",
|
||||
"hashlink 0.10.0",
|
||||
"indexmap 2.7.1",
|
||||
"log",
|
||||
"log 0.4.25",
|
||||
"memchr",
|
||||
"once_cell",
|
||||
"percent-encoding",
|
||||
@ -3497,7 +3516,7 @@ dependencies = [
|
||||
"hkdf",
|
||||
"hmac 0.12.1",
|
||||
"itoa",
|
||||
"log",
|
||||
"log 0.4.25",
|
||||
"md-5",
|
||||
"memchr",
|
||||
"once_cell",
|
||||
@ -3537,7 +3556,7 @@ dependencies = [
|
||||
"hmac 0.12.1",
|
||||
"home",
|
||||
"itoa",
|
||||
"log",
|
||||
"log 0.4.25",
|
||||
"md-5",
|
||||
"memchr",
|
||||
"once_cell",
|
||||
@ -3568,7 +3587,7 @@ dependencies = [
|
||||
"futures-intrusive",
|
||||
"futures-util",
|
||||
"libsqlite3-sys",
|
||||
"log",
|
||||
"log 0.4.25",
|
||||
"percent-encoding",
|
||||
"serde",
|
||||
"serde_urlencoded",
|
||||
@ -3594,7 +3613,7 @@ dependencies = [
|
||||
"hex",
|
||||
"hmac 0.12.1",
|
||||
"keyed_priority_queue",
|
||||
"log",
|
||||
"log 0.4.25",
|
||||
"pbkdf2",
|
||||
"rand",
|
||||
"regex",
|
||||
@ -3613,7 +3632,7 @@ checksum = "0a55cb90afac5672b00954e3291846dd262cfef3b52d1b507f580180433373d3"
|
||||
dependencies = [
|
||||
"bytes",
|
||||
"futures",
|
||||
"log",
|
||||
"log 0.4.25",
|
||||
"rand",
|
||||
"socket2",
|
||||
"srt-protocol",
|
||||
@ -3824,7 +3843,7 @@ dependencies = [
|
||||
"arrayvec",
|
||||
"bytemuck",
|
||||
"cfg-if",
|
||||
"log",
|
||||
"log 0.4.25",
|
||||
"png",
|
||||
"tiny-skia-path",
|
||||
]
|
||||
@ -3963,7 +3982,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "edc5f74e248dc973e0dbb7b74c7e0d6fcc301c694ff50049504004ef4d0cdcd9"
|
||||
dependencies = [
|
||||
"futures-util",
|
||||
"log",
|
||||
"log 0.4.25",
|
||||
"rustls 0.23.21",
|
||||
"rustls-pki-types",
|
||||
"tokio",
|
||||
@ -4115,7 +4134,7 @@ version = "0.1.41"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "784e0ac535deb450455cbfa28a6f0df145ea1bb7ae51b821cf5e7927fdcfbdd0"
|
||||
dependencies = [
|
||||
"log",
|
||||
"log 0.4.25",
|
||||
"pin-project-lite",
|
||||
"tracing-attributes",
|
||||
"tracing-core",
|
||||
@ -4161,9 +4180,9 @@ checksum = "2c591d83f69777866b9126b24c6dd9a18351f177e49d625920d19f989fd31cf8"
|
||||
|
||||
[[package]]
|
||||
name = "ttf-parser"
|
||||
version = "0.24.1"
|
||||
version = "0.25.1"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "5be21190ff5d38e8b4a2d3b6a3ae57f612cc39c96e83cedeaf7abc338a8bac4a"
|
||||
checksum = "d2df906b07856748fa3f6e0ad0cbaa047052d4a7dd609e231c4f72cee8c36f31"
|
||||
dependencies = [
|
||||
"core_maths",
|
||||
]
|
||||
@ -4179,7 +4198,7 @@ dependencies = [
|
||||
"data-encoding",
|
||||
"http 1.2.0",
|
||||
"httparse",
|
||||
"log",
|
||||
"log 0.4.25",
|
||||
"rand",
|
||||
"rustls 0.23.21",
|
||||
"rustls-pki-types",
|
||||
@ -4208,15 +4227,15 @@ checksum = "5c1cb5db39152898a79168971543b1cb5020dff7fe43c8dc468b0885f5e29df5"
|
||||
|
||||
[[package]]
|
||||
name = "unicode-bidi-mirroring"
|
||||
version = "0.3.0"
|
||||
version = "0.4.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "64af057ad7466495ca113126be61838d8af947f41d93a949980b2389a118082f"
|
||||
checksum = "5dfa6e8c60bb66d49db113e0125ee8711b7647b5579dc7f5f19c42357ed039fe"
|
||||
|
||||
[[package]]
|
||||
name = "unicode-ccc"
|
||||
version = "0.3.0"
|
||||
version = "0.4.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "260bc6647b3893a9a90668360803a15f96b85a5257b1c3a0c3daf6ae2496de42"
|
||||
checksum = "ce61d488bcdc9bc8b5d1772c404828b17fc481c0a582b5581e95fb233aef503e"
|
||||
|
||||
[[package]]
|
||||
name = "unicode-ident"
|
||||
@ -4287,9 +4306,9 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "usvg"
|
||||
version = "0.44.0"
|
||||
version = "0.45.1"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "7447e703d7223b067607655e625e0dbca80822880248937da65966194c4864e6"
|
||||
checksum = "80be9b06fbae3b8b303400ab20778c80bbaf338f563afe567cf3c9eea17b47ef"
|
||||
dependencies = [
|
||||
"base64 0.22.1",
|
||||
"data-url",
|
||||
@ -4297,7 +4316,7 @@ dependencies = [
|
||||
"fontdb",
|
||||
"imagesize",
|
||||
"kurbo",
|
||||
"log",
|
||||
"log 0.4.25",
|
||||
"pico-args",
|
||||
"roxmltree",
|
||||
"rustybuzz",
|
||||
@ -4407,7 +4426,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "2f0a0651a5c2bc21487bde11ee802ccaf4c51935d0d3d42a6101f98161700bc6"
|
||||
dependencies = [
|
||||
"bumpalo",
|
||||
"log",
|
||||
"log 0.4.25",
|
||||
"proc-macro2",
|
||||
"quote",
|
||||
"syn",
|
||||
@ -4792,9 +4811,10 @@ dependencies = [
|
||||
"http-body-util",
|
||||
"hyper 1.6.0",
|
||||
"hyper-util",
|
||||
"log",
|
||||
"log 0.4.25",
|
||||
"m3u8-rs",
|
||||
"matchit 0.8.6",
|
||||
"mustache",
|
||||
"nostr-sdk",
|
||||
"pretty_env_logger",
|
||||
"reqwest",
|
||||
@ -4814,17 +4834,20 @@ version = "0.1.0"
|
||||
dependencies = [
|
||||
"anyhow",
|
||||
"async-trait",
|
||||
"data-encoding",
|
||||
"ffmpeg-rs-raw",
|
||||
"fontdue",
|
||||
"futures-util",
|
||||
"hex",
|
||||
"itertools 0.14.0",
|
||||
"log",
|
||||
"libc",
|
||||
"log 0.4.25",
|
||||
"m3u8-rs",
|
||||
"resvg",
|
||||
"ringbuf",
|
||||
"rml_rtmp",
|
||||
"serde",
|
||||
"sha2 0.10.8",
|
||||
"srt-tokio",
|
||||
"tiny-skia",
|
||||
"tokio",
|
||||
|
||||
@ -10,10 +10,10 @@ members = [
|
||||
opt-level = 3
|
||||
lto = true
|
||||
codegen-units = 1
|
||||
panic = "abort"
|
||||
panic = "unwind"
|
||||
|
||||
[workspace.dependencies]
|
||||
ffmpeg-rs-raw = { git = "https://git.v0l.io/Kieran/ffmpeg-rs-raw.git", rev = "29ab0547478256c574766b4acc6fcda8ebf4cae6" }
|
||||
ffmpeg-rs-raw = { git = "https://github.com/v0l/ffmpeg-rs-raw.git", rev = "d79693ddb0bee2e94c1db07f789523e87bf1b0fc" }
|
||||
tokio = { version = "1.36.0", features = ["rt", "rt-multi-thread", "macros"] }
|
||||
anyhow = { version = "^1.0.91", features = ["backtrace"] }
|
||||
async-trait = "0.1.77"
|
||||
|
||||
@ -4,18 +4,9 @@ version = "0.1.0"
|
||||
edition = "2021"
|
||||
|
||||
[features]
|
||||
default = ["test-pattern", "srt", "rtmp"]
|
||||
default = ["srt", "rtmp"]
|
||||
srt = ["dep:srt-tokio"]
|
||||
rtmp = ["dep:rml_rtmp"]
|
||||
local-overseer = [] # WIP
|
||||
webhook-overseer = [] # WIP
|
||||
test-pattern = [
|
||||
"dep:resvg",
|
||||
"dep:usvg",
|
||||
"dep:tiny-skia",
|
||||
"dep:fontdue",
|
||||
"dep:ringbuf",
|
||||
]
|
||||
|
||||
[dependencies]
|
||||
ffmpeg-rs-raw.workspace = true
|
||||
@ -27,20 +18,20 @@ uuid.workspace = true
|
||||
serde.workspace = true
|
||||
hex.workspace = true
|
||||
itertools.workspace = true
|
||||
futures-util = "0.3.30"
|
||||
m3u8-rs.workspace = true
|
||||
sha2.workspace = true
|
||||
data-encoding.workspace = true
|
||||
|
||||
futures-util = "0.3.30"
|
||||
resvg = "0.45.1"
|
||||
usvg = "0.45.1"
|
||||
tiny-skia = "0.11.4"
|
||||
fontdue = "0.9.2"
|
||||
ringbuf = "0.4.7"
|
||||
|
||||
# srt
|
||||
srt-tokio = { version = "0.4.3", optional = true }
|
||||
|
||||
# rtmp
|
||||
rml_rtmp = { version = "0.8.0", optional = true }
|
||||
|
||||
# test-pattern
|
||||
resvg = { version = "0.44.0", optional = true }
|
||||
usvg = { version = "0.44.0", optional = true }
|
||||
tiny-skia = { version = "0.11.4", optional = true }
|
||||
fontdue = { version = "0.9.2", optional = true }
|
||||
ringbuf = { version = "0.4.7", optional = true }
|
||||
libc = "0.2.169"
|
||||
431
crates/core/src/generator.rs
Normal file
431
crates/core/src/generator.rs
Normal file
@ -0,0 +1,431 @@
|
||||
use crate::overseer::IngressStream;
|
||||
use anyhow::{bail, Result};
|
||||
use ffmpeg_rs_raw::ffmpeg_sys_the_third::AVColorSpace::AVCOL_SPC_RGB;
|
||||
use ffmpeg_rs_raw::ffmpeg_sys_the_third::AVPictureType::AV_PICTURE_TYPE_NONE;
|
||||
use ffmpeg_rs_raw::ffmpeg_sys_the_third::AVPixelFormat::AV_PIX_FMT_RGBA;
|
||||
use ffmpeg_rs_raw::ffmpeg_sys_the_third::AVSampleFormat::AV_SAMPLE_FMT_FLTP;
|
||||
use ffmpeg_rs_raw::ffmpeg_sys_the_third::{
|
||||
av_channel_layout_default, av_frame_alloc, av_frame_free, av_frame_get_buffer, AVFrame,
|
||||
AVPixelFormat, AVRational,
|
||||
};
|
||||
use ffmpeg_rs_raw::Scaler;
|
||||
use fontdue::layout::{CoordinateSystem, Layout, TextStyle};
|
||||
use fontdue::Font;
|
||||
use std::mem::transmute;
|
||||
use std::time::{Duration, Instant};
|
||||
use std::{ptr, slice};
|
||||
|
||||
/// Frame generator
|
||||
pub struct FrameGenerator {
|
||||
fps: f32,
|
||||
width: u16,
|
||||
height: u16,
|
||||
video_sample_fmt: AVPixelFormat,
|
||||
|
||||
audio_sample_rate: u32,
|
||||
audio_frame_size: i32,
|
||||
audio_channels: u8,
|
||||
|
||||
frame_idx: u64,
|
||||
audio_samples: u64,
|
||||
|
||||
// internal
|
||||
next_frame: *mut AVFrame,
|
||||
scaler: Scaler,
|
||||
font: Font,
|
||||
start: Instant,
|
||||
}
|
||||
|
||||
impl Drop for FrameGenerator {
|
||||
fn drop(&mut self) {
|
||||
unsafe {
|
||||
if !self.next_frame.is_null() {
|
||||
av_frame_free(&mut self.next_frame);
|
||||
self.next_frame = std::ptr::null_mut();
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl FrameGenerator {
|
||||
pub fn new(
|
||||
fps: f32,
|
||||
width: u16,
|
||||
height: u16,
|
||||
pix_fmt: AVPixelFormat,
|
||||
sample_rate: u32,
|
||||
frame_size: i32,
|
||||
channels: u8,
|
||||
) -> Result<Self> {
|
||||
let font = include_bytes!("../SourceCodePro-Regular.ttf") as &[u8];
|
||||
let font = Font::from_bytes(font, Default::default()).unwrap();
|
||||
|
||||
Ok(Self {
|
||||
fps,
|
||||
width,
|
||||
height,
|
||||
video_sample_fmt: pix_fmt,
|
||||
audio_sample_rate: sample_rate,
|
||||
audio_frame_size: frame_size,
|
||||
audio_channels: channels,
|
||||
frame_idx: 0,
|
||||
audio_samples: 0,
|
||||
font,
|
||||
start: Instant::now(),
|
||||
scaler: Scaler::default(),
|
||||
next_frame: ptr::null_mut(),
|
||||
})
|
||||
}
|
||||
|
||||
pub fn from_stream(
|
||||
video_stream: &IngressStream,
|
||||
audio_stream: Option<&IngressStream>,
|
||||
) -> Result<Self> {
|
||||
Ok(Self::new(
|
||||
video_stream.fps,
|
||||
video_stream.width as _,
|
||||
video_stream.height as _,
|
||||
unsafe { transmute(video_stream.format as i32) },
|
||||
audio_stream.map(|i| i.sample_rate as _).unwrap_or(0),
|
||||
if audio_stream.is_none() { 0 } else { 1024 },
|
||||
audio_stream.map(|i| i.channels as _).unwrap_or(0),
|
||||
)?)
|
||||
}
|
||||
|
||||
pub fn frame_no(&self) -> u64 {
|
||||
self.frame_idx
|
||||
}
|
||||
|
||||
/// Create a new frame for composing text / images
|
||||
pub fn begin(&mut self) -> Result<()> {
|
||||
if self.next_frame.is_null() {
|
||||
unsafe {
|
||||
let mut src_frame = av_frame_alloc();
|
||||
if src_frame.is_null() {
|
||||
bail!("Failed to allocate placeholder video frame");
|
||||
}
|
||||
|
||||
(*src_frame).width = self.width as _;
|
||||
(*src_frame).height = self.height as _;
|
||||
(*src_frame).pict_type = AV_PICTURE_TYPE_NONE;
|
||||
(*src_frame).key_frame = 1;
|
||||
(*src_frame).colorspace = AVCOL_SPC_RGB;
|
||||
//internally always use RGBA, we convert frame to target pixel format at the end
|
||||
(*src_frame).format = AV_PIX_FMT_RGBA as _;
|
||||
(*src_frame).pts = self.frame_idx as _;
|
||||
(*src_frame).duration = 1;
|
||||
(*src_frame).time_base = AVRational {
|
||||
num: 1,
|
||||
den: self.fps as i32,
|
||||
};
|
||||
if av_frame_get_buffer(src_frame, 0) < 0 {
|
||||
av_frame_free(&mut src_frame);
|
||||
bail!("Failed to get frame buffer");
|
||||
}
|
||||
self.next_frame = src_frame;
|
||||
}
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Write some text into the next frame
|
||||
pub fn write_text(&mut self, msg: &str, size: f32, x: f32, y: f32) -> Result<()> {
|
||||
if self.next_frame.is_null() {
|
||||
bail!("Must call begin() before writing text")
|
||||
}
|
||||
let mut layout = Layout::new(CoordinateSystem::PositiveYDown);
|
||||
layout.append(&[&self.font], &TextStyle::new(msg, size, 0));
|
||||
|
||||
self.write_layout(layout, x, y)?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Write text layout into frame
|
||||
fn write_layout(&mut self, layout: Layout, x: f32, y: f32) -> Result<()> {
|
||||
for g in layout.glyphs() {
|
||||
let (metrics, bitmap) = self.font.rasterize_config_subpixel(g.key);
|
||||
for y1 in 0..metrics.height {
|
||||
for x1 in 0..metrics.width {
|
||||
let dst_x = x as usize + x1 + g.x as usize;
|
||||
let dst_y = y as usize + y1 + g.y as usize;
|
||||
let offset_src = (x1 + y1 * metrics.width) * 3;
|
||||
unsafe {
|
||||
let offset_dst =
|
||||
4 * dst_x + dst_y * (*self.next_frame).linesize[0] as usize;
|
||||
let pixel_dst = (*self.next_frame).data[0].add(offset_dst);
|
||||
*pixel_dst.offset(0) = bitmap[offset_src];
|
||||
*pixel_dst.offset(1) = bitmap[offset_src + 1];
|
||||
*pixel_dst.offset(2) = bitmap[offset_src + 2];
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Copy data directly into the frame buffer (must be RGBA data)
|
||||
pub unsafe fn copy_frame_data(&mut self, data: &[u8]) -> Result<()> {
|
||||
if self.next_frame.is_null() {
|
||||
bail!("Must call begin() before writing frame data")
|
||||
}
|
||||
let buf = slice::from_raw_parts_mut(
|
||||
(*self.next_frame).data[0],
|
||||
(self.width as usize * self.height as usize * 4) as usize,
|
||||
);
|
||||
if buf.len() < data.len() {
|
||||
bail!("Frame buffer is too small");
|
||||
}
|
||||
buf.copy_from_slice(data);
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Generate audio to stay synchronized with video frames
|
||||
unsafe fn generate_audio_frame(&mut self) -> Result<*mut AVFrame> {
|
||||
const FREQUENCY: f32 = 440.0; // A4 note
|
||||
|
||||
// audio is disabled if sample rate is 0
|
||||
if self.audio_sample_rate == 0 {
|
||||
return Ok(ptr::null_mut());
|
||||
}
|
||||
|
||||
// Calculate how many audio samples we need to cover the next video frame
|
||||
let samples_per_frame = (self.audio_sample_rate as f32 / self.fps) as u64;
|
||||
let next_frame_needs_samples = (self.frame_idx + 1) * samples_per_frame;
|
||||
|
||||
// Generate audio if we don't have enough to cover the next video frame
|
||||
if self.audio_samples < next_frame_needs_samples {
|
||||
let audio_frame = av_frame_alloc();
|
||||
(*audio_frame).format = AV_SAMPLE_FMT_FLTP as _;
|
||||
(*audio_frame).nb_samples = self.audio_frame_size as _;
|
||||
(*audio_frame).duration = self.audio_frame_size as _;
|
||||
(*audio_frame).sample_rate = self.audio_sample_rate as _;
|
||||
(*audio_frame).pts = self.audio_samples as _;
|
||||
(*audio_frame).time_base = AVRational {
|
||||
num: 1,
|
||||
den: self.audio_sample_rate as _,
|
||||
};
|
||||
av_channel_layout_default(&mut (*audio_frame).ch_layout, self.audio_channels as _);
|
||||
av_frame_get_buffer(audio_frame, 0);
|
||||
|
||||
// Generate sine wave samples
|
||||
let data = (*audio_frame).data[0] as *mut f32;
|
||||
for i in 0..self.audio_frame_size {
|
||||
let sample_time =
|
||||
(self.audio_samples + i as u64) as f32 / self.audio_sample_rate as f32;
|
||||
let sample_value =
|
||||
(2.0 * std::f32::consts::PI * FREQUENCY * sample_time).sin() * 0.5;
|
||||
*data.add(i as _) = sample_value;
|
||||
}
|
||||
|
||||
self.audio_samples += self.audio_frame_size as u64;
|
||||
return Ok(audio_frame);
|
||||
}
|
||||
|
||||
Ok(ptr::null_mut())
|
||||
}
|
||||
|
||||
/// Return the next frame for encoding (blocking)
|
||||
pub unsafe fn next(&mut self) -> Result<*mut AVFrame> {
|
||||
// set start time to now if this is the first call to next()
|
||||
if self.frame_idx == 0 {
|
||||
self.start = Instant::now();
|
||||
}
|
||||
|
||||
// try to get audio frames before video frames (non-blocking)
|
||||
let audio_frame = self.generate_audio_frame()?;
|
||||
if !audio_frame.is_null() {
|
||||
return Ok(audio_frame);
|
||||
}
|
||||
|
||||
// auto-init frame
|
||||
if self.next_frame.is_null() {
|
||||
self.begin()?;
|
||||
}
|
||||
|
||||
let stream_time = Duration::from_secs_f64(self.frame_idx as f64 / self.fps as f64);
|
||||
let real_time = Instant::now().duration_since(self.start);
|
||||
let wait_time = if stream_time > real_time {
|
||||
stream_time - real_time
|
||||
} else {
|
||||
Duration::new(0, 0)
|
||||
};
|
||||
if !wait_time.is_zero() && wait_time.as_secs_f32() > 1f32 / self.fps {
|
||||
std::thread::sleep(wait_time);
|
||||
}
|
||||
|
||||
// convert to output pixel format, or just return internal frame if it matches output
|
||||
if self.video_sample_fmt != transmute((*self.next_frame).format) {
|
||||
let out_frame = self.scaler.process_frame(
|
||||
self.next_frame,
|
||||
self.width,
|
||||
self.height,
|
||||
self.video_sample_fmt,
|
||||
)?;
|
||||
av_frame_free(&mut self.next_frame);
|
||||
self.next_frame = ptr::null_mut();
|
||||
self.frame_idx += 1;
|
||||
Ok(out_frame)
|
||||
} else {
|
||||
let ret = self.next_frame;
|
||||
self.next_frame = ptr::null_mut();
|
||||
self.frame_idx += 1;
|
||||
Ok(ret)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
use ffmpeg_rs_raw::ffmpeg_sys_the_third::AVPixelFormat::AV_PIX_FMT_YUV420P;
|
||||
|
||||
#[test]
|
||||
fn test_frame_timing_synchronization() {
|
||||
unsafe {
|
||||
let fps = 30.0;
|
||||
let sample_rate = 44100;
|
||||
let frame_size = 1024;
|
||||
let channels = 2;
|
||||
|
||||
let mut gen = FrameGenerator::new(
|
||||
fps,
|
||||
1280,
|
||||
720,
|
||||
AV_PIX_FMT_YUV420P,
|
||||
sample_rate,
|
||||
frame_size,
|
||||
channels,
|
||||
)
|
||||
.unwrap();
|
||||
|
||||
let samples_per_frame = sample_rate as f64 / fps as f64; // Expected: 1470 samples per frame
|
||||
println!("Expected samples per video frame: {:.2}", samples_per_frame);
|
||||
|
||||
let mut audio_frames = 0;
|
||||
let mut video_frames = 0;
|
||||
let mut total_audio_samples = 0;
|
||||
|
||||
// Generate frames for 2 seconds (60 video frames at 30fps)
|
||||
for i in 0..120 {
|
||||
let mut frame = gen.next().unwrap();
|
||||
|
||||
if (*frame).sample_rate > 0 {
|
||||
// Audio frame
|
||||
audio_frames += 1;
|
||||
total_audio_samples += (*frame).nb_samples as u64;
|
||||
println!(
|
||||
"Frame {}: AUDIO - PTS: {}, samples: {}, total_samples: {}",
|
||||
i,
|
||||
(*frame).pts,
|
||||
(*frame).nb_samples,
|
||||
total_audio_samples
|
||||
);
|
||||
} else {
|
||||
// Video frame
|
||||
video_frames += 1;
|
||||
let expected_audio_samples = (video_frames as f64 * samples_per_frame) as u64;
|
||||
let audio_deficit = if total_audio_samples >= expected_audio_samples {
|
||||
0
|
||||
} else {
|
||||
expected_audio_samples - total_audio_samples
|
||||
};
|
||||
|
||||
println!("Frame {}: VIDEO - PTS: {}, frame_idx: {}, expected_audio: {}, actual_audio: {}, deficit: {}",
|
||||
i, (*frame).pts, video_frames, expected_audio_samples, total_audio_samples, audio_deficit);
|
||||
|
||||
// Verify we have enough audio for this video frame
|
||||
assert!(
|
||||
total_audio_samples >= expected_audio_samples,
|
||||
"Video frame {} needs {} audio samples but only have {}",
|
||||
video_frames,
|
||||
expected_audio_samples,
|
||||
total_audio_samples
|
||||
);
|
||||
}
|
||||
|
||||
av_frame_free(&mut frame);
|
||||
}
|
||||
|
||||
println!("\nSummary:");
|
||||
println!("Video frames: {}", video_frames);
|
||||
println!("Audio frames: {}", audio_frames);
|
||||
println!("Total audio samples: {}", total_audio_samples);
|
||||
println!(
|
||||
"Expected audio samples for {} video frames: {:.2}",
|
||||
video_frames,
|
||||
video_frames as f64 * samples_per_frame
|
||||
);
|
||||
|
||||
// Verify the ratio is correct
|
||||
let expected_total_audio = video_frames as f64 * samples_per_frame;
|
||||
let sample_accuracy = (total_audio_samples as f64 - expected_total_audio).abs();
|
||||
println!("Sample accuracy (difference): {:.2}", sample_accuracy);
|
||||
|
||||
// Allow for some tolerance due to frame size constraints
|
||||
assert!(
|
||||
sample_accuracy < frame_size as f64,
|
||||
"Audio sample count too far from expected: got {}, expected {:.2}, diff {:.2}",
|
||||
total_audio_samples,
|
||||
expected_total_audio,
|
||||
sample_accuracy
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_pts_progression() {
|
||||
unsafe {
|
||||
let fps = 30.0;
|
||||
let sample_rate = 44100;
|
||||
|
||||
let mut gen =
|
||||
FrameGenerator::new(fps, 1280, 720, AV_PIX_FMT_YUV420P, sample_rate, 1024, 2)
|
||||
.unwrap();
|
||||
|
||||
let mut last_audio_pts = -1i64;
|
||||
let mut last_video_pts = -1i64;
|
||||
let mut audio_pts_gaps = Vec::new();
|
||||
let mut video_pts_gaps = Vec::new();
|
||||
|
||||
// Generate 60 frames to test PTS progression
|
||||
for _ in 0..60 {
|
||||
let mut frame = gen.next().unwrap();
|
||||
|
||||
if (*frame).sample_rate > 0 {
|
||||
// Audio frame - check PTS progression
|
||||
if last_audio_pts >= 0 {
|
||||
let gap = (*frame).pts - last_audio_pts;
|
||||
audio_pts_gaps.push(gap);
|
||||
println!("Audio PTS gap: {}", gap);
|
||||
}
|
||||
last_audio_pts = (*frame).pts;
|
||||
} else {
|
||||
// Video frame - check PTS progression
|
||||
if last_video_pts >= 0 {
|
||||
let gap = (*frame).pts - last_video_pts;
|
||||
video_pts_gaps.push(gap);
|
||||
println!("Video PTS gap: {}", gap);
|
||||
}
|
||||
last_video_pts = (*frame).pts;
|
||||
}
|
||||
|
||||
av_frame_free(&mut frame);
|
||||
}
|
||||
|
||||
// Verify audio PTS gaps are consistent (should be 1024 samples)
|
||||
for gap in &audio_pts_gaps {
|
||||
assert_eq!(
|
||||
*gap, 1024,
|
||||
"Audio PTS should increment by frame_size (1024)"
|
||||
);
|
||||
}
|
||||
|
||||
// Verify video PTS gaps are consistent (should be 1 frame)
|
||||
for gap in &video_pts_gaps {
|
||||
assert_eq!(*gap, 1, "Video PTS should increment by 1 frame");
|
||||
}
|
||||
|
||||
println!("PTS progression test passed - all gaps are consistent");
|
||||
}
|
||||
}
|
||||
}
|
||||
@ -12,7 +12,6 @@ pub mod rtmp;
|
||||
#[cfg(feature = "srt")]
|
||||
pub mod srt;
|
||||
pub mod tcp;
|
||||
#[cfg(feature = "test-pattern")]
|
||||
pub mod test;
|
||||
|
||||
#[derive(Clone, Debug, Serialize, Deserialize)]
|
||||
|
||||
@ -15,9 +15,10 @@ pub async fn listen(out_dir: String, addr: String, overseer: Arc<dyn Overseer>)
|
||||
ip_addr: ip.to_string(),
|
||||
endpoint: addr.clone(),
|
||||
app_name: "".to_string(),
|
||||
key: "no-key-tcp".to_string(),
|
||||
key: "test".to_string(),
|
||||
};
|
||||
let socket = socket.into_std()?;
|
||||
socket.set_nonblocking(false)?;
|
||||
spawn_pipeline(
|
||||
Handle::current(),
|
||||
info,
|
||||
|
||||
@ -1,23 +1,16 @@
|
||||
use crate::generator::FrameGenerator;
|
||||
use crate::ingress::{spawn_pipeline, ConnectionInfo};
|
||||
use crate::overseer::Overseer;
|
||||
use anyhow::Result;
|
||||
use ffmpeg_rs_raw::ffmpeg_sys_the_third::AVColorSpace::AVCOL_SPC_RGB;
|
||||
use ffmpeg_rs_raw::ffmpeg_sys_the_third::AVPictureType::AV_PICTURE_TYPE_NONE;
|
||||
use ffmpeg_rs_raw::ffmpeg_sys_the_third::AVPixelFormat::{AV_PIX_FMT_RGBA, AV_PIX_FMT_YUV420P};
|
||||
use ffmpeg_rs_raw::ffmpeg_sys_the_third::AVPixelFormat::AV_PIX_FMT_YUV420P;
|
||||
use ffmpeg_rs_raw::ffmpeg_sys_the_third::AVSampleFormat::AV_SAMPLE_FMT_FLTP;
|
||||
use ffmpeg_rs_raw::ffmpeg_sys_the_third::{
|
||||
av_frame_alloc, av_frame_free, av_frame_get_buffer, av_packet_free, AVRational,
|
||||
AV_PROFILE_H264_MAIN,
|
||||
};
|
||||
use ffmpeg_rs_raw::{Encoder, Muxer, Scaler};
|
||||
use fontdue::layout::{CoordinateSystem, Layout, TextStyle};
|
||||
use fontdue::Font;
|
||||
use ffmpeg_rs_raw::ffmpeg_sys_the_third::{av_frame_free, av_packet_free, AV_PROFILE_H264_MAIN};
|
||||
use ffmpeg_rs_raw::{Encoder, Muxer};
|
||||
use log::info;
|
||||
use ringbuf::traits::{Observer, Split};
|
||||
use ringbuf::{HeapCons, HeapRb};
|
||||
use std::io::Read;
|
||||
use std::sync::Arc;
|
||||
use std::time::{Duration, Instant};
|
||||
use tiny_skia::Pixmap;
|
||||
use tokio::runtime::Handle;
|
||||
|
||||
@ -42,33 +35,31 @@ pub async fn listen(out_dir: String, overseer: Arc<dyn Overseer>) -> Result<()>
|
||||
}
|
||||
|
||||
struct TestPatternSrc {
|
||||
gen: FrameGenerator,
|
||||
video_encoder: Encoder,
|
||||
audio_encoder: Encoder,
|
||||
scaler: Scaler,
|
||||
muxer: Muxer,
|
||||
background: Pixmap,
|
||||
font: [Font; 1],
|
||||
frame_no: u64,
|
||||
audio_sample_no: u64,
|
||||
start: Instant,
|
||||
muxer: Muxer,
|
||||
reader: HeapCons<u8>,
|
||||
}
|
||||
|
||||
unsafe impl Send for TestPatternSrc {}
|
||||
|
||||
const VIDEO_FPS: f32 = 30.0;
|
||||
const VIDEO_WIDTH: u16 = 1280;
|
||||
const VIDEO_HEIGHT: u16 = 720;
|
||||
const SAMPLE_RATE: u32 = 44100;
|
||||
|
||||
impl TestPatternSrc {
|
||||
pub fn new() -> Result<Self> {
|
||||
let scaler = Scaler::new();
|
||||
let video_encoder = unsafe {
|
||||
Encoder::new_with_name("libx264")?
|
||||
.with_stream_index(0)
|
||||
.with_framerate(VIDEO_FPS)?
|
||||
.with_bitrate(1_000_000)
|
||||
.with_pix_fmt(AV_PIX_FMT_YUV420P)
|
||||
.with_width(1280)
|
||||
.with_height(720)
|
||||
.with_width(VIDEO_WIDTH as _)
|
||||
.with_height(VIDEO_HEIGHT as _)
|
||||
.with_level(51)
|
||||
.with_profile(AV_PROFILE_H264_MAIN)
|
||||
.open(None)?
|
||||
@ -80,22 +71,20 @@ impl TestPatternSrc {
|
||||
.with_default_channel_layout(1)
|
||||
.with_bitrate(128_000)
|
||||
.with_sample_format(AV_SAMPLE_FMT_FLTP)
|
||||
.with_sample_rate(44100)?
|
||||
.with_sample_rate(SAMPLE_RATE as _)?
|
||||
.open(None)?
|
||||
};
|
||||
|
||||
let svg_data = include_bytes!("../../test.svg");
|
||||
let tree = usvg::Tree::from_data(svg_data, &Default::default())?;
|
||||
let mut pixmap = Pixmap::new(1280, 720).unwrap();
|
||||
|
||||
let mut pixmap = Pixmap::new(VIDEO_WIDTH as _, VIDEO_HEIGHT as _).unwrap();
|
||||
let render_ts = tiny_skia::Transform::from_scale(
|
||||
pixmap.width() as f32 / tree.size().width(),
|
||||
pixmap.height() as f32 / tree.size().height(),
|
||||
);
|
||||
resvg::render(&tree, render_ts, &mut pixmap.as_mut());
|
||||
|
||||
let font = include_bytes!("../../SourceCodePro-Regular.ttf") as &[u8];
|
||||
let font = Font::from_bytes(font, Default::default()).unwrap();
|
||||
|
||||
let buf = HeapRb::new(1024 * 1024);
|
||||
let (writer, reader) = buf.split();
|
||||
|
||||
@ -109,140 +98,51 @@ impl TestPatternSrc {
|
||||
m
|
||||
};
|
||||
|
||||
let frame_size = unsafe { (*audio_encoder.codec_context()).frame_size as _ };
|
||||
Ok(Self {
|
||||
gen: FrameGenerator::new(
|
||||
VIDEO_FPS,
|
||||
VIDEO_WIDTH,
|
||||
VIDEO_HEIGHT,
|
||||
AV_PIX_FMT_YUV420P,
|
||||
SAMPLE_RATE,
|
||||
frame_size,
|
||||
1,
|
||||
)?,
|
||||
video_encoder,
|
||||
audio_encoder,
|
||||
scaler,
|
||||
muxer,
|
||||
background: pixmap,
|
||||
font: [font],
|
||||
frame_no: 0,
|
||||
audio_sample_no: 0,
|
||||
start: Instant::now(),
|
||||
reader,
|
||||
})
|
||||
}
|
||||
|
||||
pub unsafe fn next_pkt(&mut self) -> Result<()> {
|
||||
let stream_time = Duration::from_secs_f64(self.frame_no as f64 / VIDEO_FPS as f64);
|
||||
let real_time = Instant::now().duration_since(self.start);
|
||||
let wait_time = if stream_time > real_time {
|
||||
stream_time - real_time
|
||||
} else {
|
||||
Duration::new(0, 0)
|
||||
};
|
||||
if !wait_time.is_zero() && wait_time.as_secs_f32() > 1f32 / VIDEO_FPS {
|
||||
std::thread::sleep(wait_time);
|
||||
self.gen.begin()?;
|
||||
self.gen.copy_frame_data(self.background.data())?;
|
||||
self.gen
|
||||
.write_text(&format!("frame={}", self.gen.frame_no()), 40.0, 5.0, 5.0)?;
|
||||
|
||||
let mut frame = self.gen.next()?;
|
||||
if frame.is_null() {
|
||||
return Ok(());
|
||||
}
|
||||
|
||||
let mut src_frame = unsafe {
|
||||
let src_frame = av_frame_alloc();
|
||||
|
||||
(*src_frame).width = 1280;
|
||||
(*src_frame).height = 720;
|
||||
(*src_frame).pict_type = AV_PICTURE_TYPE_NONE;
|
||||
(*src_frame).key_frame = 1;
|
||||
(*src_frame).colorspace = AVCOL_SPC_RGB;
|
||||
(*src_frame).format = AV_PIX_FMT_RGBA as _;
|
||||
(*src_frame).pts = self.frame_no as i64;
|
||||
(*src_frame).duration = 1;
|
||||
av_frame_get_buffer(src_frame, 0);
|
||||
|
||||
self.background
|
||||
.data()
|
||||
.as_ptr()
|
||||
.copy_to((*src_frame).data[0] as *mut _, 1280 * 720 * 4);
|
||||
src_frame
|
||||
};
|
||||
let mut layout = Layout::new(CoordinateSystem::PositiveYDown);
|
||||
layout.clear();
|
||||
layout.append(
|
||||
&self.font,
|
||||
&TextStyle::new(&format!("frame={}", self.frame_no), 40.0, 0),
|
||||
);
|
||||
for g in layout.glyphs() {
|
||||
let (metrics, bitmap) = self.font[0].rasterize_config_subpixel(g.key);
|
||||
for y in 0..metrics.height {
|
||||
for x in 0..metrics.width {
|
||||
let dst_x = x + g.x as usize;
|
||||
let dst_y = y + g.y as usize;
|
||||
let offset_src = (x + y * metrics.width) * 3;
|
||||
unsafe {
|
||||
let offset_dst = 4 * dst_x + dst_y * (*src_frame).linesize[0] as usize;
|
||||
let pixel_dst = (*src_frame).data[0].add(offset_dst);
|
||||
*pixel_dst.offset(0) = bitmap[offset_src];
|
||||
*pixel_dst.offset(1) = bitmap[offset_src + 1];
|
||||
*pixel_dst.offset(2) = bitmap[offset_src + 2];
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// scale/encode video
|
||||
let mut frame = self
|
||||
.scaler
|
||||
.process_frame(src_frame, 1280, 720, AV_PIX_FMT_YUV420P)?;
|
||||
for mut pkt in self.video_encoder.encode_frame(frame)? {
|
||||
self.muxer.write_packet(pkt)?;
|
||||
av_packet_free(&mut pkt);
|
||||
}
|
||||
av_frame_free(&mut frame);
|
||||
av_frame_free(&mut src_frame);
|
||||
|
||||
// Generate and encode audio (sine wave)
|
||||
self.generate_audio_frame()?;
|
||||
|
||||
self.frame_no += 1;
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Generate audio to stay synchronized with video frames
|
||||
unsafe fn generate_audio_frame(&mut self) -> Result<()> {
|
||||
const SAMPLE_RATE: f32 = 44100.0;
|
||||
const FREQUENCY: f32 = 440.0; // A4 note
|
||||
const SAMPLES_PER_FRAME: usize = 1024; // Fixed AAC frame size
|
||||
|
||||
// Calculate how many audio samples we should have by now
|
||||
// At 30fps, each video frame = 1/30 sec = 1470 audio samples at 44.1kHz
|
||||
let audio_samples_per_video_frame = (SAMPLE_RATE / VIDEO_FPS) as u64; // ~1470 samples
|
||||
let target_audio_samples = self.frame_no * audio_samples_per_video_frame;
|
||||
|
||||
// Generate audio frames to catch up to the target
|
||||
while self.audio_sample_no < target_audio_samples {
|
||||
let mut audio_frame = av_frame_alloc();
|
||||
(*audio_frame).format = AV_SAMPLE_FMT_FLTP as _;
|
||||
(*audio_frame).nb_samples = SAMPLES_PER_FRAME as _;
|
||||
(*audio_frame).ch_layout.nb_channels = 1;
|
||||
(*audio_frame).sample_rate = SAMPLE_RATE as _;
|
||||
(*audio_frame).pts = self.audio_sample_no as i64;
|
||||
(*audio_frame).duration = 1;
|
||||
(*audio_frame).time_base = AVRational {
|
||||
num: 1,
|
||||
den: SAMPLE_RATE as _,
|
||||
};
|
||||
|
||||
av_frame_get_buffer(audio_frame, 0);
|
||||
|
||||
// Generate sine wave samples
|
||||
let data = (*audio_frame).data[0] as *mut f32;
|
||||
for i in 0..SAMPLES_PER_FRAME {
|
||||
let sample_time = (self.audio_sample_no + i as u64) as f32 / SAMPLE_RATE;
|
||||
let sample_value =
|
||||
(2.0 * std::f32::consts::PI * FREQUENCY * sample_time).sin() * 0.5;
|
||||
*data.add(i) = sample_value;
|
||||
}
|
||||
|
||||
// Encode audio frame
|
||||
for mut pkt in self.audio_encoder.encode_frame(audio_frame)? {
|
||||
// if sample_rate is set this frame is audio
|
||||
if (*frame).sample_rate > 0 {
|
||||
for mut pkt in self.audio_encoder.encode_frame(frame)? {
|
||||
self.muxer.write_packet(pkt)?;
|
||||
av_packet_free(&mut pkt);
|
||||
}
|
||||
} else {
|
||||
for mut pkt in self.video_encoder.encode_frame(frame)? {
|
||||
self.muxer.write_packet(pkt)?;
|
||||
av_packet_free(&mut pkt);
|
||||
}
|
||||
|
||||
self.audio_sample_no += SAMPLES_PER_FRAME as u64;
|
||||
av_frame_free(&mut audio_frame);
|
||||
}
|
||||
|
||||
av_frame_free(&mut frame);
|
||||
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
|
||||
@ -5,3 +5,4 @@ pub mod overseer;
|
||||
pub mod pipeline;
|
||||
pub mod variant;
|
||||
pub mod viewer;
|
||||
mod generator;
|
||||
|
||||
@ -1,15 +1,15 @@
|
||||
use crate::egress::{EgressResult, EgressSegment};
|
||||
use crate::variant::{StreamMapping, VariantStream};
|
||||
use anyhow::{bail, Result};
|
||||
use anyhow::{bail, ensure, Result};
|
||||
use ffmpeg_rs_raw::ffmpeg_sys_the_third::AVCodecID::AV_CODEC_ID_H264;
|
||||
use ffmpeg_rs_raw::ffmpeg_sys_the_third::AVMediaType::AVMEDIA_TYPE_VIDEO;
|
||||
use ffmpeg_rs_raw::ffmpeg_sys_the_third::{
|
||||
av_free, av_opt_set, av_q2d, av_write_frame, avio_close, avio_flush, avio_open, AVPacket,
|
||||
AVStream, AVIO_FLAG_WRITE, AV_PKT_FLAG_KEY,
|
||||
AVStream, AVIO_FLAG_WRITE, AV_NOPTS_VALUE, AV_PKT_FLAG_KEY,
|
||||
};
|
||||
use ffmpeg_rs_raw::{cstr, Encoder, Muxer};
|
||||
use itertools::Itertools;
|
||||
use log::{info, warn};
|
||||
use log::{info, trace, warn};
|
||||
use m3u8_rs::MediaSegment;
|
||||
use std::collections::HashMap;
|
||||
use std::fmt::Display;
|
||||
@ -72,31 +72,37 @@ impl Display for HlsVariantStream {
|
||||
|
||||
pub struct HlsVariant {
|
||||
/// Name of this variant (720p)
|
||||
pub name: String,
|
||||
name: String,
|
||||
/// MPEG-TS muxer for this variant
|
||||
pub mux: Muxer,
|
||||
mux: Muxer,
|
||||
/// List of streams ids in this variant
|
||||
pub streams: Vec<HlsVariantStream>,
|
||||
streams: Vec<HlsVariantStream>,
|
||||
/// Segment length in seconds
|
||||
pub segment_length: f32,
|
||||
segment_length: f32,
|
||||
/// Total number of segments to store for this variant
|
||||
pub segment_window: Option<u16>,
|
||||
segment_window: Option<u16>,
|
||||
/// Current segment index
|
||||
pub idx: u64,
|
||||
/// Current segment start time in seconds (duration)
|
||||
pub pkt_start: f32,
|
||||
idx: u64,
|
||||
/// Output directory (base)
|
||||
pub out_dir: String,
|
||||
out_dir: String,
|
||||
/// List of segments to be included in the playlist
|
||||
pub segments: Vec<SegmentInfo>,
|
||||
segments: Vec<SegmentInfo>,
|
||||
/// Type of segments to create
|
||||
pub segment_type: SegmentType,
|
||||
segment_type: SegmentType,
|
||||
/// Ending presentation timestamp
|
||||
end_pts: i64,
|
||||
/// Current segment duration in seconds (precise accumulation)
|
||||
duration: f64,
|
||||
/// Number of packets written to current segment
|
||||
packets_written: u64,
|
||||
/// Reference stream used to track duration
|
||||
ref_stream_index: i32,
|
||||
}
|
||||
|
||||
struct SegmentInfo {
|
||||
pub index: u64,
|
||||
pub duration: f32,
|
||||
pub kind: SegmentType,
|
||||
index: u64,
|
||||
duration: f32,
|
||||
kind: SegmentType,
|
||||
}
|
||||
|
||||
impl SegmentInfo {
|
||||
@ -146,23 +152,35 @@ impl HlsVariant {
|
||||
.build()?
|
||||
};
|
||||
let mut streams = Vec::new();
|
||||
let mut ref_stream_index = -1;
|
||||
let mut has_video = false;
|
||||
|
||||
for (var, enc) in encoded_vars {
|
||||
match var {
|
||||
VariantStream::Video(v) => unsafe {
|
||||
let stream = mux.add_stream_encoder(enc)?;
|
||||
let stream_idx = (*stream).index as usize;
|
||||
streams.push(HlsVariantStream::Video {
|
||||
group,
|
||||
index: (*stream).index as usize,
|
||||
index: stream_idx,
|
||||
id: v.id(),
|
||||
})
|
||||
});
|
||||
has_video = true;
|
||||
if ref_stream_index == -1 {
|
||||
ref_stream_index = stream_idx as _;
|
||||
}
|
||||
},
|
||||
VariantStream::Audio(a) => unsafe {
|
||||
let stream = mux.add_stream_encoder(enc)?;
|
||||
let stream_idx = (*stream).index as usize;
|
||||
streams.push(HlsVariantStream::Audio {
|
||||
group,
|
||||
index: (*stream).index as usize,
|
||||
index: stream_idx,
|
||||
id: a.id(),
|
||||
})
|
||||
});
|
||||
if !has_video && ref_stream_index == -1 {
|
||||
ref_stream_index = stream_idx as _;
|
||||
}
|
||||
},
|
||||
VariantStream::Subtitle(s) => unsafe {
|
||||
let stream = mux.add_stream_encoder(enc)?;
|
||||
@ -175,6 +193,10 @@ impl HlsVariant {
|
||||
_ => bail!("unsupported variant stream"),
|
||||
}
|
||||
}
|
||||
ensure!(
|
||||
ref_stream_index != -1,
|
||||
"No reference stream found, cant create variant"
|
||||
);
|
||||
unsafe {
|
||||
mux.open(Some(opts))?;
|
||||
}
|
||||
@ -185,10 +207,13 @@ impl HlsVariant {
|
||||
mux,
|
||||
streams,
|
||||
idx: 1,
|
||||
pkt_start: 0.0,
|
||||
segments: Vec::new(), // Start with empty segments list
|
||||
out_dir: out_dir.to_string(),
|
||||
segment_type,
|
||||
end_pts: AV_NOPTS_VALUE,
|
||||
duration: 0.0,
|
||||
packets_written: 0,
|
||||
ref_stream_index,
|
||||
})
|
||||
}
|
||||
|
||||
@ -218,34 +243,54 @@ impl HlsVariant {
|
||||
self.process_packet(pkt)
|
||||
}
|
||||
|
||||
/// Process a single packet through the muxer
|
||||
/// Process a single packet through the muxer - FFmpeg-style implementation
|
||||
unsafe fn process_packet(&mut self, pkt: *mut AVPacket) -> Result<EgressResult> {
|
||||
let mut result = EgressResult::None;
|
||||
let pkt_stream = *(*self.mux.context())
|
||||
.streams
|
||||
.add((*pkt).stream_index as usize);
|
||||
|
||||
// Match FFmpeg's segmentation logic exactly
|
||||
let can_split = (*pkt).flags & AV_PKT_FLAG_KEY == AV_PKT_FLAG_KEY
|
||||
&& (*(*pkt_stream).codecpar).codec_type == AVMEDIA_TYPE_VIDEO;
|
||||
let mut result = EgressResult::None;
|
||||
let stream_type = (*(*pkt_stream).codecpar).codec_type;
|
||||
let mut can_split = stream_type == AVMEDIA_TYPE_VIDEO
|
||||
&& ((*pkt).flags & AV_PKT_FLAG_KEY == AV_PKT_FLAG_KEY);
|
||||
let mut is_ref_pkt =
|
||||
stream_type == AVMEDIA_TYPE_VIDEO && (*pkt_stream).index == self.ref_stream_index;
|
||||
|
||||
if can_split {
|
||||
let pkt_q = av_q2d((*pkt).time_base);
|
||||
let pkt_time = (*pkt).pts as f32 * pkt_q as f32;
|
||||
let relative_time = pkt_time - self.pkt_start;
|
||||
if (*pkt).pts == AV_NOPTS_VALUE {
|
||||
can_split = false;
|
||||
is_ref_pkt = false;
|
||||
}
|
||||
|
||||
// FFmpeg checks: pkt->pts - vs->end_pts > 0 to prevent zero duration
|
||||
// and av_compare_ts for target duration
|
||||
let has_positive_duration = relative_time > 0.0;
|
||||
let target_duration_reached = relative_time >= self.segment_length;
|
||||
// check if current packet is keyframe, flush current segment
|
||||
if self.packets_written > 0 && can_split {
|
||||
trace!(
|
||||
"Segmentation check: pts={}, duration={:.3}, timebase={}/{}, target={:.3}",
|
||||
(*pkt).pts,
|
||||
self.duration,
|
||||
(*pkt).time_base.num,
|
||||
(*pkt).time_base.den,
|
||||
self.segment_length
|
||||
);
|
||||
|
||||
if has_positive_duration && target_duration_reached {
|
||||
result = self.split_next_seg(pkt_time)?;
|
||||
if self.duration >= self.segment_length as f64 {
|
||||
result = self.split_next_seg()?;
|
||||
}
|
||||
}
|
||||
|
||||
// Write packet directly like FFmpeg's ff_write_chained
|
||||
// track duration from pts
|
||||
if is_ref_pkt {
|
||||
if self.end_pts == AV_NOPTS_VALUE {
|
||||
self.end_pts = (*pkt).pts;
|
||||
}
|
||||
let pts_diff = (*pkt).pts - self.end_pts;
|
||||
if pts_diff > 0 {
|
||||
self.duration += pts_diff as f64 * av_q2d((*pkt).time_base);
|
||||
}
|
||||
self.end_pts = (*pkt).pts;
|
||||
}
|
||||
|
||||
self.mux.write_packet(pkt)?;
|
||||
self.packets_written += 1;
|
||||
Ok(result)
|
||||
}
|
||||
|
||||
@ -254,7 +299,7 @@ impl HlsVariant {
|
||||
}
|
||||
|
||||
/// Reset the muxer state and start the next segment
|
||||
unsafe fn split_next_seg(&mut self, pkt_time: f32) -> Result<EgressResult> {
|
||||
unsafe fn split_next_seg(&mut self) -> Result<EgressResult> {
|
||||
let completed_segment_idx = self.idx;
|
||||
self.idx += 1;
|
||||
|
||||
@ -282,7 +327,6 @@ impl HlsVariant {
|
||||
0,
|
||||
);
|
||||
|
||||
let duration = pkt_time - self.pkt_start;
|
||||
// Log the completed segment (previous index), not the next one
|
||||
let completed_seg_path = Self::map_segment_path(
|
||||
&self.out_dir,
|
||||
@ -296,13 +340,14 @@ impl HlsVariant {
|
||||
.map(|m| m.len())
|
||||
.unwrap_or(0);
|
||||
info!(
|
||||
"Finished segment {} [{:.3}s, {} bytes]",
|
||||
"Finished segment {} [{:.3}s, {:.2} kB, {} pkts]",
|
||||
completed_segment_path
|
||||
.file_name()
|
||||
.unwrap_or_default()
|
||||
.to_string_lossy(),
|
||||
duration,
|
||||
segment_size
|
||||
self.duration,
|
||||
segment_size as f32 / 1024f32,
|
||||
self.packets_written
|
||||
);
|
||||
|
||||
let video_var_id = self
|
||||
@ -332,14 +377,17 @@ impl HlsVariant {
|
||||
let created = EgressSegment {
|
||||
variant: video_var_id,
|
||||
idx: completed_segment_idx,
|
||||
duration,
|
||||
duration: self.duration as f32,
|
||||
path: completed_segment_path,
|
||||
};
|
||||
|
||||
if let Err(e) = self.push_segment(completed_segment_idx, duration) {
|
||||
if let Err(e) = self.push_segment(completed_segment_idx, self.duration as f32) {
|
||||
warn!("Failed to update playlist: {}", e);
|
||||
}
|
||||
self.pkt_start = pkt_time;
|
||||
|
||||
self.packets_written = 0;
|
||||
self.duration = 0.0;
|
||||
|
||||
Ok(EgressResult::Segments {
|
||||
created: vec![created],
|
||||
deleted,
|
||||
@ -381,6 +429,7 @@ impl HlsVariant {
|
||||
e
|
||||
);
|
||||
}
|
||||
info!("Removed segment file: {}", seg_path.display());
|
||||
ret.push(seg);
|
||||
}
|
||||
}
|
||||
|
||||
@ -8,12 +8,6 @@ use std::cmp::PartialEq;
|
||||
use std::path::PathBuf;
|
||||
use uuid::Uuid;
|
||||
|
||||
#[cfg(feature = "local-overseer")]
|
||||
mod local;
|
||||
|
||||
#[cfg(feature = "webhook-overseer")]
|
||||
mod webhook;
|
||||
|
||||
/// A copy of [ffmpeg_rs_raw::DemuxerInfo] without internal ptr
|
||||
#[derive(PartialEq, Clone)]
|
||||
pub struct IngressInfo {
|
||||
@ -32,6 +26,7 @@ pub struct IngressStream {
|
||||
pub height: usize,
|
||||
pub fps: f32,
|
||||
pub sample_rate: usize,
|
||||
pub channels: u8,
|
||||
pub language: String,
|
||||
}
|
||||
|
||||
|
||||
@ -7,7 +7,6 @@ use serde::{Deserialize, Serialize};
|
||||
use uuid::Uuid;
|
||||
|
||||
pub mod runner;
|
||||
pub mod placeholder;
|
||||
|
||||
#[derive(Clone, Debug, Serialize, Deserialize)]
|
||||
pub enum EgressType {
|
||||
@ -41,7 +40,7 @@ impl Display for EgressType {
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug, Serialize, Deserialize, Default)]
|
||||
#[derive(Clone)]
|
||||
pub struct PipelineConfig {
|
||||
pub id: Uuid,
|
||||
/// Transcoded/Copied stream config
|
||||
@ -49,7 +48,11 @@ pub struct PipelineConfig {
|
||||
/// Output muxers
|
||||
pub egress: Vec<EgressType>,
|
||||
/// Source stream information for placeholder generation
|
||||
pub ingress_info: Option<IngressInfo>,
|
||||
pub ingress_info: IngressInfo,
|
||||
/// Primary source video stream
|
||||
pub video_src: usize,
|
||||
/// Primary audio source stream
|
||||
pub audio_src: Option<usize>,
|
||||
}
|
||||
|
||||
impl Display for PipelineConfig {
|
||||
|
||||
@ -1,188 +0,0 @@
|
||||
use anyhow::{bail, Result};
|
||||
use crate::variant::video::VideoVariant;
|
||||
use crate::variant::audio::AudioVariant;
|
||||
use crate::overseer::{IngressStream, IngressStreamType};
|
||||
use ffmpeg_rs_raw::ffmpeg_sys_the_third::{
|
||||
av_frame_alloc, av_frame_get_buffer, av_frame_free, av_get_sample_fmt, AVFrame,
|
||||
AVPixelFormat, AVSampleFormat
|
||||
};
|
||||
use std::ffi::CString;
|
||||
|
||||
/// Placeholder frame generator for idle mode when stream disconnects
|
||||
pub struct PlaceholderGenerator;
|
||||
|
||||
impl PlaceholderGenerator {
|
||||
/// Generate a placeholder video frame based on ingress stream info
|
||||
pub unsafe fn generate_video_frame_from_stream(
|
||||
stream: &IngressStream,
|
||||
stream_time_base: (i32, i32),
|
||||
frame_index: u64
|
||||
) -> Result<*mut AVFrame> {
|
||||
let frame = av_frame_alloc();
|
||||
if frame.is_null() {
|
||||
bail!("Failed to allocate placeholder video frame");
|
||||
}
|
||||
|
||||
(*frame).format = AVPixelFormat::AV_PIX_FMT_YUV420P as i32;
|
||||
(*frame).width = stream.width as i32;
|
||||
(*frame).height = stream.height as i32;
|
||||
(*frame).time_base.num = stream_time_base.0;
|
||||
(*frame).time_base.den = stream_time_base.1;
|
||||
|
||||
// Set PTS based on frame rate and total frame index
|
||||
let fps = if stream.fps > 0.0 { stream.fps } else { 30.0 };
|
||||
let time_base_f64 = stream_time_base.0 as f64 / stream_time_base.1 as f64;
|
||||
(*frame).pts = (frame_index as f64 / fps / time_base_f64) as i64;
|
||||
|
||||
if av_frame_get_buffer(frame, 0) < 0 {
|
||||
av_frame_free(&mut frame);
|
||||
bail!("Failed to allocate buffer for placeholder video frame");
|
||||
}
|
||||
|
||||
// Fill with black (Y=16, U=V=128 for limited range YUV420P)
|
||||
let y_size = ((*frame).width * (*frame).height) as usize;
|
||||
let uv_size = y_size / 4;
|
||||
|
||||
if !(*frame).data[0].is_null() {
|
||||
std::ptr::write_bytes((*frame).data[0], 16, y_size);
|
||||
}
|
||||
if !(*frame).data[1].is_null() {
|
||||
std::ptr::write_bytes((*frame).data[1], 128, uv_size);
|
||||
}
|
||||
if !(*frame).data[2].is_null() {
|
||||
std::ptr::write_bytes((*frame).data[2], 128, uv_size);
|
||||
}
|
||||
|
||||
Ok(frame)
|
||||
}
|
||||
|
||||
/// Generate a placeholder audio frame based on ingress stream info
|
||||
pub unsafe fn generate_audio_frame_from_stream(
|
||||
stream: &IngressStream,
|
||||
stream_time_base: (i32, i32),
|
||||
frame_index: u64,
|
||||
sample_fmt: &str,
|
||||
channels: u32
|
||||
) -> Result<*mut AVFrame> {
|
||||
let frame = av_frame_alloc();
|
||||
if frame.is_null() {
|
||||
bail!("Failed to allocate placeholder audio frame");
|
||||
}
|
||||
|
||||
// Use the provided sample format
|
||||
let sample_fmt_cstr = CString::new(sample_fmt)
|
||||
.map_err(|_| anyhow::anyhow!("Invalid sample format string"))?;
|
||||
let sample_fmt_int = av_get_sample_fmt(sample_fmt_cstr.as_ptr());
|
||||
(*frame).format = sample_fmt_int;
|
||||
(*frame).channels = channels as i32;
|
||||
(*frame).sample_rate = stream.sample_rate as i32;
|
||||
(*frame).nb_samples = 1024; // Standard audio frame size
|
||||
(*frame).time_base.num = stream_time_base.0;
|
||||
(*frame).time_base.den = stream_time_base.1;
|
||||
|
||||
// Set PTS based on sample rate and frame index
|
||||
let samples_per_second = stream.sample_rate as f64;
|
||||
let time_base_f64 = stream_time_base.0 as f64 / stream_time_base.1 as f64;
|
||||
(*frame).pts = ((frame_index * 1024) as f64 / samples_per_second / time_base_f64) as i64;
|
||||
|
||||
if av_frame_get_buffer(frame, 0) < 0 {
|
||||
av_frame_free(&mut frame);
|
||||
bail!("Failed to allocate buffer for placeholder audio frame");
|
||||
}
|
||||
|
||||
// Fill with silence (zeros)
|
||||
for i in 0..8 {
|
||||
if !(*frame).data[i].is_null() && (*frame).linesize[i] > 0 {
|
||||
std::ptr::write_bytes((*frame).data[i], 0, (*frame).linesize[i] as usize);
|
||||
}
|
||||
}
|
||||
|
||||
Ok(frame)
|
||||
}
|
||||
|
||||
/// Generate a placeholder black video frame
|
||||
pub unsafe fn generate_video_frame(
|
||||
variant: &VideoVariant,
|
||||
stream_time_base: (i32, i32),
|
||||
frame_index: u64
|
||||
) -> Result<*mut AVFrame> {
|
||||
let frame = av_frame_alloc();
|
||||
if frame.is_null() {
|
||||
bail!("Failed to allocate placeholder video frame");
|
||||
}
|
||||
|
||||
(*frame).format = AVPixelFormat::AV_PIX_FMT_YUV420P as i32;
|
||||
(*frame).width = variant.width as i32;
|
||||
(*frame).height = variant.height as i32;
|
||||
(*frame).time_base.num = stream_time_base.0;
|
||||
(*frame).time_base.den = stream_time_base.1;
|
||||
|
||||
// Set PTS based on frame rate and total frame index
|
||||
let fps = if variant.fps > 0.0 { variant.fps } else { 30.0 };
|
||||
let time_base_f64 = stream_time_base.0 as f64 / stream_time_base.1 as f64;
|
||||
(*frame).pts = (frame_index as f64 / fps / time_base_f64) as i64;
|
||||
|
||||
if av_frame_get_buffer(frame, 0) < 0 {
|
||||
av_frame_free(&mut frame);
|
||||
bail!("Failed to allocate buffer for placeholder video frame");
|
||||
}
|
||||
|
||||
// Fill with black (Y=16, U=V=128 for limited range YUV420P)
|
||||
let y_size = ((*frame).width * (*frame).height) as usize;
|
||||
let uv_size = y_size / 4;
|
||||
|
||||
if !(*frame).data[0].is_null() {
|
||||
std::ptr::write_bytes((*frame).data[0], 16, y_size);
|
||||
}
|
||||
if !(*frame).data[1].is_null() {
|
||||
std::ptr::write_bytes((*frame).data[1], 128, uv_size);
|
||||
}
|
||||
if !(*frame).data[2].is_null() {
|
||||
std::ptr::write_bytes((*frame).data[2], 128, uv_size);
|
||||
}
|
||||
|
||||
Ok(frame)
|
||||
}
|
||||
|
||||
/// Generate a placeholder silent audio frame
|
||||
pub unsafe fn generate_audio_frame(
|
||||
variant: &AudioVariant,
|
||||
stream_time_base: (i32, i32),
|
||||
frame_index: u64
|
||||
) -> Result<*mut AVFrame> {
|
||||
let frame = av_frame_alloc();
|
||||
if frame.is_null() {
|
||||
bail!("Failed to allocate placeholder audio frame");
|
||||
}
|
||||
|
||||
// Use the sample format from the variant configuration
|
||||
let sample_fmt_cstr = CString::new(variant.sample_fmt.as_str())
|
||||
.map_err(|_| anyhow::anyhow!("Invalid sample format string"))?;
|
||||
let sample_fmt_int = av_get_sample_fmt(sample_fmt_cstr.as_ptr());
|
||||
(*frame).format = sample_fmt_int;
|
||||
(*frame).channels = variant.channels as i32;
|
||||
(*frame).sample_rate = variant.sample_rate as i32;
|
||||
(*frame).nb_samples = 1024; // Standard audio frame size
|
||||
(*frame).time_base.num = stream_time_base.0;
|
||||
(*frame).time_base.den = stream_time_base.1;
|
||||
|
||||
// Set PTS based on sample rate and frame index
|
||||
let samples_per_second = variant.sample_rate as f64;
|
||||
let time_base_f64 = stream_time_base.0 as f64 / stream_time_base.1 as f64;
|
||||
(*frame).pts = ((frame_index * 1024) as f64 / samples_per_second / time_base_f64) as i64;
|
||||
|
||||
if av_frame_get_buffer(frame, 0) < 0 {
|
||||
av_frame_free(&mut frame);
|
||||
bail!("Failed to allocate buffer for placeholder audio frame");
|
||||
}
|
||||
|
||||
// Fill with silence (zeros)
|
||||
for i in 0..8 {
|
||||
if !(*frame).data[i].is_null() && (*frame).linesize[i] > 0 {
|
||||
std::ptr::write_bytes((*frame).data[i], 0, (*frame).linesize[i] as usize);
|
||||
}
|
||||
}
|
||||
|
||||
Ok(frame)
|
||||
}
|
||||
}
|
||||
@ -10,18 +10,19 @@ use std::time::{Duration, Instant};
|
||||
use crate::egress::hls::HlsEgress;
|
||||
use crate::egress::recorder::RecorderEgress;
|
||||
use crate::egress::{Egress, EgressResult};
|
||||
use crate::generator::FrameGenerator;
|
||||
use crate::ingress::ConnectionInfo;
|
||||
use crate::mux::SegmentType;
|
||||
use crate::overseer::{IngressInfo, IngressStream, IngressStreamType, Overseer};
|
||||
use crate::pipeline::{EgressType, PipelineConfig};
|
||||
use crate::variant::{StreamMapping, VariantStream};
|
||||
use crate::pipeline::placeholder::PlaceholderGenerator;
|
||||
use anyhow::{bail, Result};
|
||||
use anyhow::{bail, Context, Result};
|
||||
use ffmpeg_rs_raw::ffmpeg_sys_the_third::AVCodecID::AV_CODEC_ID_WEBP;
|
||||
use ffmpeg_rs_raw::ffmpeg_sys_the_third::AVPictureType::AV_PICTURE_TYPE_NONE;
|
||||
use ffmpeg_rs_raw::ffmpeg_sys_the_third::AVPixelFormat::AV_PIX_FMT_YUV420P;
|
||||
use ffmpeg_rs_raw::ffmpeg_sys_the_third::{
|
||||
av_frame_free, av_get_sample_fmt, av_packet_free, av_q2d, av_rescale_q, AVMediaType,
|
||||
av_frame_free, av_get_sample_fmt, av_packet_free, av_rescale_q, AVFrame, AVMediaType, AVStream,
|
||||
AV_NOPTS_VALUE,
|
||||
};
|
||||
use ffmpeg_rs_raw::{
|
||||
cstr, get_frame_from_hw, AudioFifo, Decoder, Demuxer, DemuxerInfo, Encoder, Resample, Scaler,
|
||||
@ -32,15 +33,14 @@ use tokio::runtime::Handle;
|
||||
use uuid::Uuid;
|
||||
|
||||
/// Runner state for handling normal vs idle modes
|
||||
#[derive(Debug, Clone)]
|
||||
pub enum RunnerState {
|
||||
/// Normal operation - processing live stream
|
||||
Normal,
|
||||
/// Idle mode - generating placeholder content after disconnection
|
||||
Idle {
|
||||
start_time: Instant,
|
||||
variant_index: usize,
|
||||
last_frame_time: Option<Instant>,
|
||||
gen: FrameGenerator,
|
||||
},
|
||||
}
|
||||
|
||||
@ -129,142 +129,156 @@ impl PipelineRunner {
|
||||
})
|
||||
}
|
||||
|
||||
/// Process a single idle frame - generates one source frame and processes it through all variants
|
||||
unsafe fn process_single_idle_frame(&mut self, config: &PipelineConfig) -> Result<()> {
|
||||
use std::time::{Duration, Instant};
|
||||
|
||||
if config.variants.is_empty() {
|
||||
return Ok(());
|
||||
}
|
||||
/// process the frame in the pipeline
|
||||
unsafe fn process_frame(
|
||||
&mut self,
|
||||
config: &PipelineConfig,
|
||||
stream: *mut AVStream,
|
||||
frame: *mut AVFrame,
|
||||
) -> Result<Vec<EgressResult>> {
|
||||
// Copy frame from GPU if using hwaccel decoding
|
||||
let mut frame = get_frame_from_hw(frame)?;
|
||||
(*frame).time_base = (*stream).time_base;
|
||||
|
||||
// Extract timing info from current state
|
||||
let (mut last_frame_time, variant_index) = match &mut self.state {
|
||||
RunnerState::Idle { last_frame_time, variant_index, .. } => (last_frame_time, variant_index),
|
||||
_ => return Ok(()), // Only process in idle state
|
||||
};
|
||||
let p = (*stream).codecpar;
|
||||
if (*p).codec_type == AVMediaType::AVMEDIA_TYPE_VIDEO {
|
||||
// Conditionally generate thumbnails based on interval (0 = disabled)
|
||||
if self.thumb_interval > 0 && (self.frame_ctr % self.thumb_interval) == 0 {
|
||||
let thumb_start = Instant::now();
|
||||
let dst_pic = PathBuf::from(&self.out_dir)
|
||||
.join(config.id.to_string())
|
||||
.join("thumb.webp");
|
||||
{
|
||||
let mut sw = Scaler::new();
|
||||
let mut scaled_frame = sw.process_frame(
|
||||
frame,
|
||||
(*frame).width as _,
|
||||
(*frame).height as _,
|
||||
AV_PIX_FMT_YUV420P,
|
||||
)?;
|
||||
|
||||
// Time-based frame rate calculation
|
||||
let now = Instant::now();
|
||||
if let Some(last_time) = *last_frame_time {
|
||||
// Calculate target frame interval (assume 30fps for now)
|
||||
let target_interval = Duration::from_millis(33); // ~30fps
|
||||
let elapsed = now.duration_since(last_time);
|
||||
|
||||
if elapsed < target_interval {
|
||||
// Not time for next frame yet
|
||||
std::thread::sleep(target_interval - elapsed);
|
||||
let encoder = Encoder::new(AV_CODEC_ID_WEBP)?
|
||||
.with_height((*scaled_frame).height)
|
||||
.with_width((*scaled_frame).width)
|
||||
.with_pix_fmt(transmute((*scaled_frame).format))
|
||||
.open(None)?;
|
||||
|
||||
encoder.save_picture(scaled_frame, dst_pic.to_str().unwrap())?;
|
||||
av_frame_free(&mut scaled_frame);
|
||||
}
|
||||
|
||||
let thumb_duration = thumb_start.elapsed();
|
||||
info!(
|
||||
"Saved thumb ({:.2}ms) to: {}",
|
||||
thumb_duration.as_millis() as f32 / 1000.0,
|
||||
dst_pic.display(),
|
||||
);
|
||||
}
|
||||
}
|
||||
*last_frame_time = Some(Instant::now());
|
||||
|
||||
// Get source video stream info from stored ingress info
|
||||
let video_stream = config.ingress_info.as_ref()
|
||||
.and_then(|info| info.streams.iter().find(|s| matches!(s.stream_type, crate::overseer::IngressStreamType::Video)));
|
||||
|
||||
let mut egress_results = vec![];
|
||||
|
||||
// Generate one source frame and process it through all relevant variants
|
||||
if let Some(stream) = video_stream {
|
||||
// Generate a single source placeholder video frame based on original stream properties
|
||||
let fps = if stream.fps > 0.0 { stream.fps } else { 30.0 };
|
||||
let time_base = (1, fps as i32);
|
||||
let mut source_frame = PlaceholderGenerator::generate_video_frame_from_stream(stream, time_base, self.frame_ctr)?;
|
||||
|
||||
// Set the frame time_base
|
||||
(*source_frame).time_base.num = time_base.0;
|
||||
(*source_frame).time_base.den = time_base.1;
|
||||
|
||||
// Increment frame counter for all video processing
|
||||
self.frame_ctr += 1;
|
||||
|
||||
// Process this single frame through all video variants (like normal pipeline)
|
||||
for variant in &config.variants {
|
||||
if let VariantStream::Video(v) = variant {
|
||||
// Scale/encode the source frame for this variant
|
||||
if let Some(enc) = self.encoders.get_mut(&v.id()) {
|
||||
// Use scaler if needed for different resolutions
|
||||
let frame_to_encode = if v.width as i32 == (*source_frame).width &&
|
||||
v.height as i32 == (*source_frame).height {
|
||||
// Same resolution, use source frame directly
|
||||
source_frame
|
||||
} else {
|
||||
// Different resolution, need to scale
|
||||
if let Some(scaler) = self.scalers.get_mut(&v.id()) {
|
||||
scaler.process_frame(source_frame, v.width, v.height, AV_PIX_FMT_YUV420P)?
|
||||
} else {
|
||||
source_frame // Fallback to source frame
|
||||
}
|
||||
};
|
||||
|
||||
let packets = enc.encode_frame(frame_to_encode)?;
|
||||
for mut pkt in packets {
|
||||
for eg in self.egress.iter_mut() {
|
||||
let er = eg.process_pkt(pkt, &v.id())?;
|
||||
egress_results.push(er);
|
||||
}
|
||||
av_packet_free(&mut pkt);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
av_frame_free(&mut source_frame);
|
||||
}
|
||||
|
||||
// Generate and process audio frames separately (audio doesn't share like video)
|
||||
let audio_stream = config.ingress_info.as_ref()
|
||||
.and_then(|info| info.streams.iter().find(|s| matches!(s.stream_type, crate::overseer::IngressStreamType::Audio)));
|
||||
|
||||
for variant in &config.variants {
|
||||
if let VariantStream::Audio(a) = variant {
|
||||
let time_base = (1, a.sample_rate as i32);
|
||||
let mut frame = if let Some(stream) = audio_stream {
|
||||
// Use original stream properties for placeholder generation
|
||||
PlaceholderGenerator::generate_audio_frame_from_stream(stream, time_base, self.frame_ctr, &a.sample_fmt, a.channels)?
|
||||
} else {
|
||||
// Fallback to variant properties if no stream info available
|
||||
PlaceholderGenerator::generate_audio_frame(a, time_base, self.frame_ctr)?
|
||||
};
|
||||
|
||||
// Set the frame time_base
|
||||
(*frame).time_base.num = time_base.0;
|
||||
(*frame).time_base.den = time_base.1;
|
||||
|
||||
// Process through the encoding pipeline
|
||||
if let Some(enc) = self.encoders.get_mut(&a.id()) {
|
||||
let packets = enc.encode_frame(frame)?;
|
||||
for mut pkt in packets {
|
||||
for eg in self.egress.iter_mut() {
|
||||
let er = eg.process_pkt(pkt, &a.id())?;
|
||||
egress_results.push(er);
|
||||
}
|
||||
av_packet_free(&mut pkt);
|
||||
let mut egress_results = Vec::new();
|
||||
// Get the variants which want this pkt
|
||||
let pkt_vars = config
|
||||
.variants
|
||||
.iter()
|
||||
.filter(|v| v.src_index() == (*stream).index as usize);
|
||||
for var in pkt_vars {
|
||||
let enc = if let Some(enc) = self.encoders.get_mut(&var.id()) {
|
||||
enc
|
||||
} else {
|
||||
//warn!("Frame had nowhere to go in {} :/", var.id());
|
||||
continue;
|
||||
};
|
||||
|
||||
// scaling / resampling
|
||||
let mut new_frame = false;
|
||||
match var {
|
||||
VariantStream::Video(v) => {
|
||||
let mut frame = if let Some(s) = self.scalers.get_mut(&v.id()) {
|
||||
new_frame = true;
|
||||
s.process_frame(frame, v.width, v.height, transmute(v.pixel_format))?
|
||||
} else {
|
||||
frame
|
||||
};
|
||||
egress_results.extend(Self::encode_mux_frame(
|
||||
&mut self.egress,
|
||||
var,
|
||||
enc,
|
||||
frame,
|
||||
)?);
|
||||
if new_frame {
|
||||
av_frame_free(&mut frame);
|
||||
}
|
||||
}
|
||||
|
||||
av_frame_free(&mut frame);
|
||||
VariantStream::Audio(a) => {
|
||||
if let Some((r, f)) = self.resampler.get_mut(&a.id()) {
|
||||
let frame_size = (*enc.codec_context()).frame_size;
|
||||
let mut resampled_frame = r.process_frame(frame)?;
|
||||
f.buffer_frame(resampled_frame)?;
|
||||
av_frame_free(&mut resampled_frame);
|
||||
// drain FIFO
|
||||
while let Some(mut frame) = f.get_frame(frame_size as usize)? {
|
||||
// Set correct timebase for audio (1/sample_rate)
|
||||
(*frame).time_base.num = 1;
|
||||
(*frame).time_base.den = a.sample_rate as i32;
|
||||
|
||||
egress_results.extend(Self::encode_mux_frame(
|
||||
&mut self.egress,
|
||||
var,
|
||||
enc,
|
||||
frame,
|
||||
)?);
|
||||
av_frame_free(&mut frame);
|
||||
}
|
||||
} else {
|
||||
egress_results.extend(Self::encode_mux_frame(
|
||||
&mut self.egress,
|
||||
var,
|
||||
enc,
|
||||
frame,
|
||||
)?);
|
||||
}
|
||||
}
|
||||
_ => {}
|
||||
}
|
||||
}
|
||||
|
||||
// Handle egress results (same as normal processing)
|
||||
if !egress_results.is_empty() {
|
||||
self.handle.block_on(async {
|
||||
for er in egress_results {
|
||||
if let EgressResult::Segments { created, deleted } = er {
|
||||
if let Err(e) = self
|
||||
.overseer
|
||||
.on_segments(&config.id, &created, &deleted)
|
||||
.await
|
||||
{
|
||||
bail!("Failed to process segment {}", e.to_string());
|
||||
}
|
||||
}
|
||||
}
|
||||
Ok(())
|
||||
})?;
|
||||
|
||||
av_frame_free(&mut frame);
|
||||
Ok(egress_results)
|
||||
}
|
||||
|
||||
unsafe fn encode_mux_frame(
|
||||
egress: &mut Vec<Box<dyn Egress>>,
|
||||
var: &VariantStream,
|
||||
encoder: &mut Encoder,
|
||||
frame: *mut AVFrame,
|
||||
) -> Result<Vec<EgressResult>> {
|
||||
let mut ret = vec![];
|
||||
// before encoding frame, rescale timestamps
|
||||
if !frame.is_null() {
|
||||
let enc_ctx = encoder.codec_context();
|
||||
(*frame).pict_type = AV_PICTURE_TYPE_NONE;
|
||||
(*frame).pts = av_rescale_q((*frame).pts, (*frame).time_base, (*enc_ctx).time_base);
|
||||
(*frame).pkt_dts =
|
||||
av_rescale_q((*frame).pkt_dts, (*frame).time_base, (*enc_ctx).time_base);
|
||||
(*frame).duration =
|
||||
av_rescale_q((*frame).duration, (*frame).time_base, (*enc_ctx).time_base);
|
||||
(*frame).time_base = (*enc_ctx).time_base;
|
||||
}
|
||||
|
||||
Ok(())
|
||||
|
||||
let packets = encoder.encode_frame(frame)?;
|
||||
// pass new packets to egress
|
||||
for mut pkt in packets {
|
||||
for eg in egress.iter_mut() {
|
||||
let er = eg.process_pkt(pkt, &var.id())?;
|
||||
ret.push(er);
|
||||
}
|
||||
av_packet_free(&mut pkt);
|
||||
}
|
||||
|
||||
Ok(ret)
|
||||
}
|
||||
|
||||
/// EOF, cleanup
|
||||
@ -297,23 +311,36 @@ impl PipelineRunner {
|
||||
self.setup()?;
|
||||
|
||||
let config = if let Some(config) = &self.config {
|
||||
config
|
||||
config.clone()
|
||||
} else {
|
||||
bail!("Pipeline not configured, cannot run")
|
||||
};
|
||||
|
||||
// run transcoder pipeline
|
||||
let (mut pkt, stream_info) = self.demuxer.get_packet()?;
|
||||
|
||||
let (mut pkt, _) = self.demuxer.get_packet()?;
|
||||
|
||||
let src_video_stream = config
|
||||
.ingress_info
|
||||
.streams
|
||||
.iter()
|
||||
.find(|s| s.index == config.video_src)
|
||||
.unwrap();
|
||||
let src_audio_stream = config
|
||||
.ingress_info
|
||||
.streams
|
||||
.iter()
|
||||
.find(|s| Some(s.index) == config.audio_src);
|
||||
|
||||
// Handle state transitions based on packet availability
|
||||
match (&self.state, pkt.is_null()) {
|
||||
(RunnerState::Normal, true) => {
|
||||
// First time entering idle mode
|
||||
info!("Stream input disconnected, entering idle mode with placeholder content");
|
||||
info!("Stream input disconnected, entering idle mode");
|
||||
|
||||
self.state = RunnerState::Idle {
|
||||
start_time: Instant::now(),
|
||||
variant_index: 0,
|
||||
last_frame_time: None,
|
||||
gen: FrameGenerator::from_stream(src_video_stream, src_audio_stream)?,
|
||||
};
|
||||
}
|
||||
(RunnerState::Idle { start_time, .. }, true) => {
|
||||
@ -332,27 +359,23 @@ impl PipelineRunner {
|
||||
// Normal operation continues
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
// Process based on current state
|
||||
match &self.state {
|
||||
RunnerState::Idle { .. } => {
|
||||
// Process a single idle frame (rotating through variants)
|
||||
self.process_single_idle_frame(config)?;
|
||||
|
||||
// Free the null packet if needed
|
||||
if !pkt.is_null() {
|
||||
av_packet_free(&mut pkt);
|
||||
}
|
||||
|
||||
return Ok(true); // Continue processing
|
||||
let result = match &mut self.state {
|
||||
RunnerState::Idle { gen, .. } => {
|
||||
let frame = gen.next()?;
|
||||
let stream = if (*frame).sample_rate > 0 {
|
||||
self.demuxer.get_stream(
|
||||
src_audio_stream
|
||||
.context("frame generator created an audio frame with no src stream")?
|
||||
.index,
|
||||
)?
|
||||
} else {
|
||||
self.demuxer.get_stream(src_video_stream.index)?
|
||||
};
|
||||
self.process_frame(&config, stream, frame)?
|
||||
}
|
||||
RunnerState::Normal => {
|
||||
// Normal packet processing
|
||||
if pkt.is_null() {
|
||||
// This shouldn't happen in Normal state but handle gracefully
|
||||
return Ok(true);
|
||||
}
|
||||
|
||||
// TODO: For copy streams, skip decoder
|
||||
let frames = match self.decoder.decode_pkt(pkt) {
|
||||
Ok(f) => f,
|
||||
@ -364,133 +387,24 @@ impl PipelineRunner {
|
||||
|
||||
let mut egress_results = vec![];
|
||||
for (frame, stream) in frames {
|
||||
// Copy frame from GPU if using hwaccel decoding
|
||||
let mut frame = get_frame_from_hw(frame)?;
|
||||
(*frame).time_base = (*stream).time_base;
|
||||
|
||||
let p = (*stream).codecpar;
|
||||
if (*p).codec_type == AVMediaType::AVMEDIA_TYPE_VIDEO {
|
||||
// Conditionally generate thumbnails based on interval (0 = disabled)
|
||||
if self.thumb_interval > 0 && (self.frame_ctr % self.thumb_interval) == 0 {
|
||||
let thumb_start = Instant::now();
|
||||
let dst_pic = PathBuf::from(&self.out_dir)
|
||||
.join(config.id.to_string())
|
||||
.join("thumb.webp");
|
||||
{
|
||||
let mut sw = Scaler::new();
|
||||
let mut scaled_frame = sw.process_frame(
|
||||
frame,
|
||||
(*frame).width as _,
|
||||
(*frame).height as _,
|
||||
AV_PIX_FMT_YUV420P,
|
||||
)?;
|
||||
|
||||
let mut encoder = Encoder::new(AV_CODEC_ID_WEBP)?
|
||||
.with_height((*scaled_frame).height)
|
||||
.with_width((*scaled_frame).width)
|
||||
.with_pix_fmt(transmute((*scaled_frame).format))
|
||||
.open(None)?;
|
||||
|
||||
encoder.save_picture(scaled_frame, dst_pic.to_str().unwrap())?;
|
||||
av_frame_free(&mut scaled_frame);
|
||||
// Adjust frame pts time without start_offset
|
||||
// Egress streams don't have a start time offset
|
||||
if (*stream).start_time != AV_NOPTS_VALUE {
|
||||
(*frame).pts -= (*stream).start_time;
|
||||
}
|
||||
|
||||
let thumb_duration = thumb_start.elapsed();
|
||||
info!(
|
||||
"Saved thumb ({:.2}ms) to: {}",
|
||||
thumb_duration.as_millis() as f32 / 1000.0,
|
||||
dst_pic.display(),
|
||||
);
|
||||
let results = self.process_frame(&config, stream, frame)?;
|
||||
egress_results.extend(results);
|
||||
}
|
||||
|
||||
self.frame_ctr += 1;
|
||||
av_packet_free(&mut pkt);
|
||||
egress_results
|
||||
}
|
||||
|
||||
// Get the variants which want this pkt
|
||||
let pkt_vars = config
|
||||
.variants
|
||||
.iter()
|
||||
.filter(|v| v.src_index() == (*stream).index as usize);
|
||||
for var in pkt_vars {
|
||||
let enc = if let Some(enc) = self.encoders.get_mut(&var.id()) {
|
||||
enc
|
||||
} else {
|
||||
//warn!("Frame had nowhere to go in {} :/", var.id());
|
||||
continue;
|
||||
};
|
||||
|
||||
// scaling / resampling
|
||||
let mut new_frame = false;
|
||||
let mut frame = match var {
|
||||
VariantStream::Video(v) => {
|
||||
if let Some(s) = self.scalers.get_mut(&v.id()) {
|
||||
new_frame = true;
|
||||
s.process_frame(frame, v.width, v.height, transmute(v.pixel_format))?
|
||||
} else {
|
||||
frame
|
||||
}
|
||||
}
|
||||
VariantStream::Audio(a) => {
|
||||
if let Some((r, f)) = self.resampler.get_mut(&a.id()) {
|
||||
let frame_size = (*enc.codec_context()).frame_size;
|
||||
new_frame = true;
|
||||
let mut resampled_frame = r.process_frame(frame)?;
|
||||
if let Some(ret) =
|
||||
f.buffer_frame(resampled_frame, frame_size as usize)?
|
||||
{
|
||||
// Set correct timebase for audio (1/sample_rate)
|
||||
(*ret).time_base.num = 1;
|
||||
(*ret).time_base.den = a.sample_rate as i32;
|
||||
av_frame_free(&mut resampled_frame);
|
||||
ret
|
||||
} else {
|
||||
av_frame_free(&mut resampled_frame);
|
||||
continue;
|
||||
}
|
||||
} else {
|
||||
frame
|
||||
}
|
||||
}
|
||||
_ => frame,
|
||||
};
|
||||
|
||||
// before encoding frame, rescale timestamps
|
||||
if !frame.is_null() {
|
||||
let enc_ctx = enc.codec_context();
|
||||
(*frame).pict_type = AV_PICTURE_TYPE_NONE;
|
||||
(*frame).pts =
|
||||
av_rescale_q((*frame).pts, (*frame).time_base, (*enc_ctx).time_base);
|
||||
(*frame).pkt_dts =
|
||||
av_rescale_q((*frame).pkt_dts, (*frame).time_base, (*enc_ctx).time_base);
|
||||
(*frame).duration =
|
||||
av_rescale_q((*frame).duration, (*frame).time_base, (*enc_ctx).time_base);
|
||||
(*frame).time_base = (*enc_ctx).time_base;
|
||||
}
|
||||
|
||||
let packets = enc.encode_frame(frame)?;
|
||||
// pass new packets to egress
|
||||
for mut pkt in packets {
|
||||
for eg in self.egress.iter_mut() {
|
||||
let er = eg.process_pkt(pkt, &var.id())?;
|
||||
egress_results.push(er);
|
||||
}
|
||||
av_packet_free(&mut pkt);
|
||||
}
|
||||
|
||||
if new_frame {
|
||||
av_frame_free(&mut frame);
|
||||
}
|
||||
}
|
||||
|
||||
av_frame_free(&mut frame);
|
||||
}
|
||||
|
||||
av_packet_free(&mut pkt);
|
||||
};
|
||||
|
||||
// egress results - process async operations without blocking if possible
|
||||
if !egress_results.is_empty() {
|
||||
if !result.is_empty() {
|
||||
self.handle.block_on(async {
|
||||
for er in egress_results {
|
||||
for er in result {
|
||||
if let EgressResult::Segments { created, deleted } = er {
|
||||
if let Err(e) = self
|
||||
.overseer
|
||||
@ -510,8 +424,6 @@ impl PipelineRunner {
|
||||
info!("Average fps: {:.2}", n_frames as f32 / elapsed);
|
||||
self.fps_counter_start = Instant::now();
|
||||
self.fps_last_frame_ctr = self.frame_ctr;
|
||||
}
|
||||
} // Close the RunnerState::Normal match arm
|
||||
}
|
||||
Ok(true)
|
||||
}
|
||||
@ -523,6 +435,8 @@ impl PipelineRunner {
|
||||
|
||||
let info = self.demuxer.probe_input()?;
|
||||
|
||||
info!("{}", info);
|
||||
|
||||
// convert to internal type
|
||||
let i_info = IngressInfo {
|
||||
bitrate: info.bitrate,
|
||||
@ -542,18 +456,16 @@ impl PipelineRunner {
|
||||
height: s.height,
|
||||
fps: s.fps,
|
||||
sample_rate: s.sample_rate,
|
||||
channels: s.channels,
|
||||
language: s.language.clone(),
|
||||
})
|
||||
.collect(),
|
||||
};
|
||||
|
||||
let mut cfg = self
|
||||
let cfg = self
|
||||
.handle
|
||||
.block_on(async { self.overseer.start_stream(&self.connection, &i_info).await })?;
|
||||
|
||||
// Store ingress info in config for placeholder generation
|
||||
cfg.ingress_info = Some(i_info.clone());
|
||||
|
||||
|
||||
self.config = Some(cfg);
|
||||
self.info = Some(i_info);
|
||||
|
||||
@ -649,7 +561,10 @@ impl Drop for PipelineRunner {
|
||||
self.copy_stream.clear();
|
||||
self.egress.clear();
|
||||
|
||||
info!("PipelineRunner cleaned up resources for stream: {}", self.connection.key);
|
||||
info!(
|
||||
"PipelineRunner cleaned up resources for stream: {}",
|
||||
self.connection.key
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@ -7,7 +7,7 @@ edition = "2021"
|
||||
default = ["srt", "rtmp", "test-pattern"]
|
||||
srt = ["zap-stream-core/srt"]
|
||||
rtmp = ["zap-stream-core/rtmp"]
|
||||
test-pattern = ["zap-stream-core/test-pattern", "zap-stream-db/test-pattern"]
|
||||
test-pattern = ["zap-stream-db/test-pattern"]
|
||||
|
||||
[dependencies]
|
||||
zap-stream-db = { path = "../zap-stream-db" }
|
||||
|
||||
@ -5,7 +5,6 @@ endpoints:
|
||||
- "rtmp://127.0.0.1:3336"
|
||||
- "srt://127.0.0.1:3335"
|
||||
- "tcp://127.0.0.1:3334"
|
||||
- "test-pattern://"
|
||||
|
||||
# Public hostname which points to the IP address used to listen for all [endpoints]
|
||||
endpoints_public_hostname: "localhost"
|
||||
|
||||
@ -9,45 +9,55 @@
|
||||
color: white;
|
||||
font-family: monospace;
|
||||
}
|
||||
|
||||
.container {
|
||||
padding: 20px;
|
||||
max-width: 1200px;
|
||||
margin: 0 auto;
|
||||
}
|
||||
|
||||
.stream-list {
|
||||
margin: 20px 0;
|
||||
}
|
||||
|
||||
.stream-item {
|
||||
background: #333;
|
||||
margin: 10px 0;
|
||||
padding: 15px;
|
||||
border-radius: 5px;
|
||||
}
|
||||
|
||||
.stream-title {
|
||||
font-size: 18px;
|
||||
font-weight: bold;
|
||||
margin-bottom: 5px;
|
||||
}
|
||||
|
||||
.stream-link {
|
||||
color: #00ff00;
|
||||
text-decoration: none;
|
||||
}
|
||||
|
||||
.stream-link:hover {
|
||||
text-decoration: underline;
|
||||
}
|
||||
|
||||
.video-player {
|
||||
margin: 20px 0;
|
||||
max-width: 800px;
|
||||
}
|
||||
|
||||
video {
|
||||
width: 100%;
|
||||
max-width: 800px;
|
||||
background: #000;
|
||||
}
|
||||
|
||||
.no-streams {
|
||||
color: #999;
|
||||
font-style: italic;
|
||||
}
|
||||
|
||||
.player-section {
|
||||
margin-top: 30px;
|
||||
border-top: 1px solid #555;
|
||||
@ -59,19 +69,24 @@
|
||||
<body>
|
||||
<div class="container">
|
||||
<h1>Welcome to {{public_url}}</h1>
|
||||
|
||||
|
||||
<h2>Active Streams</h2>
|
||||
{{#has_streams}}
|
||||
<div class="stream-list">
|
||||
{{#streams}}
|
||||
<div class="stream-item">
|
||||
<div class="stream-title">{{title}}</div>
|
||||
{{#summary}}<div class="stream-summary">{{summary}}</div>{{/summary}}
|
||||
{{#summary}}
|
||||
<div class="stream-summary">{{summary}}</div>
|
||||
{{/summary}}
|
||||
<div>
|
||||
<a href="{{live_url}}" class="stream-link">📺 {{live_url}}</a>
|
||||
{{#viewer_count}}<span style="margin-left: 15px;">👥 {{viewer_count}} viewers</span>{{/viewer_count}}
|
||||
<a href="{{live_url}}" class="stream-link">{{live_url}}</a>
|
||||
{{#viewer_count}}<span style="margin-left: 15px;">{{viewer_count}} viewers</span>{{/viewer_count}}
|
||||
</div>
|
||||
<button onclick="playStream('{{live_url}}')" style="margin-top: 5px; background: #00ff00; color: black; border: none; padding: 5px 10px; cursor: pointer;">Play</button>
|
||||
<button onclick="playStream('{{live_url}}')"
|
||||
style="margin-top: 5px; background: #00ff00; color: black; border: none; padding: 5px 10px; cursor: pointer;">
|
||||
Play
|
||||
</button>
|
||||
</div>
|
||||
{{/streams}}
|
||||
</div>
|
||||
@ -79,15 +94,19 @@
|
||||
{{^has_streams}}
|
||||
<div class="no-streams">No active streams</div>
|
||||
{{/has_streams}}
|
||||
|
||||
|
||||
<div class="player-section">
|
||||
<h2>Stream Player</h2>
|
||||
<div class="video-player">
|
||||
<video id="video-player" controls></video>
|
||||
</div>
|
||||
<div style="margin-top: 10px;">
|
||||
<input type="text" id="stream-url" placeholder="Enter stream URL (e.g., /stream-id/live.m3u8)" style="width: 400px; padding: 5px; margin-right: 10px;">
|
||||
<button onclick="playCustomStream()" style="background: #00ff00; color: black; border: none; padding: 5px 10px; cursor: pointer;">Play URL</button>
|
||||
<input type="text" id="stream-url" placeholder="Enter stream URL (e.g., /stream-id/live.m3u8)"
|
||||
style="width: 400px; padding: 5px; margin-right: 10px;">
|
||||
<button onclick="playCustomStream()"
|
||||
style="background: #00ff00; color: black; border: none; padding: 5px 10px; cursor: pointer;">Play
|
||||
URL
|
||||
</button>
|
||||
</div>
|
||||
</div>
|
||||
</div>
|
||||
@ -104,12 +123,12 @@
|
||||
hls = new Hls();
|
||||
hls.loadSource(url);
|
||||
hls.attachMedia(video);
|
||||
hls.on(Hls.Events.MANIFEST_PARSED, function() {
|
||||
hls.on(Hls.Events.MANIFEST_PARSED, function () {
|
||||
video.play();
|
||||
});
|
||||
} else if (video.canPlayType('application/vnd.apple.mpegurl')) {
|
||||
video.src = url;
|
||||
video.addEventListener('loadedmetadata', function() {
|
||||
video.addEventListener('loadedmetadata', function () {
|
||||
video.play();
|
||||
});
|
||||
} else {
|
||||
|
||||
@ -11,18 +11,18 @@ use hyper::service::Service;
|
||||
use hyper::{Method, Request, Response};
|
||||
use log::error;
|
||||
use nostr_sdk::{serde_json, Alphabet, Event, Kind, PublicKey, SingleLetterTag, TagKind};
|
||||
use serde::{Serialize, Deserialize};
|
||||
use serde::Serialize;
|
||||
use std::future::Future;
|
||||
use std::path::PathBuf;
|
||||
use std::pin::Pin;
|
||||
use std::sync::Arc;
|
||||
use std::time::{Duration, Instant};
|
||||
use tokio::fs::File;
|
||||
use tokio::fs::File;
|
||||
use tokio::sync::RwLock;
|
||||
use tokio_util::io::ReaderStream;
|
||||
use zap_stream_core::viewer::ViewerTracker;
|
||||
|
||||
#[derive(Serialize)]
|
||||
#[derive(Serialize, Clone)]
|
||||
struct StreamData {
|
||||
id: String,
|
||||
title: String,
|
||||
@ -33,7 +33,7 @@ struct StreamData {
|
||||
viewer_count: Option<u64>,
|
||||
}
|
||||
|
||||
#[derive(Serialize)]
|
||||
#[derive(Serialize, Clone)]
|
||||
struct IndexTemplateData {
|
||||
public_url: String,
|
||||
has_streams: bool,
|
||||
@ -41,7 +41,7 @@ struct IndexTemplateData {
|
||||
streams: Vec<StreamData>,
|
||||
}
|
||||
|
||||
struct CachedStreams {
|
||||
pub struct CachedStreams {
|
||||
data: IndexTemplateData,
|
||||
cached_at: Instant,
|
||||
}
|
||||
@ -57,7 +57,12 @@ pub struct HttpServer {
|
||||
}
|
||||
|
||||
impl HttpServer {
|
||||
pub fn new(index_template: String, files_dir: PathBuf, api: Api, stream_cache: StreamCache) -> Self {
|
||||
pub fn new(
|
||||
index_template: String,
|
||||
files_dir: PathBuf,
|
||||
api: Api,
|
||||
stream_cache: StreamCache,
|
||||
) -> Self {
|
||||
Self {
|
||||
index_template,
|
||||
files_dir,
|
||||
@ -70,8 +75,11 @@ impl HttpServer {
|
||||
Self::get_cached_or_fetch_streams_static(&self.stream_cache, &self.api).await
|
||||
}
|
||||
|
||||
async fn get_cached_or_fetch_streams_static(stream_cache: &StreamCache, api: &Api) -> Result<IndexTemplateData> {
|
||||
const CACHE_DURATION: Duration = Duration::from_secs(60); // 1 minute
|
||||
async fn get_cached_or_fetch_streams_static(
|
||||
stream_cache: &StreamCache,
|
||||
api: &Api,
|
||||
) -> Result<IndexTemplateData> {
|
||||
const CACHE_DURATION: Duration = Duration::from_secs(10);
|
||||
|
||||
// Check if we have valid cached data
|
||||
{
|
||||
@ -86,7 +94,7 @@ impl HttpServer {
|
||||
// Cache is expired or missing, fetch new data
|
||||
let active_streams = api.get_active_streams().await?;
|
||||
let public_url = api.get_public_url();
|
||||
|
||||
|
||||
let template_data = if !active_streams.is_empty() {
|
||||
let streams: Vec<StreamData> = active_streams
|
||||
.into_iter()
|
||||
@ -94,10 +102,16 @@ impl HttpServer {
|
||||
let viewer_count = api.get_viewer_count(&stream.id);
|
||||
StreamData {
|
||||
id: stream.id.clone(),
|
||||
title: stream.title.unwrap_or_else(|| format!("Stream {}", &stream.id[..8])),
|
||||
title: stream
|
||||
.title
|
||||
.unwrap_or_else(|| format!("Stream {}", &stream.id[..8])),
|
||||
summary: stream.summary,
|
||||
live_url: format!("/{}/live.m3u8", stream.id),
|
||||
viewer_count: if viewer_count > 0 { Some(viewer_count) } else { None },
|
||||
viewer_count: if viewer_count > 0 {
|
||||
Some(viewer_count as _)
|
||||
} else {
|
||||
None
|
||||
},
|
||||
}
|
||||
})
|
||||
.collect();
|
||||
@ -140,13 +154,18 @@ impl HttpServer {
|
||||
playlist_path: &PathBuf,
|
||||
) -> Result<Response<BoxBody<Bytes, anyhow::Error>>, anyhow::Error> {
|
||||
// Extract stream ID from path (e.g., /uuid/live.m3u8 -> uuid)
|
||||
let path_parts: Vec<&str> = req.uri().path().trim_start_matches('/').split('/').collect();
|
||||
let path_parts: Vec<&str> = req
|
||||
.uri()
|
||||
.path()
|
||||
.trim_start_matches('/')
|
||||
.split('/')
|
||||
.collect();
|
||||
if path_parts.len() < 2 {
|
||||
return Ok(Response::builder().status(404).body(BoxBody::default())?);
|
||||
}
|
||||
|
||||
|
||||
let stream_id = path_parts[0];
|
||||
|
||||
|
||||
// Get client IP and User-Agent for tracking
|
||||
let client_ip = Self::get_client_ip(req);
|
||||
let user_agent = req
|
||||
@ -179,9 +198,10 @@ impl HttpServer {
|
||||
|
||||
// Read the playlist file
|
||||
let playlist_content = tokio::fs::read(playlist_path).await?;
|
||||
|
||||
|
||||
// Parse and modify playlist to add viewer token to URLs
|
||||
let modified_content = Self::add_viewer_token_to_playlist(&playlist_content, &viewer_token)?;
|
||||
let modified_content =
|
||||
Self::add_viewer_token_to_playlist(&playlist_content, &viewer_token)?;
|
||||
|
||||
Ok(Response::builder()
|
||||
.header("content-type", "application/vnd.apple.mpegurl")
|
||||
@ -205,7 +225,7 @@ impl HttpServer {
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
if let Some(real_ip) = req.headers().get("x-real-ip") {
|
||||
if let Ok(ip_str) = real_ip.to_str() {
|
||||
return ip_str.to_string();
|
||||
@ -220,17 +240,18 @@ impl HttpServer {
|
||||
// Parse the M3U8 playlist using the m3u8-rs crate
|
||||
let (_, playlist) = m3u8_rs::parse_playlist(content)
|
||||
.map_err(|e| anyhow::anyhow!("Failed to parse M3U8 playlist: {}", e))?;
|
||||
|
||||
|
||||
match playlist {
|
||||
m3u8_rs::Playlist::MasterPlaylist(mut master) => {
|
||||
// For master playlists, add viewer token to variant streams
|
||||
for variant in &mut master.variants {
|
||||
variant.uri = Self::add_token_to_url(&variant.uri, viewer_token);
|
||||
}
|
||||
|
||||
|
||||
// Write the modified playlist back to string
|
||||
let mut output = Vec::new();
|
||||
master.write_to(&mut output)
|
||||
master
|
||||
.write_to(&mut output)
|
||||
.map_err(|e| anyhow::anyhow!("Failed to write master playlist: {}", e))?;
|
||||
String::from_utf8(output)
|
||||
.map_err(|e| anyhow::anyhow!("Failed to convert playlist to string: {}", e))
|
||||
@ -242,7 +263,7 @@ impl HttpServer {
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
fn add_token_to_url(url: &str, viewer_token: &str) -> String {
|
||||
if url.contains('?') {
|
||||
format!("{}&vt={}", url, viewer_token)
|
||||
@ -264,7 +285,7 @@ impl Service<Request<Incoming>> for HttpServer {
|
||||
{
|
||||
let stream_cache = self.stream_cache.clone();
|
||||
let api = self.api.clone();
|
||||
|
||||
|
||||
// Compile template outside async move for better performance
|
||||
let template = match mustache::compile_str(&self.index_template) {
|
||||
Ok(t) => t,
|
||||
@ -272,40 +293,36 @@ impl Service<Request<Incoming>> for HttpServer {
|
||||
error!("Failed to compile template: {}", e);
|
||||
return Box::pin(async move {
|
||||
Ok(Response::builder()
|
||||
.status(500)
|
||||
.body(BoxBody::default()).unwrap())
|
||||
.status(500)
|
||||
.body(BoxBody::default())
|
||||
.unwrap())
|
||||
});
|
||||
}
|
||||
};
|
||||
|
||||
|
||||
return Box::pin(async move {
|
||||
// Use the existing method to get cached template data
|
||||
let template_data = Self::get_cached_or_fetch_streams_static(&stream_cache, &api).await;
|
||||
let template_data =
|
||||
Self::get_cached_or_fetch_streams_static(&stream_cache, &api).await;
|
||||
|
||||
match template_data {
|
||||
Ok(data) => {
|
||||
match template.render_to_string(&data) {
|
||||
Ok(index_html) => Ok(Response::builder()
|
||||
.header("content-type", "text/html")
|
||||
.header("server", "zap-stream-core")
|
||||
.body(
|
||||
Full::new(Bytes::from(index_html))
|
||||
.map_err(|e| match e {})
|
||||
.boxed(),
|
||||
)?),
|
||||
Err(e) => {
|
||||
error!("Failed to render template: {}", e);
|
||||
Ok(Response::builder()
|
||||
.status(500)
|
||||
.body(BoxBody::default())?)
|
||||
}
|
||||
Ok(data) => match template.render_to_string(&data) {
|
||||
Ok(index_html) => Ok(Response::builder()
|
||||
.header("content-type", "text/html")
|
||||
.header("server", "zap-stream-core")
|
||||
.body(
|
||||
Full::new(Bytes::from(index_html))
|
||||
.map_err(|e| match e {})
|
||||
.boxed(),
|
||||
)?),
|
||||
Err(e) => {
|
||||
error!("Failed to render template: {}", e);
|
||||
Ok(Response::builder().status(500).body(BoxBody::default())?)
|
||||
}
|
||||
}
|
||||
},
|
||||
Err(e) => {
|
||||
error!("Failed to fetch template data: {}", e);
|
||||
Ok(Response::builder()
|
||||
.status(500)
|
||||
.body(BoxBody::default())?)
|
||||
Ok(Response::builder().status(500).body(BoxBody::default())?)
|
||||
}
|
||||
}
|
||||
});
|
||||
@ -415,12 +432,21 @@ pub fn check_nip98_auth(req: &Request<Incoming>, public_url: &str) -> Result<Aut
|
||||
|
||||
// Construct full URI using public_url + path + query
|
||||
let request_uri = match req.uri().query() {
|
||||
Some(query) => format!("{}{}?{}", public_url.trim_end_matches('/'), req.uri().path(), query),
|
||||
Some(query) => format!(
|
||||
"{}{}?{}",
|
||||
public_url.trim_end_matches('/'),
|
||||
req.uri().path(),
|
||||
query
|
||||
),
|
||||
None => format!("{}{}", public_url.trim_end_matches('/'), req.uri().path()),
|
||||
};
|
||||
|
||||
if !url_tag.eq_ignore_ascii_case(&request_uri) {
|
||||
bail!("Invalid nostr event, URL tag invalid. Expected: {}, Got: {}", request_uri, url_tag);
|
||||
bail!(
|
||||
"Invalid nostr event, URL tag invalid. Expected: {}, Got: {}",
|
||||
request_uri,
|
||||
url_tag
|
||||
);
|
||||
}
|
||||
|
||||
// Check method tag
|
||||
|
||||
@ -29,7 +29,6 @@ use crate::monitor::BackgroundMonitor;
|
||||
use crate::overseer::ZapStreamOverseer;
|
||||
use crate::settings::Settings;
|
||||
use zap_stream_core::ingress::{file, tcp};
|
||||
use zap_stream_core::overseer::Overseer;
|
||||
|
||||
mod api;
|
||||
mod blossom;
|
||||
@ -76,7 +75,12 @@ async fn main() -> Result<()> {
|
||||
// Create shared stream cache
|
||||
let stream_cache: StreamCache = Arc::new(RwLock::new(None));
|
||||
// HTTP server
|
||||
let server = HttpServer::new(index_template.to_string(), PathBuf::from(settings.output_dir), api, stream_cache);
|
||||
let server = HttpServer::new(
|
||||
index_template.to_string(),
|
||||
PathBuf::from(settings.output_dir),
|
||||
api,
|
||||
stream_cache,
|
||||
);
|
||||
tasks.push(tokio::spawn(async move {
|
||||
let listener = TcpListener::bind(&http_addr).await?;
|
||||
|
||||
|
||||
@ -17,14 +17,14 @@ use url::Url;
|
||||
use uuid::Uuid;
|
||||
use zap_stream_core::egress::{EgressConfig, EgressSegment};
|
||||
use zap_stream_core::ingress::ConnectionInfo;
|
||||
use zap_stream_core::overseer::{IngressInfo, IngressStreamType, Overseer};
|
||||
use zap_stream_core::overseer::{IngressInfo, IngressStream, IngressStreamType, Overseer};
|
||||
use zap_stream_core::pipeline::{EgressType, PipelineConfig};
|
||||
use zap_stream_core::variant::audio::AudioVariant;
|
||||
use zap_stream_core::variant::mapping::VariantMapping;
|
||||
use zap_stream_core::variant::video::VideoVariant;
|
||||
use zap_stream_core::variant::{StreamMapping, VariantStream};
|
||||
use zap_stream_core::viewer::ViewerTracker;
|
||||
use zap_stream_db::{UserStream, UserStreamState, ZapStreamDb};
|
||||
use zap_stream_db::{IngestEndpoint, UserStream, UserStreamState, ZapStreamDb};
|
||||
|
||||
const STREAM_EVENT_KIND: u16 = 30_311;
|
||||
|
||||
@ -353,22 +353,18 @@ impl Overseer for ZapStreamOverseer {
|
||||
}
|
||||
|
||||
// Get ingest endpoint configuration based on connection type
|
||||
let endpoint_id = self
|
||||
.detect_endpoint(&connection)
|
||||
.await?
|
||||
.ok_or_else(|| anyhow::anyhow!("No ingest endpoints configured"))?;
|
||||
let endpoint = self
|
||||
.db
|
||||
.get_ingest_endpoint(endpoint_id)
|
||||
.await?
|
||||
.ok_or_else(|| anyhow::anyhow!("Ingest endpoint not found"))?;
|
||||
let endpoint = self.detect_endpoint(&connection).await?;
|
||||
|
||||
let variants = get_variants_from_endpoint(&stream_info, &endpoint)?;
|
||||
let cfg = get_variants_from_endpoint(&stream_info, &endpoint)?;
|
||||
|
||||
if cfg.video_src.is_none() || cfg.variants.is_empty() {
|
||||
bail!("No video src found");
|
||||
}
|
||||
|
||||
let mut egress = vec![];
|
||||
egress.push(EgressType::HLS(EgressConfig {
|
||||
name: "hls".to_string(),
|
||||
variants: variants.iter().map(|v| v.id()).collect(),
|
||||
variants: cfg.variants.iter().map(|v| v.id()).collect(),
|
||||
}));
|
||||
|
||||
let stream_id = Uuid::new_v4();
|
||||
@ -378,7 +374,7 @@ impl Overseer for ZapStreamOverseer {
|
||||
user_id: uid,
|
||||
starts: Utc::now(),
|
||||
state: UserStreamState::Live,
|
||||
endpoint_id: Some(endpoint_id),
|
||||
endpoint_id: Some(endpoint.id),
|
||||
..Default::default()
|
||||
};
|
||||
let stream_event = self.publish_stream_event(&new_stream, &user.pubkey).await?;
|
||||
@ -399,8 +395,11 @@ impl Overseer for ZapStreamOverseer {
|
||||
|
||||
Ok(PipelineConfig {
|
||||
id: stream_id,
|
||||
variants,
|
||||
variants: cfg.variants,
|
||||
egress,
|
||||
ingress_info: stream_info.clone(),
|
||||
video_src: cfg.video_src.unwrap().index,
|
||||
audio_src: cfg.audio_src.map(|s| s.index),
|
||||
})
|
||||
}
|
||||
|
||||
@ -525,25 +524,29 @@ impl Overseer for ZapStreamOverseer {
|
||||
|
||||
impl ZapStreamOverseer {
|
||||
/// Detect which ingest endpoint should be used based on connection info
|
||||
async fn detect_endpoint(&self, connection: &ConnectionInfo) -> Result<Option<u64>> {
|
||||
// Get all ingest endpoints and match by name against connection endpoint
|
||||
async fn detect_endpoint(&self, connection: &ConnectionInfo) -> Result<IngestEndpoint> {
|
||||
let endpoints = self.db.get_ingest_endpoints().await?;
|
||||
|
||||
for endpoint in &endpoints {
|
||||
if endpoint.name == connection.endpoint {
|
||||
return Ok(Some(endpoint.id));
|
||||
}
|
||||
}
|
||||
|
||||
// No matching endpoint found, use the most expensive one
|
||||
Ok(endpoints.into_iter().max_by_key(|e| e.cost).map(|e| e.id))
|
||||
let default = endpoints.iter().max_by_key(|e| e.cost);
|
||||
Ok(endpoints
|
||||
.iter()
|
||||
.find(|e| e.name == connection.endpoint)
|
||||
.or(default)
|
||||
.unwrap()
|
||||
.clone())
|
||||
}
|
||||
}
|
||||
|
||||
fn get_variants_from_endpoint(
|
||||
info: &IngressInfo,
|
||||
struct EndpointConfig<'a> {
|
||||
video_src: Option<&'a IngressStream>,
|
||||
audio_src: Option<&'a IngressStream>,
|
||||
variants: Vec<VariantStream>,
|
||||
}
|
||||
|
||||
fn get_variants_from_endpoint<'a>(
|
||||
info: &'a IngressInfo,
|
||||
endpoint: &zap_stream_db::IngestEndpoint,
|
||||
) -> Result<Vec<VariantStream>> {
|
||||
) -> Result<EndpointConfig<'a>> {
|
||||
let capabilities_str = endpoint.capabilities.as_deref().unwrap_or("");
|
||||
let capabilities: Vec<&str> = capabilities_str.split(',').collect();
|
||||
|
||||
@ -658,5 +661,9 @@ fn get_variants_from_endpoint(
|
||||
// Handle other capabilities like dvr:720h here if needed
|
||||
}
|
||||
|
||||
Ok(vars)
|
||||
Ok(EndpointConfig {
|
||||
audio_src,
|
||||
video_src,
|
||||
variants: vars,
|
||||
})
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user