diff --git a/.gitignore b/.gitignore
index ea8c4bf..6b39d31 100644
--- a/.gitignore
+++ b/.gitignore
@@ -1 +1,2 @@
 /target
+.idea/
\ No newline at end of file
diff --git a/Cargo.lock b/Cargo.lock
old mode 100644
new mode 100755
index 2a21ebd..12be4a3
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -8,7 +8,7 @@ version = "0.21.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "8a30b2e23b9e17a9f90641c7ab1549cd9b44f296d3ccbf309d2863cfe398a0cb"
 dependencies = [
- "gimli",
+    "gimli",
 ]
 
 [[package]]
@@ -23,9 +23,9 @@ version = "0.8.4"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "b169f7a6d4742236a0a00c541b845991d0ac43e546831af1249753ab4c3aa3a0"
 dependencies = [
- "cfg-if",
- "cipher",
- "cpufeatures",
+    "cfg-if",
+    "cipher",
+    "cpufeatures",
 ]
 
 [[package]]
@@ -34,7 +34,7 @@ version = "1.1.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "b2969dcb958b36655471fc61f7e416fa76033bdd4bfed0678d8fee1e2d07a1f0"
 dependencies = [
- "memchr",
+    "memchr",
 ]
 
 [[package]]
@@ -43,7 +43,7 @@ version = "1.0.80"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "5ad32ce52e4161730f7098c077cd2ed6229b5804ccf99e5366be1ab72a98b4e1"
 dependencies = [
- "backtrace",
+    "backtrace",
 ]
 
 [[package]]
@@ -64,9 +64,9 @@ version = "0.1.77"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "c980ee35e870bd1a4d2c8294d4c04d0499e67bca1e4b5cefcc693c2fa00caea9"
 dependencies = [
- "proc-macro2",
- "quote",
- "syn 2.0.52",
+    "proc-macro2",
+    "quote",
+    "syn 2.0.52",
 ]
 
 [[package]]
@@ -81,13 +81,13 @@ version = "0.3.69"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "2089b7e3f35b9dd2d0ed921ead4f6d318c27680d4a5bd167b3ee120edb105837"
 dependencies = [
- "addr2line",
- "cc",
- "cfg-if",
- "libc",
- "miniz_oxide",
- "object",
- "rustc-demangle",
+    "addr2line",
+    "cc",
+    "cfg-if",
+    "libc",
+    "miniz_oxide",
+    "object",
+    "rustc-demangle",
 ]
 
 [[package]]
@@ -102,18 +102,18 @@ version = "0.64.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "c4243e6031260db77ede97ad86c27e501d646a27ab57b59a574f725d98ab1fb4"
 dependencies = [
- "bitflags 1.3.2",
- "cexpr",
- "clang-sys",
- "lazy_static",
- "lazycell",
- "peeking_take_while",
- "proc-macro2",
- "quote",
- "regex",
- "rustc-hash",
- "shlex",
- "syn 1.0.109",
+    "bitflags 1.3.2",
+    "cexpr",
+    "clang-sys",
+    "lazy_static",
+    "lazycell",
+    "peeking_take_while",
+    "proc-macro2",
+    "quote",
+    "regex",
+    "rustc-hash",
+    "shlex",
+    "syn 1.0.109",
 ]
 
 [[package]]
@@ -128,7 +128,7 @@ version = "2.4.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "ed570934406eb16438a4e976b1b4500774099c13b8cb96eec99f620f05090ddf"
 dependencies = [
- "serde",
+    "serde",
 ]
 
 [[package]]
@@ -137,7 +137,7 @@ version = "0.10.4"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "3078c7629b62d3f0439517fa394996acacc5cbc91c5a20d8c658e77abd503a71"
 dependencies = [
- "generic-array",
+    "generic-array",
 ]
 
 [[package]]
@@ -158,7 +158,7 @@ version = "0.6.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "6fac387a98bb7c37292057cffc56d62ecb629900026402633ae9160df93a8766"
 dependencies = [
- "nom",
+    "nom",
 ]
 
 [[package]]
@@ -173,8 +173,8 @@ version = "0.4.4"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "773f3b9af64447d2ce9850330c473515014aa235e6a783b02db81ff39e4a3dad"
 dependencies = [
- "crypto-common",
- "inout",
+    "crypto-common",
+    "inout",
 ]
 
 [[package]]
@@ -183,9 +183,9 @@ version = "1.7.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "67523a3b4be3ce1989d607a828d036249522dd9c1c8de7f4dd2dae43a37369d1"
 dependencies = [
- "glob",
- "libc",
- "libloading",
+    "glob",
+    "libc",
+    "libloading",
 ]
 
 [[package]]
@@ -194,18 +194,18 @@ version = "0.14.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "7328b20597b53c2454f0b1919720c25c7339051c02b72b7e05409e00b14132be"
 dependencies = [
- "async-trait",
- "convert_case 0.6.0",
- "json5",
- "lazy_static",
- "nom",
- "pathdiff",
- "ron",
- "rust-ini",
- "serde",
- "serde_json",
- "toml",
- "yaml-rust",
+    "async-trait",
+    "convert_case 0.6.0",
+    "json5",
+    "lazy_static",
+    "nom",
+    "pathdiff",
+    "ron",
+    "rust-ini",
+    "serde",
+    "serde_json",
+    "toml",
+    "yaml-rust",
 ]
 
 [[package]]
@@ -214,7 +214,7 @@ version = "0.1.18"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "87e00182fe74b066627d63b85fd550ac2998d4b0bd86bfed477a0ae4c7c71359"
 dependencies = [
- "const-random-macro",
+    "const-random-macro",
 ]
 
 [[package]]
@@ -223,9 +223,9 @@ version = "0.1.16"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "f9d839f2a20b0aee515dc581a6172f2321f96cab76c1a38a4c584a194955390e"
 dependencies = [
- "getrandom",
- "once_cell",
- "tiny-keccak",
+    "getrandom",
+    "once_cell",
+    "tiny-keccak",
 ]
 
 [[package]]
@@ -240,7 +240,7 @@ version = "0.6.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "ec182b0ca2f35d8fc196cf3404988fd8b8c739a4d270ff118a398feb0cbec1ca"
 dependencies = [
- "unicode-segmentation",
+    "unicode-segmentation",
 ]
 
 [[package]]
@@ -249,7 +249,7 @@ version = "0.2.12"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "53fe5e26ff1b7aef8bca9c6080520cfb8d9333c7568e1829cef191a9723e5504"
 dependencies = [
- "libc",
+    "libc",
 ]
 
 [[package]]
@@ -264,8 +264,8 @@ version = "0.1.6"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "1bfb12502f3fc46cca1bb51ac28df9d618d813cdc3d2f25b9fe775a34af26bb3"
 dependencies = [
- "generic-array",
- "typenum",
+    "generic-array",
+    "typenum",
 ]
 
 [[package]]
@@ -274,7 +274,7 @@ version = "0.9.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "0369ee1ad671834580515889b80f2ea915f23b8be8d0daa4bbaf2ac5c7590835"
 dependencies = [
- "cipher",
+    "cipher",
 ]
 
 [[package]]
@@ -283,11 +283,11 @@ version = "0.99.17"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "4fb810d30a7c1953f91334de7244731fc3f3c10d7fe163338a35b9f640960321"
 dependencies = [
- "convert_case 0.4.0",
- "proc-macro2",
- "quote",
- "rustc_version",
- "syn 1.0.109",
+    "convert_case 0.4.0",
+    "proc-macro2",
+    "quote",
+    "rustc_version",
+    "syn 1.0.109",
 ]
 
 [[package]]
@@ -296,9 +296,9 @@ version = "0.10.7"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "9ed9a281f7bc9b7576e61468ba615a66a5c8cfdff42420a70aa82701a3b1e292"
 dependencies = [
- "block-buffer",
- "crypto-common",
- "subtle",
+    "block-buffer",
+    "crypto-common",
+    "subtle",
 ]
 
 [[package]]
@@ -307,7 +307,7 @@ version = "0.5.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "442039f5147480ba31067cb00ada1adae6892028e40e45fc5de7b7df6dcc1b5f"
 dependencies = [
- "const-random",
+    "const-random",
 ]
 
 [[package]]
@@ -316,11 +316,11 @@ version = "0.10.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "4cd405aab171cb85d6735e5c8d9db038c17d3ca007a4d2c25f337935c3d90580"
 dependencies = [
- "humantime",
- "is-terminal",
- "log",
- "regex",
- "termcolor",
+    "humantime",
+    "is-terminal",
+    "log",
+    "regex",
+    "termcolor",
 ]
 
 [[package]]
@@ -335,12 +335,12 @@ version = "6.1.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "c2529ad916d08c3562c754c21bc9b17a26c7882c0f5706cc2cd69472175f1620"
 dependencies = [
- "bindgen",
- "cc",
- "libc",
- "num_cpus",
- "pkg-config",
- "vcpkg",
+    "bindgen",
+    "cc",
+    "libc",
+    "num_cpus",
+    "pkg-config",
+    "vcpkg",
 ]
 
 [[package]]
@@ -349,7 +349,7 @@ version = "1.2.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "e13624c2627564efccf4934284bdd98cbaa14e79b0b5a141218e507b3a823456"
 dependencies = [
- "percent-encoding",
+    "percent-encoding",
 ]
 
 [[package]]
@@ -358,12 +358,12 @@ version = "0.3.30"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "645c6916888f6cb6350d2550b80fb63e734897a8498abe35cfb732b6487804b0"
 dependencies = [
- "futures-channel",
- "futures-core",
- "futures-io",
- "futures-sink",
- "futures-task",
- "futures-util",
+    "futures-channel",
+    "futures-core",
+    "futures-io",
+    "futures-sink",
+    "futures-task",
+    "futures-util",
 ]
 
 [[package]]
