feat(rtmp): add advertised publish workflow

pull/439/head
zijiren233 1 month ago
parent c7d885c4d8
commit 4e7db1cdcf
No known key found for this signature in database
GPG Key ID: 534E082AAA9B39DC

@ -28,7 +28,7 @@ CARGO_WORKSPACE_BUILD_ARGS ?= $(CARGO_BUILD_ARGS) $(CARGO_WORKSPACE_ARGS)
CARGO_WORKSPACE_ALL_TARGETS_BUILD_ARGS ?= $(CARGO_WORKSPACE_BUILD_ARGS) $(CARGO_ALL_TARGETS_ARGS)
NEXTEST_STATUS_ARGS ?= --status-level slow --final-status-level slow
DEV_FEATURES ?=
DEV_CARGO_FEATURE_ARGS := $(if $(strip $(DEV_FEATURES)),--no-default-features --features "$(DEV_FEATURES)",)
DEV_CARGO_FEATURE_ARGS := $(if $(strip $(DEV_FEATURES)),--features "$(DEV_FEATURES)",)
RELEASE_FEATURES ?=
RELEASE_CARGO_FEATURE_ARGS := $(if $(strip $(RELEASE_FEATURES)),--features "$(RELEASE_FEATURES)",)
FEATURE_CHECK_ARGS ?=

@ -159,10 +159,10 @@ User-level 2FA and notification preferences are user preferences. Provider insta
| Key | Type | Default | Meaning |
| --- | --- | --- | --- |
| `rtmp.customPublishHost` | string/null | `null` | Optional custom publish host returned to stream publishers |
| `rtmp.advertiseAddress` | string/null | `null` | Public RTMP or RTMPS address containing only the scheme, host, and optional port; the server appends the room path |
| `rtmp.tsDisguisedAsPng` | bool | `false` | Disguise TS segments as PNG paths or responses |
Set it with `{"settings":{"rtmp":{"customPublishHost":"rtmp://live.example.com"}},"updateMask":"rtmp.customPublishHost"}`. Clear it with `{"settings":{"rtmp":{}},"updateMask":"rtmp.customPublishHost"}`.
Set it with `{"settings":{"rtmp":{"advertiseAddress":"rtmps://live.example.com:443"}},"updateMask":"rtmp.advertiseAddress"}`. Clear it with `{"settings":{"rtmp":{}},"updateMask":"rtmp.advertiseAddress"}`. The address cannot contain credentials, a path, a query, or a fragment.
### Email

@ -159,10 +159,10 @@ Apple 原生授权使用 `native=true`,省略 `redirectUrl`,由 iOS 和 Mac
| Key | 类型 | 默认值 | 说明 |
| --- | --- | --- | --- |
| `rtmp.customPublishHost` | string/null | `null` | 可选的自定义发布 host |
| `rtmp.advertiseAddress` | string/null | `null` | 对外展示的 RTMP 或 RTMPS 地址,仅包含协议、主机和可选端口;服务器会追加房间业务路径 |
| `rtmp.tsDisguisedAsPng` | bool | `false` | 是否把 TS 片段伪装为 PNG 路径或响应形式 |
设置请求使用 `{"settings":{"rtmp":{"customPublishHost":"rtmp://live.example.com"}},"updateMask":"rtmp.customPublishHost"}`。清除请求使用 `{"settings":{"rtmp":{}},"updateMask":"rtmp.customPublishHost"}`
设置请求使用 `{"settings":{"rtmp":{"advertiseAddress":"rtmps://live.example.com:443"}},"updateMask":"rtmp.advertiseAddress"}`。清除请求使用 `{"settings":{"rtmp":{}},"updateMask":"rtmp.advertiseAddress"}`。该地址不能包含凭据、路径、查询参数或片段
### Email

