Auto merge of #614 - lennart/CLOUD-154, r=TroyKomodo

feat(ingest): rtmp + rtmps server
This PR adds an RTMP and an RTMPS server to ingest. The actual handling of events and data is left unimplemented.
CLOUD-154

Requested-by: TroyKomodo <[email protected]>
Reviewed-by: TroyKomodo <[email protected]>

Author: scuffle-brawl[bot]

Cargo.lock

@@ -9,6 +9,7 @@ deps = [
     "//crates/context",
     "//crates/settings",
     "//crates/signal",
+    "//crates/rtmp",
     "//cloud/video/ingest/traits",
 ]
 

cloud/video/ingest/BUILD.bazel

@@ -17,18 +17,21 @@ path = "bin/standalone/main.rs"
 unexpected_cfgs = { level = "warn", check-cfg = ['cfg(coverage_nightly)'] }
 
 [dependencies]
-anyhow = "1.0.98"
+anyhow = "1"
 ingest-traits = { path = "./traits", package = "scufflecloud-ingest-traits" }
 scuffle-bootstrap = { path = "../../../crates/bootstrap" }
 scuffle-bootstrap-telemetry = { features = ["opentelemetry-logs", "opentelemetry-traces"], path = "../../../crates/bootstrap-telemetry" }
 scuffle-context = { path = "../../../crates/context" }
+scuffle-rtmp = { path = "../../../crates/rtmp" }
 scuffle-settings = { features = ["all-formats", "bootstrap"], path = "../../../crates/settings" }
 scuffle-signal = { features = ["bootstrap"], path = "../../../crates/signal" }
-serde = "1.0.219"
-serde_derive = "1.0.219"
-smart-default = "0.7.1"
-tracing = "0.1.41"
-tracing-subscriber = { version = "0.3.20", features = ["env-filter"] }
+serde = "1"
+serde_derive = "1"
+smart-default = "0.7"
+tokio = "1"
+tokio-rustls = "0.26"
+tracing = "0.1"
+tracing-subscriber = { version = "0.3", features = ["env-filter"] }
 
 [package.metadata.sync-readme.badges]
 docs-rs = false

cloud/video/ingest/Cargo.toml

@@ -1,4 +1,5 @@
 use std::net::SocketAddr;
+use std::path::PathBuf;
 
 #[derive(serde_derive::Deserialize, smart_default::SmartDefault, Debug, Clone)]
 #[serde(default)]
@@ -7,11 +8,26 @@ pub(crate) struct Config {
     pub service_name: String,
     #[default = "info"]
     pub level: String,
+    #[default("[::]:1935".parse().unwrap())]
+    pub rtmp_bind: SocketAddr,
+    pub rtmps: Option<RtmpsConfig>,
     pub telemetry: Option<TelemetryConfig>,
 }
 
 scuffle_settings::bootstrap!(Config);
 
+#[derive(serde_derive::Deserialize, smart_default::SmartDefault, Debug, Clone)]
+pub(crate) struct RtmpsConfig {
+    // https://github.com/obsproject/obs-studio/blob/0b1229632063a13dfd26cf1cd9dd43431d8c68f6/plugins/obs-outputs/librtmp/rtmp.c#L718
+    // Yes, RTMPS' default port is 443
+    #[default("[::]:443".parse().unwrap())]
+    pub bind: SocketAddr,
+    #[default = "rtmps_certs.pem"]
+    pub cert_chain_path: PathBuf,
+    #[default = "rtmps_key.pem"]
+    pub key_path: PathBuf,
+}
+
 #[derive(serde_derive::Deserialize, smart_default::SmartDefault, Debug, Clone)]
 pub(crate) struct TelemetryConfig {
     #[default("[::1]:4317".parse().unwrap())]

cloud/video/ingest/bin/standalone/config.rs

@@ -7,9 +7,12 @@
 
 use std::sync::Arc;
 
+use anyhow::Context;
 use scuffle_bootstrap_telemetry::opentelemetry;
 use scuffle_bootstrap_telemetry::opentelemetry_sdk::logs::SdkLoggerProvider;
 use scuffle_bootstrap_telemetry::opentelemetry_sdk::trace::SdkTracerProvider;
+use tokio_rustls::rustls::pki_types::pem::PemObject;
+use tokio_rustls::rustls::pki_types::{CertificateDer, PrivateKeyDer};
 use tracing_subscriber::Layer;
 use tracing_subscriber::layer::SubscriberExt;
 use tracing_subscriber::util::SubscriberInitExt;
@@ -18,9 +21,39 @@ mod config;
 
 struct Global {
     config: config::Config,
+    rtmps: Option<RtmpsGlobal>,
     open_telemetry: opentelemetry::OpenTelemetry,
 }
 
+struct RtmpsGlobal {
+    bind: std::net::SocketAddr,
+    rustls_config: Arc<tokio_rustls::rustls::ServerConfig>,
+}
+
+impl ingest_traits::RtmpConfigInterface for RtmpsGlobal {
+    fn rtmps_bind(&self) -> std::net::SocketAddr {
+        self.bind
+    }
+
+    fn rtmps_rustls_server_config(&self) -> Arc<tokio_rustls::rustls::ServerConfig> {
+        self.rustls_config.clone()
+    }
+}
+
+impl ingest_traits::ConfigInterface for Global {
+    fn rtmp_bind(&self) -> std::net::SocketAddr {
+        self.config.rtmp_bind
+    }
+}
+
+impl ingest_traits::RtmpsInterface for Global {
+    type RtmpsConfig = RtmpsGlobal;
+
+    fn rtmps_config(&self) -> Option<&Self::RtmpsConfig> {
+        self.rtmps.as_ref()
+    }
+}
+
 impl ingest_traits::Global for Global {}
 
 impl scuffle_signal::SignalConfig for Global {}
@@ -54,14 +87,38 @@ impl scuffle_bootstrap::Global for Global {
             )
             .init();
 
+        let rtmps = if let Some(rtmps) = config.rtmps.as_ref() {
+            let cert_chain = CertificateDer::pem_file_iter(&rtmps.cert_chain_path)
+                .context("load RTMPS cert chain file")?
+                .collect::<Result<Vec<_>, _>>()
+                .context("load RTMPS cert chain")?;
+            let key_der = PrivateKeyDer::from_pem_file(&rtmps.key_path).context("load RTMPS private key")?;
+
+            let rustls_config = tokio_rustls::rustls::ServerConfig::builder()
+                .with_no_client_auth()
+                .with_single_cert(cert_chain, key_der)
+                .context("create RTMPS rustls server config")?;
+
+            Some(RtmpsGlobal {
+                bind: rtmps.bind,
+                rustls_config: Arc::new(rustls_config),
+            })
+        } else {
+            None
+        };
+
         let tracer = SdkTracerProvider::default();
         opentelemetry::global::set_tracer_provider(tracer.clone());
 
         let logger = SdkLoggerProvider::builder().build();
 
         let open_telemetry = opentelemetry::OpenTelemetry::new().with_traces(tracer).with_logs(logger);
 
-        Ok(Arc::new(Self { config, open_telemetry }))
+        Ok(Arc::new(Self {
+            config,
+            rtmps,
+            open_telemetry,
+        }))
     }
 }