@@ -372,8 +372,8 @@ version = "0.3.30"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "eac8f7d7865dcb88bd4373ab671c8cf4508703796caa2b1985a9ca867b3fcb78"
 dependencies = [
- "futures-core",
- "futures-sink",
+    "futures-core",
+    "futures-sink",
 ]
 
 [[package]]
@@ -394,9 +394,9 @@ version = "0.3.30"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "87750cf4b7a4c0625b1529e4c543c2182106e4dedc60a2a6455e00d212c489ac"
 dependencies = [
- "proc-macro2",
- "quote",
- "syn 2.0.52",
+    "proc-macro2",
+    "quote",
+    "syn 2.0.52",
 ]
 
 [[package]]
@@ -417,16 +417,16 @@ version = "0.3.30"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "3d6401deb83407ab3da39eba7e33987a73c3df0c82b4bb5813ee871c19c41d48"
 dependencies = [
- "futures-channel",
- "futures-core",
- "futures-io",
- "futures-macro",
- "futures-sink",
- "futures-task",
- "memchr",
- "pin-project-lite",
- "pin-utils",
- "slab",
+    "futures-channel",
+    "futures-core",
+    "futures-io",
+    "futures-macro",
+    "futures-sink",
+    "futures-task",
+    "memchr",
+    "pin-project-lite",
+    "pin-utils",
+    "slab",
 ]
 
 [[package]]
@@ -435,8 +435,8 @@ version = "0.14.7"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "85649ca51fd72272d7821adaf274ad91c288277713d9c18820d8499a7ff69e9a"
 dependencies = [
- "typenum",
- "version_check",
+    "typenum",
+    "version_check",
 ]
 
 [[package]]
@@ -445,9 +445,9 @@ version = "0.2.12"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "190092ea657667030ac6a35e305e62fc4dd69fd98ac98631e5d3a2b1575a12b5"
 dependencies = [
- "cfg-if",
- "libc",
- "wasi",
+    "cfg-if",
+    "libc",
+    "wasi",
 ]
 
 [[package]]
@@ -492,7 +492,7 @@ version = "0.12.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "6c49c37c09c17a53d937dfbb742eb3a961d65a994e6bcdcf37e7399d0cc8ab5e"
 dependencies = [
- "digest",
+    "digest",
 ]
 
 [[package]]
@@ -507,8 +507,8 @@ version = "0.5.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "634d9b1461af396cad843f47fdba5597a4f9e6ddd4bfb6ff5d85028c25cb12f6"
 dependencies = [
- "unicode-bidi",
- "unicode-normalization",
+    "unicode-bidi",
+    "unicode-normalization",
 ]
 
 [[package]]
@@ -517,8 +517,8 @@ version = "2.2.5"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "7b0b929d511467233429c45a44ac1dcaa21ba0f5ba11e4879e6ed28ddb4f9df4"
 dependencies = [
- "equivalent",
- "hashbrown 0.14.3",
+    "equivalent",
+    "hashbrown 0.14.3",
 ]
 
 [[package]]
@@ -527,7 +527,7 @@ version = "0.1.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "a0c10553d664a4d0bcff9f4215d0aac67a639cc68ef660840afe309b807bc9f5"
 dependencies = [
- "generic-array",
+    "generic-array",
 ]
 
 [[package]]
@@ -536,9 +536,9 @@ version = "0.4.12"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "f23ff5ef2b80d608d61efee834934d862cd92461afc0560dedf493e4c033738b"
 dependencies = [
- "hermit-abi",
- "libc",
- "windows-sys 0.52.0",
+    "hermit-abi",
+    "libc",
+    "windows-sys 0.52.0",
 ]
 
 [[package]]
@@ -553,9 +553,9 @@ version = "0.4.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "96b0db21af676c1ce64250b5f40f3ce2cf27e4e47cb91ed91eb6fe9350b430c1"
 dependencies = [
- "pest",
- "pest_derive",
- "serde",
+    "pest",
+    "pest_derive",
+    "serde",
 ]
 
 [[package]]
@@ -564,7 +564,7 @@ version = "0.4.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "4ee7893dab2e44ae5f9d0173f26ff4aa327c10b01b06a72b52dd9405b628640d"
 dependencies = [
- "indexmap",
+    "indexmap",
 ]
 
 [[package]]
@@ -591,8 +591,8 @@ version = "0.8.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "0c2a198fb6b0eada2a8df47933734e6d35d350665a33a3593d7164fa52c75c19"
 dependencies = [
- "cfg-if",
- "windows-targets 0.52.4",
+    "cfg-if",
+    "windows-targets 0.52.4",
 ]
 
 [[package]]
@@ -625,7 +625,7 @@ version = "0.7.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "9d811f3e15f28568be3407c8e7fdb6514c1cda3cb30683f15b6a1a1dc4ea14a7"
 dependencies = [
- "adler",
+    "adler",
 ]
 
 [[package]]
@@ -634,9 +634,9 @@ version = "0.8.11"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "a4a650543ca06a924e8b371db273b2756685faae30f8487da1b56505a8f78b0c"
 dependencies = [
- "libc",
- "wasi",
- "windows-sys 0.48.0",
+    "libc",
+    "wasi",
+    "windows-sys 0.48.0",
 ]
 
 [[package]]
@@ -645,8 +645,8 @@ version = "7.1.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "d273983c5a657a70a3e8f2a01329822f3b8c8172b73826411a55751e404a0a4a"
 dependencies = [
- "memchr",
- "minimal-lexical",
+    "memchr",
+    "minimal-lexical",
 ]
 
 [[package]]
@@ -655,7 +655,7 @@ version = "0.2.18"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "da0df0e5185db44f69b44f26786fe401b6c293d1907744beaa7fa62b2e5a517a"
 dependencies = [
- "autocfg",
+    "autocfg",
 ]
 
 [[package]]
@@ -664,8 +664,8 @@ version = "1.16.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "4161fcb6d602d4d2081af7c3a45852d875a03dd337a6bfdd6e06407b61342a43"
 dependencies = [
- "hermit-abi",
- "libc",
+    "hermit-abi",
+    "libc",
 ]
 
 [[package]]
@@ -674,7 +674,7 @@ version = "0.32.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "a6a622008b6e321afc04970976f62ee297fdbaa6f95318ca343e3eebb9648441"
 dependencies = [
- "memchr",
+    "memchr",
 ]
 
 [[package]]
@@ -689,8 +689,8 @@ version = "0.6.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "4ed8acf08e98e744e5384c8bc63ceb0364e68a6854187221c18df61c4797690e"
 dependencies = [
- "dlv-list",
- "hashbrown 0.13.2",
+    "dlv-list",
+    "hashbrown 0.13.2",
 ]
 
 [[package]]
@@ -705,7 +705,7 @@ version = "0.12.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "f8ed6a7761f76e3b9f92dfb0a60a6a6477c61024b775147ff0973a02653abaf2"
 dependencies = [
- "digest",
+    "digest",
 ]
 
 [[package]]
@@ -726,9 +726,9 @@ version = "2.7.8"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "56f8023d0fb78c8e03784ea1c7f3fa36e68a723138990b8d5a47d916b651e7a8"
 dependencies = [
- "memchr",
- "thiserror",
- "ucd-trie",
+    "memchr",
+    "thiserror",
+    "ucd-trie",
 ]
 
 [[package]]
@@ -737,8 +737,8 @@ version = "2.7.8"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "b0d24f72393fd16ab6ac5738bc33cdb6a9aa73f8b902e8fe29cf4e67d7dd1026"
 dependencies = [
- "pest",
- "pest_generator",
+    "pest",
+    "pest_generator",
 ]
 
 [[package]]
@@ -747,11 +747,11 @@ version = "2.7.8"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "fdc17e2a6c7d0a492f0158d7a4bd66cc17280308bbaff78d5bef566dca35ab80"
 dependencies = [
- "pest",
- "pest_meta",
- "proc-macro2",
- "quote",
- "syn 2.0.52",
+    "pest",
+    "pest_meta",
+    "proc-macro2",
+    "quote",
+    "syn 2.0.52",
 ]
 
 [[package]]
@@ -760,9 +760,9 @@ version = "2.7.8"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "934cd7631c050f4674352a6e835d5f6711ffbfb9345c2fc0107155ac495ae293"
 dependencies = [
- "once_cell",
- "pest",
- "sha2",
+    "once_cell",
+    "pest",
+    "sha2",
 ]
 
 [[package]]
@@ -801,8 +801,8 @@ version = "0.5.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "865724d4dbe39d9f3dd3b52b88d859d66bcb2d6a0acfd5ea68a65fb66d4bdc1c"
 dependencies = [
- "env_logger",
- "log",
+    "env_logger",
+    "log",
 ]
 
 [[package]]
@@ -811,7 +811,7 @@ version = "1.0.78"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "e2422ad645d89c99f8f3e6b88a9fdeca7fabeac836b1002371c4367c8f984aae"
 dependencies = [
- "unicode-ident",
+    "unicode-ident",
 ]
 
 [[package]]
@@ -820,7 +820,7 @@ version = "1.0.35"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "291ec9ab5efd934aaf503a6466c5d5251535d108ee747472c3977cc5acc868ef"
 dependencies = [
- "proc-macro2",
+    "proc-macro2",
 ]
 
 [[package]]
@@ -829,9 +829,9 @@ version = "0.8.5"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "34af8d1a0e25924bc5b7c43c079c942339d8f0a8b57c39049bef581b46327404"
 dependencies = [
- "libc",
- "rand_chacha",
- "rand_core",
+    "libc",
+    "rand_chacha",
+    "rand_core",
 ]
 
 [[package]]
