1use std::time::Duration;23use sigil_protocol::{4 KeyframeRequestReasonV3, MediaControlRequestV3, write_media_control_request_v3,5};67use crate::commands::state::{AppState, MediaControlRequestSender};89const MEDIA_CONTROL_WRITE_TIMEOUT: Duration = Duration::from_secs(1);1011pub(crate) async fn run_media_control_writer_v3(12 mut stream: iroh::endpoint::SendStream,13 mut requests: tokio::sync::mpsc::Receiver<(KeyframeRequestReasonV3, Option<u64>)>,14) {15 let mut request_id = 0_u64;16 while let Some((reason, last_sequence)) = requests.recv().await {17 let Some(next_request_id) = request_id.checked_add(1) else {18 eprintln!("[client] media v3 keyframe request id overflowed");19 break;20 };21 request_id = next_request_id;22 let request = MediaControlRequestV3::request_keyframe(request_id, last_sequence, reason);23 match tokio::time::timeout(24 MEDIA_CONTROL_WRITE_TIMEOUT,25 write_media_control_request_v3(&mut stream, &request),26 )27 .await28 {29 Ok(Ok(())) => {}30 Ok(Err(error)) => {31 eprintln!("[client] media v3 keyframe request failed: {error}");32 break;33 }34 Err(_) => {35 eprintln!("[client] media v3 keyframe request timed out");36 break;37 }38 }39 }40 let _ = stream.finish();41}4243fn parse_keyframe_request_reason(reason: &str) -> Result<KeyframeRequestReasonV3, String> {44 match reason {45 "join" => Ok(KeyframeRequestReasonV3::Join),46 "transport-gap" => Ok(KeyframeRequestReasonV3::TransportGap),47 "delivery-timeout" => Ok(KeyframeRequestReasonV3::DeliveryTimeout),48 "discontinuity" | "decoder-reset" | "decoder-error" => {49 Ok(KeyframeRequestReasonV3::DecoderReset)50 }51 "frontend-backpressure" => Ok(KeyframeRequestReasonV3::FrontendBackpressure),52 _ => Err(format!("Unsupported keyframe request reason: {reason}")),53 }54}5556pub(crate) fn try_queue_media_keyframe_request(57 sender: Option<&MediaControlRequestSender>,58 reason: KeyframeRequestReasonV3,59 last_sequence: Option<u64>,60) {61 if let Some(sender) = sender {62 let _ = sender.try_send((reason, last_sequence));63 }64}6566pub(crate) async fn request_keyframe(67 state: &AppState,68 generation: u64,69 reason: String,70) -> Result<bool, String> {71 let reason = parse_keyframe_request_reason(&reason)?;72 let control = state.media_control.lock().await;73 let Some((current_generation, sender)) = control.as_ref() else {74 return Ok(false);75 };76 if *current_generation != generation {77 return Ok(false);78 }79 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}878889mod tests {90 use super::*;9192 93 fn keyframe_request_reason_mapping_is_strict_and_coalesces_decoder_reasons() {94 assert_eq!(95 parse_keyframe_request_reason("transport-gap").unwrap(),96 KeyframeRequestReasonV3::TransportGap97 );98 assert_eq!(99 parse_keyframe_request_reason("discontinuity").unwrap(),100 KeyframeRequestReasonV3::DecoderReset101 );102 assert_eq!(103 parse_keyframe_request_reason("decoder-error").unwrap(),104 KeyframeRequestReasonV3::DecoderReset105 );106 assert!(parse_keyframe_request_reason("please").is_err());107 }108}