@ -80,7 +80,7 @@ fn runtime_settings_patch_from_admin_proto_with_oauth2(
.map(|patch| oauth2_settings_patch_from_admin_proto(patch, current_oauth2))
.transpose()?,
rtmp: settings.rtmp.map(|patch| RtmpSettingsPatch {
custom_publish_host: patch.custom_publish_host.map(OptionalConfigPatch::Set),
advertise_address: patch.advertise_address.map(OptionalConfigPatch::Set),
ts_disguised_as_png: patch.ts_disguised_as_png,
}),
email: settings.email.map(email_settings_patch_from_admin_proto),
@ -217,8 +217,8 @@ pub fn runtime_settings_replacement_patch_from_admin_proto(
allowed_redirect_urls: Some(oauth2.allowed_redirect_urls),
}),
rtmp: Some(RtmpSettingsPatch {
custom_publish_host: Some(match rtmp.custom_publish_host {
Some(host) => OptionalConfigPatch::Set(host),
advertise_address: Some(match rtmp.advertise_address {
Some(address) => OptionalConfigPatch::Set(address),
None => OptionalConfigPatch::Clear,
}),
ts_disguised_as_png: Some(rtmp.ts_disguised_as_png),
@ -365,7 +365,7 @@ fn select_runtime_settings_patch(
"oauth2.allowedRedirectUrls" | "oauth2.allowed_redirect_urls" => {
select_required!(oauth2, allowed_redirect_urls, path);
}
"rtmp.custom_publish_host" => select_optional!(rtmp, custom_publish_host),
"rtmp.advertise_address" => select_optional!(rtmp, advertise_address),
"rtmp.ts_disguised_as_png" => {
select_required!(rtmp, ts_disguised_as_png, path);
}

@ -243,15 +243,16 @@ impl AdminApiImpl {
"Publish key service is not available on this server.".to_string(),
)
})?;
let response = crate::impls::client::stream::issue_room_publish_key(
let advertise_address = crate::impls::client::stream::rtmp_advertise_address(
self.runtime_settings_store.as_deref(),
)?;
let response = crate::impls::client::stream::RoomPublishKeyIssuer::new(
publish_key_service,
&self.runtime_settings,
advertise_address.as_deref(),
&self.public_id_codec,
room_id_value,
media_id_value,
actor_user_id,
options,
)?;
)
.issue(room_id_value, media_id_value, actor_user_id, options)?;
tracing::info!(
room_id,

@ -69,7 +69,7 @@ pub struct OAuth2SettingsPatch {
#[derive(Debug, Clone, Default)]
pub struct RtmpSettingsPatch {
pub custom_publish_host: Option<OptionalConfigPatch<String>>,
pub advertise_address: Option<OptionalConfigPatch<String>>,
pub ts_disguised_as_png: Option<bool>,
}
@ -590,7 +590,7 @@ impl AdminApiImpl {
.collect(),
}),
rtmp: Some(synctv_proto::admin::RtmpSettings {
custom_publish_host: settings.rtmp.custom_publish_host,
advertise_address: settings.rtmp.advertise_address,
ts_disguised_as_png: settings.rtmp.ts_disguised_as_png,
}),
email: Some(synctv_proto::admin::EmailSettings {
@ -744,12 +744,12 @@ impl AdminApiImpl {
}
if let Some(rtmp) = patch.rtmp {
if let Some(patch) = rtmp.custom_publish_host {
current.rtmp.custom_publish_host = match patch {
if let Some(patch) = rtmp.advertise_address {
current.rtmp.advertise_address = match patch {
OptionalConfigPatch::Set(value) => Some(value),
OptionalConfigPatch::Clear => None,
};
update_mask.rtmp.custom_publish_host = true;
update_mask.rtmp.advertise_address = true;
}
if let Some(value) = rtmp.ts_disguised_as_png {
current.rtmp.ts_disguised_as_png = value;

@ -122,7 +122,7 @@ fn test_runtime_settings() -> synctv_core::service::RuntimeSettings {
allowed_redirect_urls: Vec::new(),
},
rtmp: synctv_core::service::RtmpRuntimeSettings {
custom_publish_host: None,
advertise_address: None,
ts_disguised_as_png: false,
},
email: synctv_core::service::EmailRuntimeSettings {
@ -2955,28 +2955,28 @@ fn test_runtime_settings_snapshot_replacement_round_trips_all_sections() -> Test
}
#[test]
fn test_optional_rtmp_publish_host_supports_set_and_clear() -> TestResult {
fn test_optional_rtmp_advertise_address_supports_set_and_clear() -> TestResult {
let current = test_runtime_settings();
let set_patch = crate::admin_settings_mapping::runtime_settings_patch_from_admin_proto(
runtime_settings_request(
synctv_proto::admin::RuntimeSettingsPatch {
rtmp: Some(synctv_proto::admin::RtmpSettingsPatch {
custom_publish_host: Some("rtmp://live.example.com".to_string()),
advertise_address: Some("rtmps://live.example.com".to_string()),
..Default::default()
}),
..Default::default()
},
&["rtmp.custom_publish_host"],
&["rtmp.advertise_address"],
),
)
.map_err(|error| test_error(format!("{error:?}")))?;
let set_result = AdminApiImpl::apply_runtime_settings_patch(current, set_patch)
.map_err(|error| test_error(format!("{error:?}")))?;
assert_eq!(
set_result.settings.rtmp.custom_publish_host.as_deref(),
Some("rtmp://live.example.com")
set_result.settings.rtmp.advertise_address.as_deref(),
Some("rtmps://live.example.com")
);
assert!(set_result.update_mask.rtmp.custom_publish_host);
assert!(set_result.update_mask.rtmp.advertise_address);
let clear_patch = crate::admin_settings_mapping::runtime_settings_patch_from_admin_proto(
runtime_settings_request(
@ -2984,14 +2984,14 @@ fn test_optional_rtmp_publish_host_supports_set_and_clear() -> TestResult {
rtmp: Some(synctv_proto::admin::RtmpSettingsPatch::default()),
..Default::default()
},
&["rtmp.custom_publish_host"],
&["rtmp.advertise_address"],
),
)
.map_err(|error| test_error(format!("{error:?}")))?;
let clear_result = AdminApiImpl::apply_runtime_settings_patch(set_result.settings, clear_patch)
.map_err(|error| test_error(format!("{error:?}")))?;
assert_eq!(clear_result.settings.rtmp.custom_publish_host, None);
assert!(clear_result.update_mask.rtmp.custom_publish_host);
assert_eq!(clear_result.settings.rtmp.advertise_address, None);
assert!(clear_result.update_mask.rtmp.advertise_address);
Ok(())
}

@ -2093,7 +2093,7 @@ impl ClientApiImpl {
webauthn_signup_need_review: s.webauthn_signup_need_review,
enable_guest: s.enable_guest,
ts_disguised_as_png: s.ts_disguised_as_png,
custom_publish_host: s.custom_publish_host,
advertise_address: s.advertise_address,
email_whitelist_enabled: s.email_whitelist_enabled,
email_whitelist_domains: s.email_whitelist_domains,
})

@ -108,10 +108,41 @@ pub(crate) fn ensure_room_accepts_live_publish(room: &Room) -> Result<(), ApiErr
Ok(())
}
fn build_publish_rtmp_url(runtime_settings: &crate::ApiRuntimeSettings, room_id: &str) -> String {
fn build_publish_rtmp_url(
runtime_settings: &crate::ApiRuntimeSettings,
advertise_address: Option<&str>,
room_id: &str,
) -> Result<String, ApiError> {
if let Some(address) = advertise_address {
let mut url =
synctv_core::service::parse_rtmp_advertise_address(address).map_err(ApiError::from)?;
url.path_segments_mut()
.map_err(|()| {
ApiError::InvalidInput(
"rtmp.advertise_address must support path segments".to_string(),
)
})?
.pop_if_empty()
.push(room_id);
return Ok(url.to_string());
}
let rtmp_host = runtime_settings.public_rtmp_host();
let rtmp_port = runtime_settings.livestream.rtmp_port;
format!("rtmp://{rtmp_host}:{rtmp_port}/{room_id}")
Ok(format!("rtmp://{rtmp_host}:{rtmp_port}/{room_id}"))
}
pub(crate) fn rtmp_advertise_address(
settings: Option<&synctv_core::service::RuntimeSettingsStore>,
) -> Result<Option<String>, ApiError> {
settings.map_or(Ok(None), |settings| {
settings
.rtmp
.advertise_address
.get()
.map(|value| value.0)
.map_err(ApiError::from)
})
}
async fn filter_usable_stream_media_ids(
@ -158,44 +189,74 @@ pub(crate) fn publish_key_options(
}))
}
pub(crate) fn issue_room_publish_key(
publish_key_service: &dyn synctv_core::service::StreamingPublishKeyService,
runtime_settings: &crate::ApiRuntimeSettings,
public_id_codec: &synctv_adapter::PublicIdCodec,
room_id: RoomId,
media_id: MediaId,
actor_user_id: &UserId,
options: Option<PublishKeyOptions>,
) -> Result<CreateRoomPublishKeyResponse, ApiError> {
let publish_key = match options {
Some(options) => publish_key_service.generate_publish_key_with_options(
&room_id,
&media_id,
actor_user_id,
options,
),
None => publish_key_service.generate_publish_key(&room_id, &media_id, actor_user_id),
pub(crate) struct RoomPublishKeyIssuer<'a> {
publish_key_service: &'a dyn synctv_core::service::StreamingPublishKeyService,
runtime_settings: &'a crate::ApiRuntimeSettings,
advertise_address: Option<&'a str>,
public_id_codec: &'a synctv_adapter::PublicIdCodec,
}
impl<'a> RoomPublishKeyIssuer<'a> {
pub(crate) const fn new(
publish_key_service: &'a dyn synctv_core::service::StreamingPublishKeyService,
runtime_settings: &'a crate::ApiRuntimeSettings,
advertise_address: Option<&'a str>,
public_id_codec: &'a synctv_adapter::PublicIdCodec,
) -> Self {
Self {
publish_key_service,
runtime_settings,
advertise_address,
public_id_codec,
}
}
pub(crate) fn issue(
&self,
room_id: RoomId,
media_id: MediaId,
actor_user_id: &UserId,
options: Option<PublishKeyOptions>,
) -> Result<CreateRoomPublishKeyResponse, ApiError> {
let publish_key = match options {
Some(options) => self.publish_key_service.generate_publish_key_with_options(
&room_id,
&media_id,
actor_user_id,
options,
),
None => {
self.publish_key_service
.generate_publish_key(&room_id, &media_id, actor_user_id)
}
}
.map_err(|error| ApiError::InvalidInput(error.to_string()))?;
let room_id = self
.public_id_codec
.encode_room_id(room_id)
.map_err(|error| ApiError::Internal(format!("Failed to encode room id: {error}")))?;
let media_id = self
.public_id_codec
.encode_media_id(media_id)
.map_err(|error| ApiError::Internal(format!("Failed to encode media id: {error}")))?;
let stream_key = format!("{media_id}?token={}", publish_key.token);
Ok(CreateRoomPublishKeyResponse {
publish_key: publish_key.token,
rtmp_url: build_publish_rtmp_url(
self.runtime_settings,
self.advertise_address,
&room_id,
)?,
stream_key,
expires_at: publish_key.expires_at,
r#type: match publish_key.key_type {
CorePublishKeyType::SingleUse => PublishKeyType::SingleUse as i32,
CorePublishKeyType::Expiring => PublishKeyType::Expiring as i32,
CorePublishKeyType::Permanent => PublishKeyType::Permanent as i32,
},
})
}
.map_err(|error| ApiError::InvalidInput(error.to_string()))?;
let room_id = public_id_codec
.encode_room_id(room_id)
.map_err(|error| ApiError::Internal(format!("Failed to encode room id: {error}")))?;
let media_id = public_id_codec
.encode_media_id(media_id)
.map_err(|error| ApiError::Internal(format!("Failed to encode media id: {error}")))?;
let stream_key = format!("{media_id}?token={}", publish_key.token);
Ok(CreateRoomPublishKeyResponse {
publish_key: publish_key.token,
rtmp_url: build_publish_rtmp_url(runtime_settings, &room_id),
stream_key,
expires_at: publish_key.expires_at,
r#type: match publish_key.key_type {
CorePublishKeyType::SingleUse => PublishKeyType::SingleUse as i32,
CorePublishKeyType::Expiring => PublishKeyType::Expiring as i32,
CorePublishKeyType::Permanent => PublishKeyType::Permanent as i32,
},
})
}
pub async fn fetch_stream_info(
@ -280,15 +341,14 @@ impl ClientApiImpl {
.publish_key_service
.as_deref()
.ok_or_else(publish_key_service_unavailable_error)?;
issue_room_publish_key(
let advertise_address = rtmp_advertise_address(self.runtime_settings_store.as_deref())?;
RoomPublishKeyIssuer::new(
publish_key_service,
&self.runtime_settings,
advertise_address.as_deref(),
&self.public_id_codec,
rid,
media_id,
&uid,
options,
)
.issue(rid, media_id, &uid, options)
}
pub async fn list_room_streams(
@ -414,8 +474,8 @@ impl ClientApiImpl {
#[cfg(test)]
mod tests {
use super::{
build_room_streams_request, build_room_streams_response, ensure_room_accepts_live_publish,
filter_usable_stream_media_ids, publish_key_options,
build_publish_rtmp_url, build_room_streams_request, build_room_streams_response,
ensure_room_accepts_live_publish, filter_usable_stream_media_ids, publish_key_options,
};
use crate::impls::ApiError;
@ -567,6 +627,35 @@ mod tests {
Ok(())
}
#[test]
fn publish_url_prefers_runtime_advertise_address() -> TestResult {
let runtime_settings = crate::ApiRuntimeSettings::default();
let url = api_ok(build_publish_rtmp_url(
&runtime_settings,
Some("rtmps://live.example.com:443"),
"room_AbC123",
))?;
assert_eq!(url, "rtmps://live.example.com:443/room_AbC123");
Ok(())
}
#[test]
fn publish_url_falls_back_to_static_livestream_address() -> TestResult {
let mut runtime_settings = crate::ApiRuntimeSettings::default();
runtime_settings.livestream.public_rtmp_host = "live.internal".to_string();
runtime_settings.livestream.rtmp_port = 1936;
let url = api_ok(build_publish_rtmp_url(
&runtime_settings,
None,
"room_AbC123",
))?;
assert_eq!(url, "rtmp://live.internal:1936/room_AbC123");
Ok(())
}
#[test]
fn build_room_streams_request_normalizes_defaults() -> TestResult {
let req = api_ok(build_room_streams_request(

@ -24,7 +24,7 @@ use axum::{
response::IntoResponse,
};
use futures::{SinkExt, StreamExt};
use prost_reflect::{DynamicMessage, ReflectMessage};
use prost_reflect::{DeserializeOptions, DynamicMessage, ReflectMessage};
use std::convert::Infallible;
use std::future::Future;
use std::sync::{
@ -743,7 +743,8 @@ impl WebSocketMessageSender {
fn decode_client_message_json(text: &str) -> Result<ClientMessage, String> {
let descriptor = ClientMessage::default().descriptor();
let mut deserializer = serde_json::Deserializer::from_str(text);
let dynamic = DynamicMessage::deserialize(descriptor, &mut deserializer)
let options = DeserializeOptions::new().deny_unknown_fields(false);
let dynamic = DynamicMessage::deserialize_with_options(descriptor, &mut deserializer, &options)
.map_err(|e| format!("Failed to decode JSON message: {e}"))?;
deserializer
.end()

@ -276,6 +276,40 @@ fn test_websocket_json_uses_integer_enum_values() -> TestResult {
Ok(())
}
#[test]
fn test_websocket_json_ignores_fields_from_newer_clients() -> TestResult {
let message = decode_client_message_json(
r#"{
"observeResource": {
"observeId": "playback",
"playback": {
"playbackClientProfile": {
"profileVersion": 2,
"supportsP2pMediaLoader": true
}
}
}
}"#,
)
.map_err(test_error)?;
let Some(synctv_proto::client::client_message::Message::ObserveResource(observe)) =
message.message
else {
return Err(test_error("expected observe resource message"));
};
let Some(synctv_proto::client::observe_resource::Resource::Playback(playback)) =
observe.resource
else {
return Err(test_error("expected playback observation"));
};
let Some(profile) = playback.playback_client_profile else {
return Err(test_error("expected playback client profile"));
};
assert_eq!(profile.profile_version, 2);
Ok(())
}
#[test]
fn test_notification_requires_state_resync() {
let message = ServerMessage {

@ -34,17 +34,17 @@ use tracing::warn;
mod types;
pub use types::{
ChatRuntimeSettings, ConfiguredIceServer, CorsAllowedOrigins, CorsRuntimeSettings,
EmailRuntimeSettings, IceServerList, OAuth2AllowedRedirectUrls, OAuth2AppleProviderConfig,
OAuth2CasdoorProviderConfig, OAuth2DiscordProviderConfig, OAuth2FeishuProviderConfig,
OAuth2GiteeProviderConfig, OAuth2GithubProviderConfig, OAuth2GoogleProviderConfig,
OAuth2LogtoProviderConfig, OAuth2MicrosoftProviderConfig, OAuth2OidcProviderConfig,
OAuth2ProviderConfig, OAuth2ProviderConfigs, OAuth2ProviderPrivateConfig,
OAuth2QqProviderConfig, OAuth2RuntimeSettings, OAuth2SignupPolicy, OptionalRuntimeConfig,
PermissionRuntimeSettings, PermissionSet, PlaybackHistoryRuntimeSettings, PublicSettings,
RoomCreationRuntimeSettings, RoomDefaultsRuntimeSettings, RoomPasswordPolicy,
RtmpRuntimeSettings, RuntimeSettings, RuntimeSettingsUpdateMask, ServerRuntimeSettings,
UserRuntimeSettings, WebRtcRuntimeSettings,
parse_rtmp_advertise_address, ChatRuntimeSettings, ConfiguredIceServer, CorsAllowedOrigins,
CorsRuntimeSettings, EmailRuntimeSettings, IceServerList, OAuth2AllowedRedirectUrls,
OAuth2AppleProviderConfig, OAuth2CasdoorProviderConfig, OAuth2DiscordProviderConfig,
OAuth2FeishuProviderConfig, OAuth2GiteeProviderConfig, OAuth2GithubProviderConfig,
OAuth2GoogleProviderConfig, OAuth2LogtoProviderConfig, OAuth2MicrosoftProviderConfig,
OAuth2OidcProviderConfig, OAuth2ProviderConfig, OAuth2ProviderConfigs,
OAuth2ProviderPrivateConfig, OAuth2QqProviderConfig, OAuth2RuntimeSettings, OAuth2SignupPolicy,
OptionalRuntimeConfig, PermissionRuntimeSettings, PermissionSet,
PlaybackHistoryRuntimeSettings, PublicSettings, RoomCreationRuntimeSettings,
RoomDefaultsRuntimeSettings, RoomPasswordPolicy, RtmpRuntimeSettings, RuntimeSettings,
RuntimeSettingsUpdateMask, ServerRuntimeSettings, UserRuntimeSettings, WebRtcRuntimeSettings,
};
/// Maximum allowed value for `default_max_chat_messages` setting (0 = unlimited)
@ -210,15 +210,13 @@ setting!(
setting!(EnableGuestSetting, bool, "user.enable_guest", true);
setting!(
CustomPublishHostSetting,
RtmpAdvertiseAddressSetting,
OptionalRuntimeConfig<String>,
"rtmp.custom_publish_host",
"rtmp.advertise_address",
OptionalRuntimeConfig::default(),
|value: &OptionalRuntimeConfig<String>| -> crate::Result<()> {
if value.0.as_ref().is_some_and(|host| host.trim().is_empty()) {
return Err(crate::Error::InvalidInput(
"rtmp.custom_publish_host must be non-empty when configured".to_string(),
));
if let Some(address) = value.0.as_deref() {
parse_rtmp_advertise_address(address)?;
}
Ok(())
}
@ -581,7 +579,7 @@ pub struct OAuth2SettingsStore {
#[derive(Clone)]
pub struct RtmpSettingsStore {
pub custom_publish_host: CustomPublishHostSetting,
pub advertise_address: RtmpAdvertiseAddressSetting,
pub ts_disguised_as_png: TsDisguisedAsPngSetting,
}
@ -839,7 +837,7 @@ impl RuntimeSettingsStore {
};
let rtmp = RtmpSettingsStore {
custom_publish_host: CustomPublishHostSetting::new(storage.clone()),
advertise_address: RtmpAdvertiseAddressSetting::new(storage.clone()),
ts_disguised_as_png: TsDisguisedAsPngSetting::new(storage.clone()),
};
@ -1003,11 +1001,7 @@ impl RuntimeSettingsStore {
.0,
},
rtmp: RtmpRuntimeSettings {
custom_publish_host: self
.rtmp
.custom_publish_host
.get_from_snapshot(&snapshot)?
.0,
advertise_address: self.rtmp.advertise_address.get_from_snapshot(&snapshot)?.0,
ts_disguised_as_png: self.rtmp.ts_disguised_as_png.get_from_snapshot(&snapshot)?,
},
email: EmailRuntimeSettings {
@ -1206,9 +1200,9 @@ impl RuntimeSettingsStore {
)?;
Self::push_update_entry(
&mut entries,
update_mask.rtmp.custom_publish_host,
&self.rtmp.custom_publish_host,
&OptionalRuntimeConfig(settings.rtmp.custom_publish_host.clone()),
update_mask.rtmp.advertise_address,
&self.rtmp.advertise_address,
&OptionalRuntimeConfig(settings.rtmp.advertise_address.clone()),
)?;
Self::push_update_entry(
&mut entries,
@ -1412,7 +1406,7 @@ impl RuntimeSettingsStore {
enable_email: settings.email.enabled,
enable_webauthn: false,
ts_disguised_as_png: settings.rtmp.ts_disguised_as_png,
custom_publish_host: settings.rtmp.custom_publish_host,
advertise_address: settings.rtmp.advertise_address,
email_whitelist_enabled: settings.email.whitelist_enabled,
email_whitelist_domains,
})
@ -1823,14 +1817,14 @@ mod tests {
}
#[test]
fn test_public_settings_includes_nonempty_custom_publish_host() {
fn test_public_settings_includes_nonempty_rtmp_advertise_address() {
let mut settings = PublicSettings::defaults();
settings.custom_publish_host = Some("rtmp://live.example.com".to_string());
settings.advertise_address = Some("rtmp://live.example.com".to_string());
let json = ok(
serde_json::to_string(&settings),
"public settings should serialize",
);
assert!(json.contains("custom_publish_host"));
assert!(json.contains("advertise_address"));
assert!(json.contains("rtmp://live.example.com"));
}
}

@ -190,7 +190,10 @@ impl std::str::FromStr for PermissionSet {
#[cfg(test)]
mod tests {
use super::{validate_webrtc_settings, IceServerList, PermissionSet, WebRtcRuntimeSettings};
use super::{
parse_rtmp_advertise_address, validate_webrtc_settings, IceServerList, PermissionSet,
WebRtcRuntimeSettings,
};
use crate::models::RoomAdminPermissionBits;
use std::str::FromStr;
@ -237,6 +240,38 @@ mod tests {
.is_err());
}
}
#[test]
fn rtmp_advertise_address_accepts_origin_without_business_path() {
let plain = parse_rtmp_advertise_address("rtmp://live.example.com")
.expect("RTMP origin should be accepted");
assert_eq!(plain.scheme(), "rtmp");
assert_eq!(plain.host_str(), Some("live.example.com"));
assert_eq!(plain.port(), None);
let secure = parse_rtmp_advertise_address("rtmps://live.example.com:8443")
.expect("RTMPS origin with an explicit port should be accepted");
assert_eq!(secure.scheme(), "rtmps");
assert_eq!(secure.host_str(), Some("live.example.com"));
assert_eq!(secure.port(), Some(8443));
}
#[test]
fn rtmp_advertise_address_rejects_non_origin_components() {
for address in [
"",
"http://live.example.com",
"rtmp://user:secret@live.example.com",
"rtmp://live.example.com/app",
"rtmp://live.example.com?token=secret",
"rtmp://live.example.com#fragment",
] {
assert!(
parse_rtmp_advertise_address(address).is_err(),
"unexpectedly accepted {address}"
);
}
}
}
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize)]
@ -987,7 +1022,7 @@ impl OAuth2RuntimeSettingsUpdateMask {
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
pub struct RtmpRuntimeSettingsUpdateMask {
pub custom_publish_host: bool,
pub advertise_address: bool,
pub ts_disguised_as_png: bool,
}
@ -995,14 +1030,14 @@ impl RtmpRuntimeSettingsUpdateMask {
#[must_use]
pub const fn all() -> Self {
Self {
custom_publish_host: true,
advertise_address: true,
ts_disguised_as_png: true,
}
}
#[must_use]
pub const fn is_empty(&self) -> bool {
!self.custom_publish_host && !self.ts_disguised_as_png
!self.advertise_address && !self.ts_disguised_as_png
}
}
@ -1183,10 +1218,52 @@ pub struct OAuth2RuntimeSettings {
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct RtmpRuntimeSettings {
pub custom_publish_host: Option<String>,
pub advertise_address: Option<String>,
pub ts_disguised_as_png: bool,
}
pub fn parse_rtmp_advertise_address(address: &str) -> crate::Result<url::Url> {
let address = address.trim();
if address.is_empty() {
return Err(crate::Error::InvalidInput(
"rtmp.advertise_address must be non-empty when configured".to_string(),
));
}
let url = url::Url::parse(address).map_err(|error| {
crate::Error::InvalidInput(format!(
"rtmp.advertise_address must be a valid URL: {error}"
))
})?;
if !matches!(url.scheme(), "rtmp" | "rtmps") {
return Err(crate::Error::InvalidInput(
"rtmp.advertise_address must use the rtmp or rtmps scheme".to_string(),
));
}
if url.host_str().is_none() || url.cannot_be_a_base() {
return Err(crate::Error::InvalidInput(
"rtmp.advertise_address must include a host".to_string(),
));
}
if !url.username().is_empty() || url.password().is_some() {
return Err(crate::Error::InvalidInput(
"rtmp.advertise_address must not include credentials".to_string(),
));
}
if !matches!(url.path(), "" | "/") {
return Err(crate::Error::InvalidInput(
"rtmp.advertise_address must not include a path".to_string(),
));
}
if url.query().is_some() || url.fragment().is_some() {
return Err(crate::Error::InvalidInput(
"rtmp.advertise_address must not include a query or fragment".to_string(),
));
}
Ok(url)
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct EmailRuntimeSettings {
pub enabled: bool,
@ -1340,14 +1417,8 @@ fn validate_user_settings(
}
fn validate_rtmp_settings(settings: &RtmpRuntimeSettings) -> crate::Result<()> {
if settings
.custom_publish_host
.as_ref()
.is_some_and(|host| host.trim().is_empty())
{
return Err(crate::Error::InvalidInput(
"rtmp.custom_publish_host must be non-empty when configured".to_string(),
));
if let Some(address) = settings.advertise_address.as_deref() {
parse_rtmp_advertise_address(address)?;
}
let _ = settings.ts_disguised_as_png;
Ok(())
@ -1509,7 +1580,7 @@ pub struct PublicSettings {
pub enable_webauthn: bool,
pub ts_disguised_as_png: bool,
#[serde(skip_serializing_if = "Option::is_none")]
pub custom_publish_host: Option<String>,
pub advertise_address: Option<String>,
pub email_whitelist_enabled: bool,
#[serde(default, skip_serializing_if = "Vec::is_empty")]
pub email_whitelist_domains: Vec<String>,
@ -1536,7 +1607,7 @@ impl PublicSettings {
enable_email: false,
enable_webauthn: false,
ts_disguised_as_png: false,
custom_publish_host: None,
advertise_address: None,
email_whitelist_enabled: false,
email_whitelist_domains: Vec::new(),
}

@ -100,15 +100,16 @@ pub use file_upload_policies::{
MAX_PLAYLIST_COVER_SIZE_BYTES, MAX_ROOM_COVER_SIZE_BYTES, MAX_USER_AVATAR_SIZE_BYTES,
};
pub use global_settings::{
AdminDefaultPermissionsSetting, ChatRuntimeSettings, ConfiguredIceServer, CorsAllowedOrigins,
CorsAllowedOriginsSetting, CorsRuntimeSettings, DefaultMaxChatMessagesSetting,
DefaultMaxMembersSetting, EmailEnabledSetting, EmailFromEmailSetting, EmailFromNameSetting,
EmailRuntimeSettings, EmailSmtpCredentialsSetting, EmailSmtpHostSetting, EmailSmtpPortSetting,
EmailSmtpProxySetting, EmailUseTlsSetting, EmailWhitelistEnabledSetting, EmailWhitelistSetting,
EnableEmailSignupSetting, EnableGuestSetting, EnablePasswordSignupSetting,
EnableWebauthnSignupSetting, ExternalIceServersSetting, GuestDefaultPermissionsSetting,
IceServerList, MaxMessagesPerRoomSetting, MaxPinnedMessagesPerRoomSetting,
MaxRoomsPerUserSetting, MaxVoiceParticipantsPerRoomSetting, MemberDefaultPermissionsSetting,
parse_rtmp_advertise_address, AdminDefaultPermissionsSetting, ChatRuntimeSettings,
ConfiguredIceServer, CorsAllowedOrigins, CorsAllowedOriginsSetting, CorsRuntimeSettings,
DefaultMaxChatMessagesSetting, DefaultMaxMembersSetting, EmailEnabledSetting,
EmailFromEmailSetting, EmailFromNameSetting, EmailRuntimeSettings, EmailSmtpCredentialsSetting,
EmailSmtpHostSetting, EmailSmtpPortSetting, EmailSmtpProxySetting, EmailUseTlsSetting,
EmailWhitelistEnabledSetting, EmailWhitelistSetting, EnableEmailSignupSetting,
EnableGuestSetting, EnablePasswordSignupSetting, EnableWebauthnSignupSetting,
ExternalIceServersSetting, GuestDefaultPermissionsSetting, IceServerList,
MaxMessagesPerRoomSetting, MaxPinnedMessagesPerRoomSetting, MaxRoomsPerUserSetting,
MaxVoiceParticipantsPerRoomSetting, MemberDefaultPermissionsSetting,
MessageRetentionDaysSetting, OAuth2AppleProviderConfig, OAuth2CasdoorProviderConfig,
OAuth2DiscordProviderConfig, OAuth2FeishuProviderConfig, OAuth2GiteeProviderConfig,
OAuth2GithubProviderConfig, OAuth2GoogleProviderConfig, OAuth2LogtoProviderConfig,
@ -118,11 +119,11 @@ pub use global_settings::{
PermissionSet, PlaybackHistoryRuntimeSettings, PublicSettings,
RoomCreationApprovalRequiredSetting, RoomCreationEnabledSetting,
RoomCreationPasswordPolicySetting, RoomCreationRuntimeSettings, RoomDefaultsRuntimeSettings,
RoomPasswordPolicy, RtmpRuntimeSettings, RuntimeEmailConfigProvider, RuntimeSettings,
RuntimeSettingsStore, RuntimeSettingsUpdateMask, ServerIdentityIdSetting, ServerNameSetting,
ServerRuntimeSettings, TsDisguisedAsPngSetting, UserRuntimeSettings, WebRtcRuntimeSettings,
DEFAULT_MAX_VOICE_PARTICIPANTS_PER_ROOM, MAX_RUNTIME_SETTINGS_IMPORT_REQUEST_BYTES,
MAX_RUNTIME_SETTINGS_SNAPSHOT_BYTES,
RoomPasswordPolicy, RtmpAdvertiseAddressSetting, RtmpRuntimeSettings,
RuntimeEmailConfigProvider, RuntimeSettings, RuntimeSettingsStore, RuntimeSettingsUpdateMask,
ServerIdentityIdSetting, ServerNameSetting, ServerRuntimeSettings, TsDisguisedAsPngSetting,
UserRuntimeSettings, WebRtcRuntimeSettings, DEFAULT_MAX_VOICE_PARTICIPANTS_PER_ROOM,
MAX_RUNTIME_SETTINGS_IMPORT_REQUEST_BYTES, MAX_RUNTIME_SETTINGS_SNAPSHOT_BYTES,
};
pub use media::{
AddMediaRequest, BackendPlaybackRequest, CreateMediaCoverUploadSession,

@ -86,14 +86,14 @@ fn test_cors_allowed_origins_updates() {
}
#[test]
fn test_public_settings_skips_empty_custom_publish_host() {
fn test_public_settings_skips_empty_rtmp_advertise_address() {
let defaults = PublicSettings::defaults();
let json = ok(
serde_json::to_string(&defaults),
"public settings should serialize",
);
assert!(!json.contains("custom_publish_host"));
assert!(!json.contains("advertise_address"));
assert_eq!(defaults.max_pinned_chat_messages_per_room, 20);
assert!(json.contains("max_pinned_chat_messages_per_room"));
}

@ -28,7 +28,8 @@ use tracing::{error, info, warn};
pub use super::tracker::{StreamSubscriberGuard, StreamTracker};
const KICK_PUBLISHER_EVENT_SEND_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(5);
const HLS_GENERATION_READY_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(2);
const HLS_ACTIVE_GENERATION_READY_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(2);
const HLS_GENERATION_READY_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(15);
const HLS_GENERATION_READY_POLL_INTERVAL: std::time::Duration =
std::time::Duration::from_millis(25);
@ -47,7 +48,12 @@ pub(crate) async fn find_hls_generation_state(
.get(&stream_key)
.map(|entry| Arc::clone(entry.value()))
{
return Some(state);
if !state.read().playlist.segments.is_empty() {
return Some(state);
}
if !wait_for_ready {
return None;
}
}
if !wait_for_ready || tokio::time::Instant::now() >= deadline {
return None;
@ -720,7 +726,7 @@ impl HlsStreamingApi {
room_id: &str,
media_id: &str,
) -> Result<Option<StreamGeneration>> {
let deadline = tokio::time::Instant::now() + std::time::Duration::from_secs(2);
let deadline = tokio::time::Instant::now() + HLS_ACTIVE_GENERATION_READY_TIMEOUT;
loop {
let generation = infrastructure
.registry
@ -782,7 +788,7 @@ impl HlsStreamingApi {
"Failed to resolve active HLS generation: {error}"
))
})?;
if generation.is_some() || external_source.is_none() {
if generation.is_some() {
return Ok(generation);
}
@ -1069,7 +1075,74 @@ mod tests {
}
#[tokio::test]
async fn active_local_playlist_waits_for_remuxer_generation_state() -> TestResult {
async fn rtmp_hls_master_waits_for_delayed_active_generation() -> TestResult {
let registry = Arc::new(TestStreamRegistry::new());
let (event_sender, _event_receiver) = mpsc::channel(8);
let infrastructure = LiveStreamingInfrastructure::new(
registry.clone(),
event_sender,
Arc::new(StreamTracker::new()),
"node-local".to_string(),
synctv_common::ssrf::SsrfGuard::disabled(),
)?;
let generation_id = synctv_xiu::streamhub::utils::Uuid::new().to_string();
let delayed_generation_id = generation_id.clone();
let register_task = tokio::spawn(async move {
tokio::time::sleep(std::time::Duration::from_millis(50)).await;
registry
.try_activate_generation(
"room-master",
"media-master",
"node-local",
"",
"127.0.0.1:50051",
&delayed_generation_id,
)
.await
});
let generation = HlsStreamingApi::resolve_active_generation_with_pull(
&infrastructure,
"room-master",
"media-master",
None,
)
.await?
.ok_or_else(|| test_error("master should wait for publisher registration"))?;
assert!(register_task.await??);
assert_eq!(generation.generation_id, generation_id);
Ok(())
}
#[tokio::test(start_paused = true)]
async fn rtmp_hls_master_times_out_when_stream_stays_offline() -> TestResult {
let registry = Arc::new(TestStreamRegistry::new());
let (event_sender, _event_receiver) = mpsc::channel(8);
let infrastructure = LiveStreamingInfrastructure::new(
registry,
event_sender,
Arc::new(StreamTracker::new()),
"node-local".to_string(),
synctv_common::ssrf::SsrfGuard::disabled(),
)?;
let started_at = tokio::time::Instant::now();
let generation = HlsStreamingApi::resolve_active_generation_with_pull(
&infrastructure,
"room-offline",
"media-offline",
None,
)
.await?;
assert!(generation.is_none());
assert!(started_at.elapsed() >= HLS_ACTIVE_GENERATION_READY_TIMEOUT);
Ok(())
}
#[tokio::test]
async fn active_local_playlist_waits_for_first_complete_segment() -> TestResult {
let registry = Arc::new(TestStreamRegistry::new());
let generation_id = synctv_xiu::streamhub::utils::Uuid::new();
let generation_id_string = generation_id.to_string();
@ -1096,35 +1169,38 @@ mod tests {
)?
.with_hls_stream_registry(hls_registry.clone());
let delayed_registry = hls_registry;
let delayed_generation_id = generation_id_string.clone();
let stream_state = Arc::new(parking_lot::RwLock::new(
synctv_xiu::hls::StreamProcessorState {
app_name: "room-ready".to_string(),
stream_name: "media-ready".to_string(),
playlist: synctv_xiu::hls::HlsPlaylist::new(),
generation_id,
marked_for_cleanup: false,
cleanup_segment_names: Vec::new(),
},
));
hls_registry.insert(
synctv_xiu::hls::generation_registry_key(
"room-ready",
"media-ready",
&generation_id_string,
),
stream_state.clone(),
);
let delayed_state = stream_state;
let insert_task = tokio::spawn(async move {
tokio::time::sleep(std::time::Duration::from_millis(50)).await;
let mut playlist = synctv_xiu::hls::HlsPlaylist::new();
playlist.push_segment(synctv_xiu::hls::SegmentInfo {
sequence: 0,
duration_ms: 1_000,
started_at_ms: 0,
ts_name: "ready-segment".to_string(),
discontinuity: false,
});
delayed_registry.insert(
synctv_xiu::hls::generation_registry_key(
"room-ready",
"media-ready",
&delayed_generation_id,
),
Arc::new(parking_lot::RwLock::new(
synctv_xiu::hls::StreamProcessorState {
app_name: "room-ready".to_string(),
stream_name: "media-ready".to_string(),
playlist,
generation_id,
marked_for_cleanup: false,
cleanup_segment_names: Vec::new(),
},
)),
);
delayed_state
.write()
.playlist
.push_segment(synctv_xiu::hls::SegmentInfo {
sequence: 0,
duration_ms: 1_000,
started_at_ms: 0,
ts_name: "ready-segment".to_string(),
discontinuity: false,
});
});
let playlist = HlsStreamingApi::generate_playlist_simple(
@ -1142,6 +1218,66 @@ mod tests {
Ok(())
}
#[tokio::test(start_paused = true)]
async fn active_local_playlist_times_out_before_returning_an_empty_manifest() -> TestResult {
let registry = Arc::new(TestStreamRegistry::new());
let generation_id = synctv_xiu::streamhub::utils::Uuid::new();
let generation_id_string = generation_id.to_string();
assert!(
registry
.try_activate_generation(
"room-empty",
"media-empty",
"node-local",
"",
"127.0.0.1:50051",
&generation_id_string,
)
.await?
);
let (event_sender, _event_receiver) = mpsc::channel(8);
let hls_registry = Arc::new(dashmap::DashMap::new());
hls_registry.insert(
synctv_xiu::hls::generation_registry_key(
"room-empty",
"media-empty",
&generation_id_string,
),
Arc::new(parking_lot::RwLock::new(
synctv_xiu::hls::StreamProcessorState {
app_name: "room-empty".to_string(),
stream_name: "media-empty".to_string(),
playlist: synctv_xiu::hls::HlsPlaylist::new(),
generation_id,
marked_for_cleanup: false,
cleanup_segment_names: Vec::new(),
},
)),
);
let infrastructure = LiveStreamingInfrastructure::new(
registry,
event_sender,
Arc::new(StreamTracker::new()),
"node-local".to_string(),
synctv_common::ssrf::SsrfGuard::disabled(),
)?
.with_hls_stream_registry(hls_registry);
let started_at = tokio::time::Instant::now();
let playlist = HlsStreamingApi::generate_playlist_simple(
&infrastructure,
"room-empty",
"media-empty",
&generation_id_string,
"/segments/",
)
.await?;
assert!(playlist.is_none());
assert!(started_at.elapsed() >= HLS_GENERATION_READY_TIMEOUT);
Ok(())
}
#[tokio::test]
async fn ended_local_playlist_remains_available_after_publisher_unregister() -> TestResult {
let registry = Arc::new(TestStreamRegistry::new());

@ -346,7 +346,7 @@ message OAuth2GiteeProviderConfig {
}
message RtmpSettings {
optional string custom_publish_host = 1;
optional string advertise_address = 1;
bool ts_disguised_as_png = 2;
}
@ -454,7 +454,7 @@ message OAuth2SettingsPatch {
}
message RtmpSettingsPatch {
optional string custom_publish_host = 1;
optional string advertise_address = 1;
optional bool ts_disguised_as_png = 2;
}

@ -5203,7 +5203,7 @@ message GetPublicSettingsResponse {
// RTMP settings
bool ts_disguised_as_png = 18;
optional string custom_publish_host = 19;
optional string advertise_address = 19;
// Email settings
bool email_whitelist_enabled = 20;

Loading…
Cancel
Save