@@ -840,8 +840,8 @@ version = "0.3.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "e6c10a63a0fa32252be49d21e7709d4d4baf8d231c2dbce1eaa8141b9b127d88"
 dependencies = [
- "ppv-lite86",
- "rand_core",
+    "ppv-lite86",
+    "rand_core",
 ]
 
 [[package]]
@@ -850,7 +850,7 @@ version = "0.6.4"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "ec0be4795e2f6a28069bec0b5ff3e2ac9bafc99e6a9a7dc3547996c5c816922c"
 dependencies = [
- "getrandom",
+    "getrandom",
 ]
 
 [[package]]
@@ -859,10 +859,10 @@ version = "1.10.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "b62dbe01f0b06f9d8dc7d49e05a0785f153b00b2c227856282f671e0318c9b15"
 dependencies = [
- "aho-corasick",
- "memchr",
- "regex-automata",
- "regex-syntax",
+    "aho-corasick",
+    "memchr",
+    "regex-automata",
+    "regex-syntax",
 ]
 
 [[package]]
@@ -871,9 +871,9 @@ version = "0.4.6"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "86b83b8b9847f9bf95ef68afb0b8e6cdb80f498442f5179a29fad448fcc1eaea"
 dependencies = [
- "aho-corasick",
- "memchr",
- "regex-syntax",
+    "aho-corasick",
+    "memchr",
+    "regex-syntax",
 ]
 
 [[package]]
@@ -888,10 +888,10 @@ version = "0.8.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "b91f7eff05f748767f183df4320a63d6936e9c6107d97c9e6bdd9784f4289c94"
 dependencies = [
- "base64",
- "bitflags 2.4.2",
- "serde",
- "serde_derive",
+    "base64",
+    "bitflags 2.4.2",
+    "serde",
+    "serde_derive",
 ]
 
 [[package]]
@@ -900,8 +900,8 @@ version = "0.19.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "7e2a3bcec1f113553ef1c88aae6c020a369d03d55b58de9869a0908930385091"
 dependencies = [
- "cfg-if",
- "ordered-multimap",
+    "cfg-if",
+    "ordered-multimap",
 ]
 
 [[package]]
@@ -922,7 +922,7 @@ version = "0.4.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "bfa0f585226d2e68097d4f95d113b15b83a82e819ab25717ec0590d9584ef366"
 dependencies = [
- "semver",
+    "semver",
 ]
 
 [[package]]
@@ -943,7 +943,7 @@ version = "1.0.197"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "3fb1c873e1b9b056a4dc4c0c198b24c3ffa059243875552b2bd0933b1aee4ce2"
 dependencies = [
- "serde_derive",
+    "serde_derive",
 ]
 
 [[package]]
@@ -952,9 +952,9 @@ version = "1.0.197"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "7eb0b34b42edc17f6b7cac84a52a1c5f0e1bb2227e997ca9011ea3dd34e8610b"
 dependencies = [
- "proc-macro2",
- "quote",
- "syn 2.0.52",
+    "proc-macro2",
+    "quote",
+    "syn 2.0.52",
 ]
 
 [[package]]
@@ -963,9 +963,9 @@ version = "1.0.114"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "c5f09b1bd632ef549eaa9f60a1f8de742bdbc698e6cee2095fc84dde5f549ae0"
 dependencies = [
- "itoa",
- "ryu",
- "serde",
+    "itoa",
+    "ryu",
+    "serde",
 ]
 
 [[package]]
@@ -974,7 +974,7 @@ version = "0.6.5"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "eb3622f419d1296904700073ea6cc23ad690adbd66f13ea683df73298736f0c1"
 dependencies = [
- "serde",
+    "serde",
 ]
 
 [[package]]
@@ -983,9 +983,9 @@ version = "0.10.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "f5058ada175748e33390e40e872bd0fe59a19f265d0158daa551c5a88a76009c"
 dependencies = [
- "cfg-if",
- "cpufeatures",
- "digest",
+    "cfg-if",
+    "cpufeatures",
+    "digest",
 ]
 
 [[package]]
@@ -994,9 +994,9 @@ version = "0.10.8"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "793db75ad2bcafc3ffa7c68b215fee268f537982cd901d132f89c6343f3a3dc8"
 dependencies = [
- "cfg-if",
- "cpufeatures",
- "digest",
+    "cfg-if",
+    "cpufeatures",
+    "digest",
 ]
 
 [[package]]
@@ -1011,7 +1011,7 @@ version = "0.4.9"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "8f92a496fb766b417c996b9c5e57daf2f7ad3b0bebe1ccfca4856390e3d3bb67"
 dependencies = [
- "autocfg",
+    "autocfg",
 ]
 
 [[package]]
@@ -1020,8 +1020,8 @@ version = "0.5.6"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "05ffd9c0a93b7543e062e759284fcf5f5e3b098501104bfbdde4d404db792871"
 dependencies = [
- "libc",
- "windows-sys 0.52.0",
+    "libc",
+    "windows-sys 0.52.0",
 ]
 
 [[package]]
@@ -1030,26 +1030,26 @@ version = "0.4.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "db1a69f75dbd56399adf74daf49e077d31969431aa5cfa2ed1aa8ae175335ea7"
 dependencies = [
- "aes",
- "array-init",
- "arraydeque",
- "bitflags 2.4.2",
- "bytes",
- "cipher",
- "ctr",
- "derive_more",
- "hex",
- "hmac",
- "keyed_priority_queue",
- "log",
- "pbkdf2",
- "rand",
- "regex",
- "sha-1",
- "streaming-stats",
- "take-until",
- "thiserror",
- "url",
+    "aes",
+    "array-init",
+    "arraydeque",
+    "bitflags 2.4.2",
+    "bytes",
+    "cipher",
+    "ctr",
+    "derive_more",
+    "hex",
+    "hmac",
+    "keyed_priority_queue",
+    "log",
+    "pbkdf2",
+    "rand",
+    "regex",
+    "sha-1",
+    "streaming-stats",
+    "take-until",
+    "thiserror",
+    "url",
 ]
 
 [[package]]
@@ -1058,36 +1058,36 @@ version = "0.4.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "7a2383b0b3530aa62f72950ac949b1e27be4f8184305ada5c8ebc0fc9bac8010"
 dependencies = [
- "bytes",
- "futures",
- "log",
- "rand",
- "socket2",
- "srt-protocol",
- "tokio",
- "tokio-stream",
+    "bytes",
+    "futures",
+    "log",
+    "rand",
+    "socket2",
+    "srt-protocol",
+    "tokio",
+    "tokio-stream",
 ]
 
 [[package]]
 name = "stream-core"
 version = "0.1.0"
 dependencies = [
- "anyhow",
- "async-trait",
- "bytes",
- "config",
- "ffmpeg-sys-next",
- "futures-util",
- "libc",
- "log",
- "pretty-hex",
- "pretty_env_logger",
- "serde",
- "srt-tokio",
- "tokio",
- "tokio-stream",
- "url",
- "uuid",
+    "anyhow",
+    "async-trait",
+    "bytes",
+    "config",
+    "ffmpeg-sys-next",
+    "futures-util",
+    "libc",
+    "log",
+    "pretty-hex",
+    "pretty_env_logger",
+    "serde",
+    "srt-tokio",
+    "tokio",
+    "tokio-stream",
+    "url",
+    "uuid",
 ]
 
 [[package]]
@@ -1096,7 +1096,7 @@ version = "0.2.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "b0d670ce4e348a2081843569e0f79b21c99c91bb9028b3b3ecb0f050306de547"
 dependencies = [
- "num-traits",
+    "num-traits",
 ]
 
 [[package]]
@@ -1111,9 +1111,9 @@ version = "1.0.109"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "72b64191b275b66ffe2469e8af2c1cfe3bafa67b529ead792a6d0160888b4237"
 dependencies = [
- "proc-macro2",
- "quote",
- "unicode-ident",
+    "proc-macro2",
+    "quote",
+    "unicode-ident",
 ]
 
 [[package]]
@@ -1122,9 +1122,9 @@ version = "2.0.52"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "b699d15b36d1f02c3e7c69f8ffef53de37aefae075d8488d4ba1a7788d574a07"
 dependencies = [
- "proc-macro2",
- "quote",
- "unicode-ident",
+    "proc-macro2",
+    "quote",
+    "unicode-ident",
 ]
 
 [[package]]
@@ -1139,7 +1139,7 @@ version = "1.4.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "06794f8f6c5c898b3275aebefa6b8a1cb24cd2c6c79397ab15774837a0bc5755"
 dependencies = [
- "winapi-util",
+    "winapi-util",
 ]
 
 [[package]]
@@ -1148,7 +1148,7 @@ version = "1.0.57"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "1e45bcbe8ed29775f228095caf2cd67af7a4ccf756ebff23a306bf3e8b47b24b"
 dependencies = [
- "thiserror-impl",
+    "thiserror-impl",
 ]
 
 [[package]]
@@ -1157,9 +1157,9 @@ version = "1.0.57"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "a953cb265bef375dae3de6663da4d3804eee9682ea80d8e2542529b73c531c81"
 dependencies = [
- "proc-macro2",
- "quote",
- "syn 2.0.52",
+    "proc-macro2",
+    "quote",
+    "syn 2.0.52",
 ]
 
 [[package]]
@@ -1168,7 +1168,7 @@ version = "2.0.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "2c9d3793400a45f954c52e73d068316d76b6f4e36977e3fcebb13a2721e80237"
 dependencies = [
- "crunchy",
+    "crunchy",
 ]
 
 [[package]]
@@ -1177,7 +1177,7 @@ version = "1.6.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "87cc5ceb3875bb20c2890005a4e226a4651264a5c75edb2421b52861a0a0cb50"
 dependencies = [
- "tinyvec_macros",
+    "tinyvec_macros",
 ]
 
 [[package]]
