1use std::time::Duration;23use sigil_protocol::{4 ClientControlEnvelopeV2, ControlKeyframeReasonV2, KeyframeRequestReasonV3,5 MediaControlRequestV3, write_media_control_request_v3,6};7use std::sync::atomic::Ordering;89use crate::commands::state::{AppState, MediaControlRequestSender};1011const MEDIA_CONTROL_WRITE_TIMEOUT: Duration = Duration::from_secs(1);1213pub(crate) async fn run_media_control_writer_v3(14 mut stream: iroh::endpoint::SendStream,15 mut requests: tokio::sync::mpsc::Receiver<(KeyframeRequestReasonV3, Option<u64>)>,16) {17 let mut request_id = 0_u64;18 while let Some((reason, last_sequence)) = requests.recv().await {19 let Some(next_request_id) = request_id.checked_add(1) else {20 eprintln!("[client] media v3 keyframe request id overflowed");21 break;22 };23 request_id = next_request_id;24 let request = MediaControlRequestV3::request_keyframe(request_id, last_sequence, reason);25 match tokio::time::timeout(26 MEDIA_CONTROL_WRITE_TIMEOUT,27 write_media_control_request_v3(&mut stream, &request),28 )29 .await30 {31 Ok(Ok(())) => {}32 Ok(Err(error)) => {33 eprintln!("[client] media v3 keyframe request failed: {error}");34 break;35 }36 Err(_) => {37 eprintln!("[client] media v3 keyframe request timed out");38 break;39 }40 }41 }42 let _ = stream.finish();43}4445fn parse_keyframe_request_reason(reason: &str) -> Result<KeyframeRequestReasonV3, String> {46 match reason {47 "join" => Ok(KeyframeRequestReasonV3::Join),48 "transport-gap" => Ok(KeyframeRequestReasonV3::TransportGap),49 "delivery-timeout" => Ok(KeyframeRequestReasonV3::DeliveryTimeout),50 "discontinuity" | "decoder-reset" | "decoder-error" => {51 Ok(KeyframeRequestReasonV3::DecoderReset)52 }53 "frontend-backpressure" => Ok(KeyframeRequestReasonV3::FrontendBackpressure),54 _ => Err(format!("Unsupported keyframe request reason: {reason}")),55 }56}5758pub(crate) fn try_queue_media_keyframe_request(59 sender: Option<&MediaControlRequestSender>,60 reason: KeyframeRequestReasonV3,61 last_sequence: Option<u64>,62) {63 if let Some(sender) = sender {64 let _ = sender.try_send((reason, last_sequence));65 }66}6768pub(crate) async fn request_keyframe(69 state: &AppState,70 generation: u64,71 reason: String,72) -> Result<bool, String> {73 let reason = parse_keyframe_request_reason(&reason)?;74 let control = state.media_control.lock().await;75 if let Some((current_generation, sender)) = control.as_ref() {76 if *current_generation != generation {77 return Ok(false);78 }79 return match sender.try_send((reason, None)) {80 Ok(()) => Ok(true),81 Err(tokio::sync::mpsc::error::TrySendError::Full(_)) => Ok(false),82 Err(tokio::sync::mpsc::error::TrySendError::Closed(_)) => {83 Err("Media control channel closed".to_string())84 }85 };86 }87 drop(control);88 let control = state.session_control.lock().await;89 let Some((current_generation, sender)) = control.as_ref() else {90 return Ok(false);91 };92 if *current_generation != generation {93 return Ok(false);94 }95 let request_id = state96 .control_request_id97 .fetch_add(1, Ordering::Relaxed)98 .checked_add(1)99 .ok_or_else(|| "Control v2 request id overflowed".to_string())?;100 match sender.try_send(ClientControlEnvelopeV2::Keyframe {101 request_id,102 last_sequence: None,103 reason: ControlKeyframeReasonV2::from(reason),104 }) {105 Ok(()) => Ok(true),106 Err(tokio::sync::mpsc::error::TrySendError::Full(_)) => Ok(false),107 Err(tokio::sync::mpsc::error::TrySendError::Closed(_)) => {108 Err("Control v2 channel closed".to_string())109 }110 }111}112113114mod tests {115 use super::*;116117 118 fn keyframe_request_reason_mapping_is_strict_and_coalesces_decoder_reasons() {119 assert_eq!(120 parse_keyframe_request_reason("transport-gap").unwrap(),121 KeyframeRequestReasonV3::TransportGap122 );123 assert_eq!(124 parse_keyframe_request_reason("discontinuity").unwrap(),125 KeyframeRequestReasonV3::DecoderReset126 );127 assert_eq!(128 parse_keyframe_request_reason("decoder-error").unwrap(),129 KeyframeRequestReasonV3::DecoderReset130 );131 assert!(parse_keyframe_request_reason("please").is_err());132 }133}