@@ -1192,15 +1192,15 @@ version = "1.36.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "61285f6515fa018fb2d1e46eb21223fff441ee8db5d0f1435e8ab4f5cdb80931"
 dependencies = [
- "backtrace",
- "bytes",
- "libc",
- "mio",
- "num_cpus",
- "pin-project-lite",
- "socket2",
- "tokio-macros",
- "windows-sys 0.48.0",
+    "backtrace",
+    "bytes",
+    "libc",
+    "mio",
+    "num_cpus",
+    "pin-project-lite",
+    "socket2",
+    "tokio-macros",
+    "windows-sys 0.48.0",
 ]
 
 [[package]]
@@ -1209,9 +1209,9 @@ version = "2.2.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "5b8a1e28f2deaa14e508979454cb3a223b10b938b45af148bc0986de36f1923b"
 dependencies = [
- "proc-macro2",
- "quote",
- "syn 2.0.52",
+    "proc-macro2",
+    "quote",
+    "syn 2.0.52",
 ]
 
 [[package]]
@@ -1220,10 +1220,10 @@ version = "0.1.14"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "397c988d37662c7dda6d2208364a706264bf3d6138b11d436cbac0ad38832842"
 dependencies = [
- "futures-core",
- "pin-project-lite",
- "tokio",
- "tokio-util",
+    "futures-core",
+    "pin-project-lite",
+    "tokio",
+    "tokio-util",
 ]
 
 [[package]]
@@ -1232,11 +1232,11 @@ version = "0.7.10"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "5419f34732d9eb6ee4c3578b7989078579b7f039cbbb9ca2c4da015749371e15"
 dependencies = [
- "bytes",
- "futures-core",
- "futures-sink",
- "pin-project-lite",
- "tokio",
+    "bytes",
+    "futures-core",
+    "futures-sink",
+    "pin-project-lite",
+    "tokio",
 ]
 
 [[package]]
@@ -1245,10 +1245,10 @@ version = "0.8.12"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "e9dd1545e8208b4a5af1aa9bbd0b4cf7e9ea08fabc5d0a5c67fcaafa17433aa3"
 dependencies = [
- "serde",
- "serde_spanned",
- "toml_datetime",
- "toml_edit",
+    "serde",
+    "serde_spanned",
+    "toml_datetime",
+    "toml_edit",
 ]
 
 [[package]]
@@ -1257,7 +1257,7 @@ version = "0.6.5"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "3550f4e9685620ac18a50ed434eb3aec30db8ba93b0287467bca5826ea25baf1"
 dependencies = [
- "serde",
+    "serde",
 ]
 
 [[package]]
@@ -1266,11 +1266,11 @@ version = "0.22.8"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "c12219811e0c1ba077867254e5ad62ee2c9c190b0d957110750ac0cda1ae96cd"
 dependencies = [
- "indexmap",
- "serde",
- "serde_spanned",
- "toml_datetime",
- "winnow",
+    "indexmap",
+    "serde",
+    "serde_spanned",
+    "toml_datetime",
+    "winnow",
 ]
 
 [[package]]
@@ -1303,7 +1303,7 @@ version = "0.1.23"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "a56d1686db2308d901306f92a263857ef59ea39678a5458e7cb17f01415101f5"
 dependencies = [
- "tinyvec",
+    "tinyvec",
 ]
 
 [[package]]
@@ -1318,9 +1318,9 @@ version = "2.5.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "31e6302e3bb753d46e83516cae55ae196fc0c309407cf11ab35cc51a4c2a4633"
 dependencies = [
- "form_urlencoded",
- "idna",
- "percent-encoding",
+    "form_urlencoded",
+    "idna",
+    "percent-encoding",
 ]
 
 [[package]]
@@ -1329,8 +1329,8 @@ version = "1.8.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "a183cf7feeba97b4dd1c0d46788634f6221d87fa961b305bed08c851829efcc0"
 dependencies = [
- "getrandom",
- "serde",
+    "getrandom",
+    "serde",
 ]
 
 [[package]]
@@ -1357,8 +1357,8 @@ version = "0.3.9"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "5c839a674fcd7a98952e593242ea400abe93992746761e38641405d28b00f419"
 dependencies = [
- "winapi-i686-pc-windows-gnu",
- "winapi-x86_64-pc-windows-gnu",
+    "winapi-i686-pc-windows-gnu",
+    "winapi-x86_64-pc-windows-gnu",
 ]
 
 [[package]]
@@ -1373,7 +1373,7 @@ version = "0.1.6"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "f29e6f9198ba0d26b4c9f07dbe6f9ed633e1f3d5b8b414090084349e46a52596"
 dependencies = [
- "winapi",
+    "winapi",
 ]
 
 [[package]]
@@ -1388,7 +1388,7 @@ version = "0.48.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "677d2418bec65e3338edb076e806bc1ec15693c5d0104683f2efe857f61056a9"
 dependencies = [
- "windows-targets 0.48.5",
+    "windows-targets 0.48.5",
 ]
 
 [[package]]
@@ -1397,7 +1397,7 @@ version = "0.52.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "282be5f36a8ce781fad8c8ae18fa3f9beff57ec1b52cb3de0789201425d9a33d"
 dependencies = [
- "windows-targets 0.52.4",
+    "windows-targets 0.52.4",
 ]
 
 [[package]]
@@ -1406,13 +1406,13 @@ version = "0.48.5"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "9a2fa6e2155d7247be68c096456083145c183cbbbc2764150dda45a87197940c"
 dependencies = [
- "windows_aarch64_gnullvm 0.48.5",
- "windows_aarch64_msvc 0.48.5",
- "windows_i686_gnu 0.48.5",
- "windows_i686_msvc 0.48.5",
- "windows_x86_64_gnu 0.48.5",
- "windows_x86_64_gnullvm 0.48.5",
- "windows_x86_64_msvc 0.48.5",
+    "windows_aarch64_gnullvm 0.48.5",
+    "windows_aarch64_msvc 0.48.5",
+    "windows_i686_gnu 0.48.5",
+    "windows_i686_msvc 0.48.5",
+    "windows_x86_64_gnu 0.48.5",
+    "windows_x86_64_gnullvm 0.48.5",
+    "windows_x86_64_msvc 0.48.5",
 ]
 
 [[package]]
@@ -1421,13 +1421,13 @@ version = "0.52.4"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "7dd37b7e5ab9018759f893a1952c9420d060016fc19a472b4bb20d1bdd694d1b"
 dependencies = [
- "windows_aarch64_gnullvm 0.52.4",
- "windows_aarch64_msvc 0.52.4",
- "windows_i686_gnu 0.52.4",
- "windows_i686_msvc 0.52.4",
- "windows_x86_64_gnu 0.52.4",
- "windows_x86_64_gnullvm 0.52.4",
- "windows_x86_64_msvc 0.52.4",
+    "windows_aarch64_gnullvm 0.52.4",
+    "windows_aarch64_msvc 0.52.4",
+    "windows_i686_gnu 0.52.4",
+    "windows_i686_msvc 0.52.4",
+    "windows_x86_64_gnu 0.52.4",
+    "windows_x86_64_gnullvm 0.52.4",
+    "windows_x86_64_msvc 0.52.4",
 ]
 
 [[package]]
@@ -1520,7 +1520,7 @@ version = "0.6.5"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "dffa400e67ed5a4dd237983829e66475f0a4a26938c4b04c21baede6262215b8"
 dependencies = [
- "memchr",
+    "memchr",
 ]
 
 [[package]]
@@ -1529,5 +1529,5 @@ version = "0.4.5"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "56c1936c4cc7a1c9ab21a1ebb602eb942ba868cbd44a99cb7cdc5892335e1c85"
 dependencies = [
- "linked-hash-map",
+    "linked-hash-map",
 ]
diff --git a/Cargo.toml b/Cargo.toml
old mode 100644
new mode 100755
index 6ae537a..c7ba3d9
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -5,7 +5,7 @@ edition = "2021"
 
 [dependencies]
 srt-tokio = "0.4.3"
-tokio = { version = "1.36.0" , features = ["rt-multi-thread", "sync"]}
+tokio = { version = "1.36.0", features = ["rt-multi-thread", "sync"] }
 anyhow = { version = "1.0.80", features = ["backtrace"] }
 pretty_env_logger = "0.5.0"
 bytes = "1.5.0"
@@ -13,7 +13,7 @@ tokio-stream = "0.1.14"
 futures-util = "0.3.30"
 async-trait = "0.1.77"
 log = "0.4.21"
-ffmpeg-sys-next = { version = "6.1.0", features = ["avformat", "avcodec", "swscale", "avfilter"]}
+ffmpeg-sys-next = { version = "6.1.0", features = ["avformat", "avcodec", "swscale", "avfilter"] }
 libc = "0.2.153"
 pretty-hex = "0.4.1"
 uuid = { version = "1.8.0", features = ["v4", "serde"] }
diff --git a/Dockerfile b/Dockerfile
old mode 100644
new mode 100755
diff --git a/README.md b/README.md
old mode 100644
new mode 100755
diff --git a/config.toml b/config.toml
old mode 100644
new mode 100755
diff --git a/src/decode/mod.rs b/src/decode/mod.rs
index 1856b2e..f6b4dc0 100644
--- a/src/decode/mod.rs
+++ b/src/decode/mod.rs
@@ -3,12 +3,12 @@ use std::ptr;
 
 use anyhow::Error;
 use ffmpeg_sys_next::{
-    av_frame_alloc, av_packet_unref, avcodec_alloc_context3, avcodec_find_decoder,
-    avcodec_free_context, avcodec_open2, avcodec_parameters_to_context, avcodec_receive_frame,
-    avcodec_send_packet, AVCodec, AVCodecContext, AVPacket, AVStream, AVERROR, AVERROR_EOF,
+    av_frame_alloc, av_packet_unref, AVCodec, avcodec_alloc_context3,
+    avcodec_find_decoder, avcodec_free_context, avcodec_open2, avcodec_parameters_to_context,
+    avcodec_receive_frame, avcodec_send_packet, AVCodecContext, AVERROR, AVERROR_EOF, AVPacket, AVStream,
 };
 use tokio::sync::broadcast;
-use tokio::sync::mpsc::{Receiver, UnboundedReceiver};
+use tokio::sync::mpsc::UnboundedReceiver;
 
 use crate::pipeline::PipelinePayload;
 
@@ -32,6 +32,7 @@ pub struct Decoder {
 }
 
 unsafe impl Send for Decoder {}
+
 unsafe impl Sync for Decoder {}
 
 impl Decoder {
@@ -46,16 +47,23 @@ impl Decoder {
         }
     }
 
-    pub unsafe fn decode_pkt(&mut self, pkt: *mut AVPacket) -> Result<(), Error> {
+    pub unsafe fn decode_pkt(&mut self, pkt: *mut AVPacket) -> Result<usize, Error> {
         let stream_index = (*pkt).stream_index as i32;
         let stream = (*pkt).opaque as *mut AVStream;
-        let codec_par = (*stream).codecpar;
-        let has_codec_params = codec_par != ptr::null_mut();
-        if !has_codec_params {
-            panic!("Cant handle pkt, dropped!");
-        }
+        assert_eq!(
+            stream_index,
+            (*stream).index,
+            "Passed stream reference does not match stream_index of packet"
+        );
 
-        if has_codec_params && !self.codecs.contains_key(&stream_index) {
+        let codec_par = (*stream).codecpar;
+        assert_ne!(
+            codec_par,
+            ptr::null_mut(),
+            "Codec parameters are missing from stream"
+        );
+
+        if !self.codecs.contains_key(&stream_index) {
             let codec = avcodec_find_decoder((*codec_par).codec_id);
             if codec == ptr::null_mut() {
                 return Err(Error::msg("Failed to find codec"));
@@ -81,6 +89,7 @@ impl Decoder {
                 return Err(Error::msg(format!("Failed to decode packet {}", ret)));
             }
 
+            let mut frames = 0;
             while ret >= 0 {
                 let frame = av_frame_alloc();
                 ret = avcodec_receive_frame(ctx.context, frame);
@@ -91,22 +100,26 @@ impl Decoder {
                     return Err(Error::msg(format!("Failed to decode {}", ret)));
                 }
                 (*frame).time_base = (*pkt).time_base;
+                (*frame).opaque = stream as *mut libc::c_void;
                 self.chan_out.send(PipelinePayload::AvFrame(frame))?;
+                frames += 1;
             }
+            return Ok(frames);
         }
-        Ok(())
+        Ok(0)
     }
 
-    pub fn process(&mut self) -> Result<(), Error> {
+    pub fn process(&mut self) -> Result<usize, Error> {
         while let Ok(pkg) = self.chan_in.try_recv() {
-            if let PipelinePayload::AvPacket(pkt) = pkg {
+            return if let PipelinePayload::AvPacket(pkt) = pkg {
                 unsafe {
-                    self.decode_pkt(pkt)?;
+                    let frames = self.decode_pkt(pkt)?;
+                    Ok(frames)
                 }
             } else {
-                return Err(Error::msg("Payload not supported"));
-            }
+                Err(Error::msg("Payload not supported"))
+            };
         }
-        Ok(())
+        Ok(0)
     }
 }
diff --git a/src/demux/info.rs b/src/demux/info.rs
index 89a6fd8..df4990a 100644
--- a/src/demux/info.rs
+++ b/src/demux/info.rs
@@ -1,4 +1,3 @@
-use ffmpeg_sys_next::AVCodecParameters;
 use crate::fraction::Fraction;
 
 #[derive(Clone, Debug, PartialEq)]
@@ -18,12 +17,9 @@ pub struct StreamInfoChannel {
     pub channel_type: StreamChannelType,
     pub width: usize,
     pub height: usize,
-    pub codec_params: *const AVCodecParameters,
+    pub fps: f32,
 }
 
-unsafe impl Sync for StreamInfoChannel {}
-unsafe impl Send for StreamInfoChannel {}
-
 impl TryInto<Fraction> for StreamInfoChannel {
     type Error = ();
 
diff --git a/src/demux/mod.rs b/src/demux/mod.rs
index baf12b1..0d8c1f7 100644
--- a/src/demux/mod.rs
+++ b/src/demux/mod.rs
@@ -3,9 +3,9 @@ use std::time::{Duration, SystemTime};
 
 use anyhow::Error;
 use bytes::{Bytes, BytesMut};
-use ffmpeg_sys_next::AVMediaType::{AVMEDIA_TYPE_AUDIO, AVMEDIA_TYPE_VIDEO};
 use ffmpeg_sys_next::*;
-use log::info;
+use ffmpeg_sys_next::AVMediaType::{AVMEDIA_TYPE_AUDIO, AVMEDIA_TYPE_VIDEO};
+use log::{info, warn};
 use tokio::sync::mpsc::{UnboundedReceiver, UnboundedSender};
 use tokio::time::Instant;
 
@@ -32,6 +32,7 @@ pub(crate) struct Demuxer {
 }
 
 unsafe impl Send for Demuxer {}
+
 unsafe impl Sync for Demuxer {}
 
 unsafe extern "C" fn read_data(
@@ -105,17 +106,12 @@ impl Demuxer {
             av_find_best_stream(self.ctx, AVMEDIA_TYPE_VIDEO, -1, -1, ptr::null_mut(), 0) as usize;
         if video_stream_index != AVERROR_STREAM_NOT_FOUND as usize {
             let video_stream = *(*self.ctx).streams.add(video_stream_index);
-            let codec_copy = unsafe {
-                let ptr = avcodec_parameters_alloc();
-                avcodec_parameters_copy(ptr, (*video_stream).codecpar);
-                ptr
-            };
             channel_infos.push(StreamInfoChannel {
                 index: video_stream_index,
                 channel_type: StreamChannelType::Video,
                 width: (*(*video_stream).codecpar).width as usize,
                 height: (*(*video_stream).codecpar).height as usize,
-                codec_params: codec_copy,
+                fps: av_q2d((*video_stream).avg_frame_rate) as f32,
             });
         }
 
@@ -133,7 +129,7 @@ impl Demuxer {
                 channel_type: StreamChannelType::Audio,
                 width: (*(*audio_stream).codecpar).width as usize,
                 height: (*(*audio_stream).codecpar).height as usize,
-                codec_params: codec_copy,
+                fps: 0.0,
             });
         }
 
@@ -149,6 +145,7 @@ impl Demuxer {
         if ret == AVERROR_EOF {
             // reset EOF flag, stream never ends
             (*(*self.ctx).pb).eof_reached = 0;
+            warn!("EOF was reached, stream might skip frames");
             return Ok(());
         }
         if ret < 0 {
@@ -156,10 +153,13 @@ impl Demuxer {
             return Err(Error::msg(msg));
         }
         let stream = *(*self.ctx).streams.add((*pkt).stream_index as usize);
-        (*pkt).time_base = (*stream).time_base;
+        if (*pkt).time_base.num == 0 {
+            (*pkt).time_base = (*stream).time_base;
+        }
         (*pkt).opaque = stream as *mut libc::c_void;
 
-        self.chan_out.send(PipelinePayload::AvPacket(pkt))?;
+        let pkg = PipelinePayload::AvPacket(pkt);
+        self.chan_out.send(pkg)?;
         Ok(())
     }
 
diff --git a/src/egress/hls.rs b/src/egress/hls.rs
index 748df3d..a908ea3 100644
--- a/src/egress/hls.rs
+++ b/src/egress/hls.rs
@@ -3,28 +3,28 @@ use std::mem::transmute;
 use std::ptr;
 use std::ptr::slice_from_raw_parts;
 
-use crate::demux::info::{DemuxStreamInfo, StreamChannelType};
-use crate::fraction::Fraction;
 use anyhow::Error;
+use ffmpeg_sys_next::{
+    AV_CH_LAYOUT_STEREO, av_channel_layout_default, av_dump_format, av_get_sample_fmt,
+    av_interleaved_write_frame, av_opt_set, av_packet_rescale_ts, av_write_frame, AVChannelLayout,
+    AVChannelLayout__bindgen_ty_1, avcodec_send_frame, avcodec_send_packet, AVCodecContext,
+    avformat_alloc_output_context2, avformat_new_stream, avformat_write_header, AVFormatContext, AVPacket,
+    AVRational,
+};
 use ffmpeg_sys_next::AVChannelOrder::AV_CHANNEL_ORDER_NATIVE;
 use ffmpeg_sys_next::AVColorSpace::AVCOL_SPC_BT709;
 use ffmpeg_sys_next::AVMediaType::{AVMEDIA_TYPE_AUDIO, AVMEDIA_TYPE_VIDEO};
 use ffmpeg_sys_next::AVPixelFormat::AV_PIX_FMT_YUV420P;
 use ffmpeg_sys_next::AVSampleFormat::AV_SAMPLE_FMT_FLT;
-use ffmpeg_sys_next::{
-    av_channel_layout_default, av_dump_format, av_interleaved_write_frame, av_opt_set,
-    av_packet_rescale_ts, av_write_frame, avcodec_send_frame, avcodec_send_packet,
-    avformat_alloc_output_context2, avformat_new_stream, avformat_write_header, AVChannelLayout,
-    AVChannelLayout__bindgen_ty_1, AVCodecContext, AVFormatContext, AVPacket, AVRational,
-    AV_CH_LAYOUT_STEREO,
-};
 use futures_util::StreamExt;
-use tokio::sync::mpsc::{Receiver, UnboundedReceiver};
 use log::info;
+use tokio::sync::mpsc::{Receiver, UnboundedReceiver};
 use uuid::{Bytes, Uuid, Variant};
 
+use crate::demux::info::{DemuxStreamInfo, StreamChannelType};
+use crate::fraction::Fraction;
 use crate::pipeline::{HLSEgressConfig, PipelinePayload};
-use crate::utils::get_ffmpeg_error_msg;
+use crate::utils::{get_ffmpeg_error_msg, id_ref_to_uuid};
 use crate::variant::{VariantStream, VideoVariant};
 
 pub struct HlsEgress {
@@ -36,10 +36,15 @@ pub struct HlsEgress {
 }
 
 unsafe impl Send for HlsEgress {}
+
 unsafe impl Sync for HlsEgress {}
 
 impl HlsEgress {
-    pub fn new(chan_in: UnboundedReceiver<PipelinePayload>, id: Uuid, config: HLSEgressConfig) -> Self {
+    pub fn new(
+        chan_in: UnboundedReceiver<PipelinePayload>,
+        id: Uuid,
+        config: HLSEgressConfig,
+    ) -> Self {
         Self {
             id,
             config,
@@ -137,7 +142,9 @@ impl HlsEgress {
 
                     (*params).codec_id = transmute(va.codec as i32);
                     (*params).codec_type = AVMEDIA_TYPE_AUDIO;
-                    (*params).format = AV_SAMPLE_FMT_FLT as libc::c_int;
+                    (*params).format = av_get_sample_fmt(
+                        format!("{}\0", va.sample_fmt).as_ptr() as *const libc::c_char
+                    ) as libc::c_int;
                     (*params).bit_rate = va.bitrate as i64;
                     (*params).sample_rate = va.sample_rate as libc::c_int;
                     (*params).ch_layout = AVChannelLayout {
@@ -165,19 +172,17 @@ impl HlsEgress {
     }
 
     unsafe fn process_pkt(&mut self, pkt: *mut AVPacket) -> Result<(), Error> {
-        let slice_raw = slice_from_raw_parts((*(*pkt).opaque_ref).data, 16);
-        let binding = Bytes::from(*(slice_raw as *const [u8; 16]));
-        let variant_id = Uuid::from_bytes_ref(&binding);
+        let variant_id = id_ref_to_uuid((*pkt).opaque_ref);
         let dst_stream_index = self.config.variants.iter().find_map(|v| match &v {
             VariantStream::Video(vv) => {
-                if vv.id.eq(variant_id) {
+                if vv.id.eq(&variant_id) {
                     Some(vv.dst_index)
                 } else {
                     None
                 }
             }
             VariantStream::Audio(va) => {
-                if va.id.eq(variant_id) {
+                if va.id.eq(&variant_id) {
                     Some(va.dst_index)
                 } else {
                     None
diff --git a/src/encode/mod.rs b/src/encode/mod.rs
index 9c88595..a08917d 100644
--- a/src/encode/mod.rs
+++ b/src/encode/mod.rs
@@ -1,23 +1,22 @@
 use std::mem::transmute;
 use std::ptr;
 
-use crate::ipc::Rx;
 use anyhow::Error;
-use async_trait::async_trait;
+use ffmpeg_sys_next::{
+    av_buffer_ref, AV_CH_LAYOUT_STEREO, AV_CODEC_FLAG_GLOBAL_HEADER, av_get_sample_fmt, av_opt_set,
+    av_packet_alloc, av_packet_free, AVBufferRef, AVChannelLayout,
+    AVChannelLayout__bindgen_ty_1, AVCodec, avcodec_alloc_context3, avcodec_find_encoder, avcodec_open2,
+    avcodec_receive_packet, avcodec_send_frame, AVCodecContext, AVERROR, AVFrame, AVRational,
+    AVStream,
+};
 use ffmpeg_sys_next::AVChannelOrder::AV_CHANNEL_ORDER_NATIVE;
 use ffmpeg_sys_next::AVPixelFormat::AV_PIX_FMT_YUV420P;
-use ffmpeg_sys_next::AVSampleFormat::AV_SAMPLE_FMT_FLT;
-use ffmpeg_sys_next::{
-    av_buffer_allocz, av_opt_set, av_packet_alloc, av_packet_free, avcodec_alloc_context3,
-    avcodec_find_encoder, avcodec_open2, avcodec_receive_packet, avcodec_send_frame, memcpy,
-    AVChannelLayout, AVChannelLayout__bindgen_ty_1, AVCodec, AVCodecContext, AVFrame, AVRational,
-    AVERROR, AV_CH_LAYOUT_STEREO,
-};
 use libc::EAGAIN;
-use tokio::sync::mpsc::{UnboundedSender};
+use tokio::sync::mpsc::UnboundedSender;
 
+use crate::ipc::Rx;
 use crate::pipeline::PipelinePayload;
-use crate::utils::get_ffmpeg_error_msg;
+use crate::utils::{get_ffmpeg_error_msg, variant_id_ref};
 use crate::variant::VariantStream;
 
 pub struct Encoder<T> {
@@ -26,26 +25,30 @@ pub struct Encoder<T> {
     codec: *const AVCodec,
     chan_in: T,
     chan_out: UnboundedSender<PipelinePayload>,
+    var_id_ref: *mut AVBufferRef,
 }
 
 unsafe impl<T> Send for Encoder<T> {}
+
 unsafe impl<T> Sync for Encoder<T> {}
 
 impl<TRecv> Encoder<TRecv>
-where
-    TRecv: Rx<PipelinePayload>,
+    where
+        TRecv: Rx<PipelinePayload>,
 {
     pub fn new(
         chan_in: TRecv,
         chan_out: UnboundedSender<PipelinePayload>,
         variant: VariantStream,
     ) -> Self {
+        let id_ref = variant_id_ref(&variant).unwrap();
         Self {
             ctx: ptr::null_mut(),
             codec: ptr::null(),
             variant,
             chan_in,
             chan_out,
+            var_id_ref: id_ref,
         }
     }
 
@@ -87,11 +90,9 @@ where
                     );
                 }
                 VariantStream::Audio(va) => {
-                    (*ctx).sample_fmt = if (*encoder).sample_fmts != ptr::null() {
-                        *(*encoder).sample_fmts.add(0)
-                    } else {
-                        AV_SAMPLE_FMT_FLT
-                    };
+                    (*ctx).sample_fmt = av_get_sample_fmt(
+                        format!("{}\0", va.sample_fmt).as_ptr() as *const libc::c_char
+                    );
                     (*ctx).bit_rate = va.bitrate as i64;
                     (*ctx).sample_rate = va.sample_rate as libc::c_int;
                     (*ctx).ch_layout = AVChannelLayout {
@@ -105,7 +106,7 @@ where
                     (*ctx).time_base = AVRational {
                         num: 1,
                         den: va.sample_rate as libc::c_int,
-                    }
+                    };
                 }
                 _ => {
                     // nothing
@@ -124,19 +125,23 @@ where
     }
 
     unsafe fn process_frame(&mut self, frame: *mut AVFrame) -> Result<(), Error> {
+        let stream = (*frame).opaque as *mut AVStream;
+        if (*stream).index as usize != self.variant.src_index() {
+            return Ok(());
+        }
         self.setup_encoder(frame)?;
 
         let mut ret = avcodec_send_frame(self.ctx, frame);
-        if ret < 0 {
+        if ret < 0 && ret != AVERROR(EAGAIN) {
             return Err(Error::msg(get_ffmpeg_error_msg(ret)));
         }
 
-        while ret > 0 {
+        while ret > 0 || ret == AVERROR(EAGAIN) {
             let mut pkt = av_packet_alloc();
             ret = avcodec_receive_packet(self.ctx, pkt);
             if ret < 0 {
+                av_packet_free(&mut pkt);
                 if ret == AVERROR(EAGAIN) {
-                    av_packet_free(&mut pkt);
                     return Ok(());
                 }
                 return Err(Error::msg(get_ffmpeg_error_msg(ret)));
@@ -144,27 +149,8 @@ where
 
             (*pkt).duration = (*frame).duration;
             (*pkt).time_base = (*frame).time_base;
-            (*pkt).opaque_ref = match &self.variant {
-                VariantStream::Audio(va) => {
-                    let buf = av_buffer_allocz(16);
-                    memcpy(
-                        (*buf).data as *mut libc::c_void,
-                        va.id.as_bytes().as_ptr() as *const libc::c_void,
-                        16,
-                    );
-                    buf
-                }
-                VariantStream::Video(vv) => {
-                    let buf = av_buffer_allocz(16);
-                    memcpy(
-                        (*buf).data as *mut libc::c_void,
-                        vv.id.as_bytes().as_ptr() as *const libc::c_void,
-                        16,
-                    );
-                    buf
-                }
-                _ => return Err(Error::msg("Cannot assign pkt stream index")),
-            };
+            (*pkt).opaque = stream as *mut libc::c_void;
+            (*pkt).opaque_ref = av_buffer_ref(self.var_id_ref);
             self.chan_out.send(PipelinePayload::AvPacket(pkt))?;
         }
 
@@ -172,7 +158,7 @@ where
     }
 
     pub fn process(&mut self) -> Result<(), Error> {
-        while let Ok(pkg) = self.chan_in.try_recv() {
+        while let Ok(pkg) = self.chan_in.try_recv_next() {
             match pkg {
                 PipelinePayload::AvFrame(frm) => unsafe {
                     self.process_frame(frm)?;
@@ -182,4 +168,4 @@ where
         }
         Ok(())
     }
-}
\ No newline at end of file
+}
diff --git a/src/ingress/srt.rs b/src/ingress/srt.rs
index e8c4160..4aa0b78 100644
--- a/src/ingress/srt.rs
+++ b/src/ingress/srt.rs
@@ -1,11 +1,14 @@
-use crate::ingress::ConnectionInfo;
-use crate::pipeline::builder::PipelineBuilder;
-use crate::pipeline::runner::PipelineRunner;
+use std::time::Instant;
+
 use futures_util::{StreamExt, TryStreamExt};
 use log::{info, warn};
 use srt_tokio::{SrtListener, SrtSocket};
 use tokio::sync::mpsc::unbounded_channel;
 
+use crate::ingress::ConnectionInfo;
+use crate::pipeline::builder::PipelineBuilder;
+use crate::pipeline::runner::PipelineRunner;
+
 pub async fn listen(addr: String, builder: PipelineBuilder) -> Result<(), anyhow::Error> {
     let (_binding, mut packets) = SrtListener::builder().bind(addr.clone()).await?;
 
diff --git a/src/ingress/tcp.rs b/src/ingress/tcp.rs
index a27fa70..9e867d6 100644
--- a/src/ingress/tcp.rs
+++ b/src/ingress/tcp.rs
@@ -1,15 +1,17 @@
-use crate::ingress::ConnectionInfo;
-use crate::pipeline::builder::PipelineBuilder;
-use crate::pipeline::runner::PipelineRunner;
+use std::io;
+
 use bytes::BytesMut;
 use futures_util::{StreamExt, TryStreamExt};
 use log::{error, info, warn};
 use srt_tokio::{SrtListener, SrtSocket};
-use std::io;
 use tokio::io::AsyncReadExt;
 use tokio::net::{TcpListener, TcpSocket};
 use tokio::sync::mpsc::unbounded_channel;
 
+use crate::ingress::ConnectionInfo;
+use crate::pipeline::builder::PipelineBuilder;
+use crate::pipeline::runner::PipelineRunner;
+
 pub async fn listen(addr: String, builder: PipelineBuilder) -> Result<(), anyhow::Error> {
     let listener = TcpListener::bind(addr.clone()).await.unwrap();
 
diff --git a/src/ipc.rs b/src/ipc.rs
index df49522..3a3992e 100644
--- a/src/ipc.rs
+++ b/src/ipc.rs
@@ -4,32 +4,32 @@ use async_trait::async_trait;
 #[async_trait]
 pub trait Rx<T> {
     async fn recv(&mut self) -> Result<T, Error>;
-    fn try_recv(&mut self) -> Result<T, Error>;
+    fn try_recv_next(&mut self) -> Result<T, Error>;
 }
 
 #[async_trait]
 impl<T> Rx<T> for tokio::sync::mpsc::UnboundedReceiver<T>
-where
-    T: Send + Sync,
+    where
+        T: Send + Sync,
 {
     async fn recv(&mut self) -> Result<T, Error> {
         self.recv().await.ok_or(Error::msg("recv error"))
     }
 
-    fn try_recv(&mut self) -> Result<T, Error> {
+    fn try_recv_next(&mut self) -> Result<T, Error> {
         Ok(self.try_recv()?)
     }
 }
 
 #[async_trait]
 impl<T> Rx<T> for tokio::sync::broadcast::Receiver<T>
-where
-    T: Send + Sync,
+    where
+        T: Send + Sync + Clone,
 {
     async fn recv(&mut self) -> Result<T, Error> {
         Ok(self.recv().await?)
     }
-    fn try_recv(&mut self) -> Result<T, Error> {
+    fn try_recv_next(&mut self) -> Result<T, Error> {
         Ok(self.try_recv()?)
     }
 }
diff --git a/src/main.rs b/src/main.rs
index 64f969d..9ad68ba 100644
--- a/src/main.rs
+++ b/src/main.rs
@@ -1,3 +1,16 @@
+use std::ffi::CStr;
+
+use config::Config;
+use futures_util::future::join_all;
+use futures_util::StreamExt;
+use log::{error, info};
+use tokio::sync::futures;
+use url::Url;
+
+use crate::pipeline::builder::PipelineBuilder;
+use crate::settings::Settings;
+use crate::webhook::Webhook;
+
 mod decode;
 mod demux;
 mod egress;
@@ -12,17 +25,6 @@ mod variant;
 mod webhook;
 mod ipc;
 
-use crate::pipeline::builder::PipelineBuilder;
-use crate::settings::Settings;
-use crate::webhook::Webhook;
-use config::Config;
-use futures_util::StreamExt;
-use log::{error, info};
-use std::ffi::CStr;
-use futures_util::future::join_all;
-use tokio::sync::futures;
-use url::Url;
-
 /// Test:  ffmpeg -re -f lavfi -i testsrc -g 2 -r 30 -pix_fmt yuv420p -s 1280x720 -c:v h264 -b:v 2000k -f mpegts srt://localhost:3333
 #[tokio::main]
 async fn main() -> anyhow::Result<()> {
diff --git a/src/pipeline/builder.rs b/src/pipeline/builder.rs
index 0d3a584..0b9a9ee 100644
--- a/src/pipeline/builder.rs
+++ b/src/pipeline/builder.rs
@@ -1,4 +1,5 @@
 use tokio::sync::mpsc::UnboundedReceiver;
+
 use crate::ingress::ConnectionInfo;
 use crate::pipeline::runner::PipelineRunner;
 use crate::webhook::Webhook;
diff --git a/src/pipeline/mod.rs b/src/pipeline/mod.rs
index 7994bef..6f85c62 100644
--- a/src/pipeline/mod.rs
+++ b/src/pipeline/mod.rs
@@ -1,7 +1,10 @@
 use std::ops::{Deref, DerefMut};
 
 use async_trait::async_trait;
-use ffmpeg_sys_next::{AVFrame, AVPacket};
+use ffmpeg_sys_next::{
+    av_frame_alloc, av_frame_free, av_frame_ref, av_packet_alloc, av_packet_free, av_packet_ref,
+    AVFrame, AVPacket,
+};
 use serde::{Deserialize, Serialize};
 
 use crate::demux::info::DemuxStreamInfo;
@@ -34,7 +37,7 @@ pub struct PipelineConfig {
     pub egress: Vec<EgressType>,
 }
 
-#[derive(Debug, Clone, PartialEq)]
+#[derive(Debug, PartialEq)]
 pub enum PipelinePayload {
     /// No output
     Empty,
@@ -49,8 +52,45 @@ pub enum PipelinePayload {
 }
 
 unsafe impl Send for PipelinePayload {}
+
 unsafe impl Sync for PipelinePayload {}
 
+impl Clone for PipelinePayload {
+    fn clone(&self) -> Self {
+        match self {
+            PipelinePayload::Empty => PipelinePayload::Empty,
+            PipelinePayload::Bytes(b) => PipelinePayload::Bytes(b.clone()),
+            PipelinePayload::AvPacket(p) => unsafe {
+                let new_pkt = av_packet_alloc();
+                av_packet_ref(new_pkt, *p);
+                PipelinePayload::AvPacket(new_pkt)
+            },
+            PipelinePayload::AvFrame(p) => unsafe {
+                let new_frame = av_frame_alloc();
+                av_frame_ref(new_frame, *p);
+                PipelinePayload::AvFrame(new_frame)
+            },
+            PipelinePayload::SourceInfo(i) => PipelinePayload::SourceInfo(i.clone()),
+        }
+    }
+}
+
+impl Drop for PipelinePayload {
+    fn drop(&mut self) {
+        match self {
+            PipelinePayload::Empty => {}
+            PipelinePayload::Bytes(_) => {}
+            PipelinePayload::AvPacket(p) => unsafe {
+                av_packet_free(p);
+            },
+            PipelinePayload::AvFrame(p) => unsafe {
+                av_frame_free(p);
+            },
+            PipelinePayload::SourceInfo(_) => {}
+        }
+    }
+}
+
 #[async_trait]
 pub trait PipelineStep {
     fn name(&self) -> String;
diff --git a/src/pipeline/runner.rs b/src/pipeline/runner.rs
index 63d0a86..59ac198 100644
--- a/src/pipeline/runner.rs
+++ b/src/pipeline/runner.rs
@@ -1,15 +1,19 @@
+use std::ops::{Add, AddAssign};
+use std::time::{Duration, Instant};
+
+use anyhow::Error;
+use log::info;
+use tokio::sync::broadcast;
+use tokio::sync::mpsc::{unbounded_channel, UnboundedReceiver};
+
 use crate::decode::Decoder;
-use crate::demux::info::{DemuxStreamInfo, StreamChannelType};
 use crate::demux::Demuxer;
+use crate::demux::info::{DemuxStreamInfo, StreamChannelType};
 use crate::egress::hls::HlsEgress;
 use crate::encode::Encoder;
 use crate::pipeline::{EgressType, PipelineConfig, PipelinePayload, PipelineStep};
 use crate::scale::Scaler;
 use crate::variant::VariantStream;
-use anyhow::Error;
-use log::info;
-use tokio::sync::broadcast;
-use tokio::sync::mpsc::{unbounded_channel, UnboundedReceiver};
 
 struct ScalerEncoder {
     pub scaler: Scaler,
@@ -24,6 +28,9 @@ pub struct PipelineRunner {
     scalers: Vec<ScalerEncoder>,
     encoders: Vec<Encoder<broadcast::Receiver<PipelinePayload>>>,
     egress: Vec<HlsEgress>,
+    started: Instant,
+    frame_no: u64,
+    stream_info: Option<DemuxStreamInfo>,
 }
 
 impl PipelineRunner {
@@ -38,14 +45,39 @@ impl PipelineRunner {
             scalers: vec![],
             encoders: vec![],
             egress: vec![],
+            started: Instant::now(),
+            frame_no: 0,
+            stream_info: None,
         }
     }
 
     pub fn run(&mut self) -> Result<(), Error> {
+        if let Some(info) = &self.stream_info {
+            if let Some(v_stream) = info
+                .channels
+                .iter()
+                .find(|s| s.channel_type == StreamChannelType::Video)
+            {
+                let duration = self.frame_no as f64 / v_stream.fps as f64;
+                let target_time = self.started.add(Duration::from_secs_f64(duration));
+                let now = Instant::now();
+                if now < target_time {
+                    let poll_sleep = target_time - now;
+                    std::thread::sleep(poll_sleep);
+                }
+            }
+        }
         if let Some(cfg) = self.demuxer.process()? {
             self.configure_pipeline(cfg)?;
         }
-        self.decoder.process()?;
+        let frames = self.decoder.process()?;
+        if let Some(v) = self.frame_no.checked_add(frames as u64) {
+            self.frame_no = v;
+        } else {
+            panic!("Frame number overflowed, maybe you need a bigger number!");
+        }
+
+        // video scalar-encoder chains
         for sw in &mut self.scalers {
             sw.scaler.process()?;
             sw.encoder.process()?;
@@ -53,15 +85,22 @@ impl PipelineRunner {
                 eg.process()?;
             }
         }
+        // audio encoder chains
+        for enc in &mut self.encoders {
+            enc.process()?;
+            for eg in &mut self.egress {
+                eg.process()?;
+            }
+        }
         Ok(())
     }
 
     fn configure_pipeline(&mut self, info: DemuxStreamInfo) -> Result<(), Error> {
-        // configure scalers
-        if self.scalers.len() != 0 {
+        if self.stream_info.is_some() {
             return Err(Error::msg("Pipeline already configured!"));
         }
         info!("Configuring pipeline {:?}", info);
+        self.stream_info = Some(info.clone());
 
         let video_stream = info
             .channels
@@ -100,7 +139,7 @@ impl PipelineRunner {
                                     return Err(Error::msg(format!(
                                         "Variant config not supported {:?}",
                                         c
-                                    )))
+                                    )));
                                 }
                             }
                         }
@@ -109,6 +148,11 @@ impl PipelineRunner {
                 }
             }
         }
-        Ok(())
+
+        if self.egress.len() == 0 {
+            Err(Error::msg("No egress config, pipeline misconfigured!"))
+        } else {
+            Ok(())
+        }
     }
 }
diff --git a/src/scale/mod.rs b/src/scale/mod.rs
index da32d63..36129c3 100644
--- a/src/scale/mod.rs
+++ b/src/scale/mod.rs
@@ -3,14 +3,14 @@ use std::ptr;
 
 use anyhow::Error;
 use ffmpeg_sys_next::{
-    av_frame_alloc, av_frame_copy_props, av_frame_unref, AVFrame, SWS_BILINEAR, sws_getContext,
-    sws_scale_frame, SwsContext,
+    av_buffer_ref, av_frame_alloc, av_frame_copy_props, av_frame_unref, AVBufferRef,
+    AVFrame, SWS_BILINEAR, sws_getContext, sws_scale_frame, SwsContext,
 };
 use tokio::sync::broadcast;
 use tokio::sync::mpsc::UnboundedSender;
 
 use crate::pipeline::PipelinePayload;
-use crate::utils::get_ffmpeg_error_msg;
+use crate::utils::{get_ffmpeg_error_msg, video_variant_id_ref};
 use crate::variant::VideoVariant;
 
 pub struct Scaler {
@@ -18,9 +18,11 @@ pub struct Scaler {
     ctx: *mut SwsContext,
     chan_in: broadcast::Receiver<PipelinePayload>,
     chan_out: UnboundedSender<PipelinePayload>,
+    var_id_ref: *mut AVBufferRef,
 }
 
 unsafe impl Send for Scaler {}
+
 unsafe impl Sync for Scaler {}
 
 impl Scaler {
@@ -29,11 +31,13 @@ impl Scaler {
         chan_out: UnboundedSender<PipelinePayload>,
         variant: VideoVariant,
     ) -> Self {
+        let id_ref = video_variant_id_ref(&variant);
         Self {
             chan_in,
             chan_out,
             variant,
             ctx: ptr::null_mut(),
+            var_id_ref: id_ref,
         }
     }
 
@@ -78,6 +82,8 @@ impl Scaler {
         (*dst_frame).time_base = (*frame).time_base;
         (*dst_frame).pts = (*frame).pts;
         (*dst_frame).pkt_dts = (*frame).pkt_dts;
+        (*dst_frame).opaque_ref = av_buffer_ref(self.var_id_ref);
+
         self.chan_out.send(PipelinePayload::AvFrame(dst_frame))?;
         Ok(())
     }
diff --git a/src/utils.rs b/src/utils.rs
index c60c608..1d0b61c 100644
--- a/src/utils.rs
+++ b/src/utils.rs
@@ -1,6 +1,11 @@
-use ffmpeg_sys_next::av_make_error_string;
 use std::ffi::CStr;
 
+use anyhow::Error;
+use ffmpeg_sys_next::{av_buffer_allocz, av_make_error_string, AVBufferRef, memcpy};
+use uuid::{Bytes, Uuid};
+
+use crate::variant::{VariantStream, VideoVariant};
+
 pub fn get_ffmpeg_error_msg(ret: libc::c_int) -> String {
     unsafe {
         const BUF_SIZE: usize = 512;
@@ -9,3 +14,48 @@ pub fn get_ffmpeg_error_msg(ret: libc::c_int) -> String {
         String::from(CStr::from_ptr(buf.as_ptr()).to_str().unwrap())
     }
 }
+
+pub fn variant_id_ref(var: &VariantStream) -> Result<*mut AVBufferRef, Error> {
+    unsafe {
+        match var {
+            VariantStream::Audio(va) => {
+                let buf = av_buffer_allocz(16);
+                memcpy(
+                    (*buf).data as *mut libc::c_void,
+                    va.id.as_bytes().as_ptr() as *const libc::c_void,
+                    16,
+                );
+                Ok(buf)
+            }
+            VariantStream::Video(vv) => {
+                let buf = av_buffer_allocz(16);
+                memcpy(
+                    (*buf).data as *mut libc::c_void,
+                    vv.id.as_bytes().as_ptr() as *const libc::c_void,
+                    16,
+                );
+                Ok(buf)
+            }
+            _ => return Err(Error::msg("Cannot assign pkt stream index")),
+        }
+    }
+}
+
+pub fn video_variant_id_ref(var: &VideoVariant) -> *mut AVBufferRef {
+    unsafe {
+        let buf = av_buffer_allocz(16);
+        memcpy(
+            (*buf).data as *mut libc::c_void,
+            var.id.as_bytes().as_ptr() as *const libc::c_void,
+            16,
+        );
+        buf
+    }
+}
+
+pub fn id_ref_to_uuid(buf: *mut AVBufferRef) -> Uuid {
+    unsafe {
+        let binding = Bytes::from(*((*buf).data as *const [u8; 16]));
+        Uuid::from_bytes_ref(&binding).clone()
+    }
+}
diff --git a/src/variant.rs b/src/variant.rs
index 8058f72..ed9b82d 100644
--- a/src/variant.rs
+++ b/src/variant.rs
@@ -1,5 +1,5 @@
-use crate::fraction::Fraction;
 use std::fmt::{Display, Formatter};
+
 use serde::{Deserialize, Serialize};
 use uuid::Uuid;
 
@@ -71,7 +71,7 @@ impl Display for VideoVariant {
 pub struct AudioVariant {
     /// Unique ID of this variant
     pub id: Uuid,
-    
+
     /// Source video stream to use for this variant
     pub src_index: usize,
 
@@ -89,6 +89,9 @@ pub struct AudioVariant {
 
     /// Sample rate
     pub sample_rate: usize,
+
+    /// Sample format as ffmpeg sample format string
+    pub sample_fmt: String,
 }
 
 impl Display for AudioVariant {
@@ -102,3 +105,14 @@ impl Display for AudioVariant {
         )
     }
 }
+
+impl VariantStream {
+    pub fn src_index(&self) -> usize {
+        match self {
+            VariantStream::Video(v) => v.src_index,
+            VariantStream::Audio(v) => v.src_index,
+            VariantStream::CopyVideo(v) => v.clone(),
+            VariantStream::CopyAudio(v) => v.clone(),
+        }
+    }
+}
diff --git a/src/webhook.rs b/src/webhook.rs
index 6feab1a..f9f3847 100644
--- a/src/webhook.rs
+++ b/src/webhook.rs
@@ -1,10 +1,12 @@
+use std::fmt::Display;
+
+use ffmpeg_sys_next::{AV_LEVEL_UNKNOWN, AV_PROFILE_H264_HIGH};
+use ffmpeg_sys_next::AVCodecID::{AV_CODEC_ID_AAC, AV_CODEC_ID_H264};
+use uuid::Uuid;
+
 use crate::ingress::ConnectionInfo;
 use crate::pipeline::{EgressType, HLSEgressConfig, PipelineConfig};
 use crate::variant::{AudioVariant, VariantStream, VideoVariant};
-use ffmpeg_sys_next::AVCodecID::{AV_CODEC_ID_AAC, AV_CODEC_ID_H264};
-use ffmpeg_sys_next::{AV_LEVEL_UNKNOWN, AV_PROFILE_H264_HIGH};
-use std::fmt::Display;
-use uuid::Uuid;
 
 #[derive(Clone)]
 pub struct Webhook {
@@ -54,6 +56,7 @@ impl Webhook {
             codec: 86018,
             channels: 2,
             sample_rate: 44_100,
+            sample_fmt: "fltp".to_owned(),
         };
 
         let audio_var_2 = AudioVariant {
@@ -64,6 +67,7 @@ impl Webhook {
             codec: 86018,
             channels: 2,
             sample_rate: 44_100,
+            sample_fmt: "fltp".to_owned(),
         };
 
         Ok(PipelineConfig {