smooth_runtime/client/
engine.rs

1//! [`SmoothClient`]: the sans-IO Smooth Streaming session.
2
3use std::collections::VecDeque;
4use std::time::Duration;
5
6use broadcast_smooth::{SmoothManifest, StreamIndex, StreamType};
7use transmux::dash::rfc6381_track_codec_string;
8use transmux::pipeline::build_init_segment;
9use transmux::smooth_fragment::{
10    FragmentState, fragment_state, fragment_track_id, manifest_declares_protection,
11};
12use transmux::{TrackId, TrackSpec, track_spec_from_quality_level};
13use url::Url;
14
15use broadcast_common::backoff::{RetrySchedule, mix_seed};
16use broadcast_common::url_ref;
17
18use super::action::{Action, FetchFailure, RequestId, StreamId};
19use super::config::{SYNTHETIC_MOVIE_TIMESCALE, SmoothConfig};
20use super::output::{Dropped, ManifestFault, Output, QualityInfo};
21use super::stream::{Chunk, Cursor, HeldFragment, Stream, demux_fragment, live_join};
22use crate::error::SmoothError;
23use broadcast_common::units::MICROS_PER_SEC;
24
25/// The shortest retry delay: a zero step must never become a burst of retries at one instant.
26const MIN_BACKOFF_MS: u64 = 1;
27const REFRESH_DIVISOR: u64 = 2;
28const FIRST_TRACK_ID: u32 = 1;
29const FOURCC_AVC: [&str; 2] = ["H264", "AVC1"];
30const FOURCC_HEVC: [&str; 3] = ["H265", "HEVC", "HVC1"];
31const FOURCC_HEVC_ALT: &str = "HEV1";
32
33fn codec_family(four_cc: &str) -> String {
34    let upper = four_cc.to_ascii_uppercase();
35    if FOURCC_AVC.contains(&upper.as_str()) {
36        FOURCC_AVC[0].to_string()
37    } else if FOURCC_HEVC.contains(&upper.as_str()) || upper == FOURCC_HEVC_ALT {
38        FOURCC_HEVC[0].to_string()
39    } else {
40        upper
41    }
42}
43
44fn timescale_of(manifest: &SmoothManifest, index: &StreamIndex) -> Option<u32> {
45    u32::try_from(index.timescale.unwrap_or(manifest.timescale))
46        .ok()
47        .filter(|&t| t != 0)
48}
49
50fn primary_video(manifest: &SmoothManifest) -> Option<&StreamIndex> {
51    manifest.streams.iter().find(|s| {
52        s.stream_type == StreamType::Video
53            && !s.qualities.is_empty()
54            && timescale_of(manifest, s).is_some()
55    })
56}
57
58#[derive(Debug, Clone, Copy)]
59enum ManifestCursor {
60    Idle,
61    InFlight(RequestId),
62}
63
64/// The Smooth Streaming client engine. See the [crate docs](crate).
65pub struct SmoothClient {
66    config: SmoothConfig,
67    manifest_url: Url,
68    streams: Vec<Stream>,
69    started: bool,
70    announced: bool,
71    is_live: bool,
72    ended: bool,
73    fatal: bool,
74    video_quality_count: usize,
75    manifest: ManifestCursor,
76    manifest_at: Option<u64>,
77    manifest_failures: u32,
78    retry: RetrySchedule,
79    next_request: u64,
80    actions: VecDeque<Action>,
81    outputs: VecDeque<Output>,
82}
83
84impl SmoothClient {
85    /// A client for the manifest at `manifest_url`; the first [`Action`] is its fetch.
86    pub fn new(manifest_url: Url, config: SmoothConfig) -> Self {
87        let retry = RetrySchedule::from_steps(
88            &config
89                .backoff_ms
90                .iter()
91                .map(|ms| Duration::from_millis((*ms).max(MIN_BACKOFF_MS)))
92                .collect::<Vec<_>>(),
93            Duration::from_millis(config.backoff_cap_ms.max(MIN_BACKOFF_MS)),
94            mix_seed(config.jitter_seed),
95        );
96        Self {
97            config,
98            manifest_url,
99            streams: Vec::new(),
100            started: false,
101            announced: false,
102            is_live: false,
103            ended: false,
104            fatal: false,
105            video_quality_count: 0,
106            manifest: ManifestCursor::Idle,
107            manifest_at: Some(0),
108            manifest_failures: 0,
109            retry,
110            next_request: 0,
111            actions: VecDeque::new(),
112            outputs: VecDeque::new(),
113        }
114    }
115
116    /// Re-base relative fragment URLs on `url`, after a redirect of the manifest.
117    pub fn set_manifest_url(&mut self, url: Url) {
118        self.manifest_url = url;
119    }
120
121    /// True once a manifest has been accepted and the streams are resolved.
122    pub fn is_started(&self) -> bool {
123        self.started
124    }
125
126    /// True while the presentation is live and the session has not ended.
127    pub fn is_live(&self) -> bool {
128        self.is_live && !self.ended
129    }
130
131    /// True once a static presentation has played to its end.
132    pub fn is_ended(&self) -> bool {
133        self.ended
134    }
135
136    /// True once the session has ended fatally.
137    pub fn is_fatal(&self) -> bool {
138        self.fatal
139    }
140
141    /// How many `QualityLevel`s the first video stream offers (0 before the first manifest).
142    pub fn video_quality_count(&self) -> usize {
143        self.video_quality_count
144    }
145
146    /// Fetch the first video stream's later fragments at `QualityLevel` `index`
147    /// (an index into [`Output::Qualities`]); false when there is no such level.
148    ///
149    /// Fragments already requested keep their level. Once the tracks are
150    /// announced, a level of the same codec family is accepted even when its
151    /// codec configuration (parameter sets, resolution) differs: the new
152    /// configuration is announced as [`Output::TracksChanged`] before the first
153    /// sample of the new level. A level of another codec family is refused (false).
154    pub fn select_video_quality(&mut self, index: usize) -> bool {
155        let Some(si) = self
156            .streams
157            .iter()
158            .position(|s| s.index.stream_type == StreamType::Video)
159        else {
160            return false;
161        };
162        let Some(quality) = self.streams[si].index.qualities.get(index).cloned() else {
163            return false;
164        };
165        if self.announced && !self.switchable(si, &quality) {
166            return false;
167        }
168        self.streams[si].quality = quality;
169        true
170    }
171
172    fn switchable(&self, si: usize, candidate: &broadcast_smooth::QualityLevel) -> bool {
173        let stream = &self.streams[si];
174        codec_family(&stream.applied.four_cc) == codec_family(&candidate.four_cc)
175            && track_spec_from_quality_level(
176                TrackId::new(FIRST_TRACK_ID),
177                stream.timescale,
178                stream.index.stream_type,
179                candidate,
180            )
181            .is_ok()
182    }
183
184    /// The earliest instant, in microseconds, at which [`Self::poll`] has more to do
185    /// without any new input.
186    pub fn next_deadline(&self) -> Option<u64> {
187        if self.fatal || self.ended {
188            return None;
189        }
190        let backoffs = self.streams.iter().filter_map(|s| match s.cursor {
191            Cursor::Backoff { until, .. } => Some(until),
192            _ => None,
193        });
194        backoffs.chain(self.manifest_at).min()
195    }
196
197    /// The next fetch the host should perform, if any.
198    pub fn poll(&mut self, now: u64) -> Option<Action> {
199        if !self.fatal && !self.ended {
200            self.fire_timers(now);
201            self.schedule();
202            self.advance(now);
203        }
204        self.actions.pop_front()
205    }
206
207    /// The next thing to act on, if any.
208    pub fn next_output(&mut self) -> Option<Output> {
209        self.outputs.pop_front()
210    }
211
212    /// A manifest body arrived for request `id`.
213    pub fn on_manifest(&mut self, now: u64, id: RequestId, bytes: &[u8]) {
214        if !self.take_manifest(id) {
215            return;
216        }
217        let Ok(text) = std::str::from_utf8(bytes) else {
218            return self.manifest_failed(now, ManifestFault::Utf8, 0);
219        };
220        if manifest_declares_protection(text) {
221            return self.fail(SmoothError::ManifestProtected);
222        }
223        match SmoothManifest::parse(text) {
224            Ok(manifest) => self.accept_manifest(&manifest),
225            Err(_) => self.manifest_failed(now, ManifestFault::Parse, 0),
226        }
227        self.advance(now);
228    }
229
230    /// A fragment body arrived for request `id`.
231    pub fn on_fragment(&mut self, now: u64, id: RequestId, bytes: &[u8]) {
232        let Some((si, chunk)) = self.take_fragment(id) else {
233            return;
234        };
235        match fragment_state(bytes) {
236            FragmentState::Clear => {}
237            FragmentState::Encrypted => return self.fail(SmoothError::FragmentEncrypted),
238            _ => return self.fragment_failed(now, si, chunk, None, 0),
239        }
240        if self.streams[si].local_track_id.is_none() {
241            match fragment_track_id(bytes) {
242                Ok(tid) => self.streams[si].local_track_id = Some(tid),
243                Err(_) => return self.fragment_failed(now, si, chunk, None, 0),
244            }
245        }
246        self.streams[si].failures = 0;
247        self.streams[si].cursor = Cursor::Idle;
248        let quality = self.streams[si].fetched.clone();
249        if self.announced {
250            self.emit_fragment(si, chunk.0, bytes, &quality);
251        } else {
252            self.streams[si].held.push(HeldFragment {
253                start: chunk.0,
254                body: bytes.to_vec(),
255                quality,
256            });
257            self.try_announce();
258        }
259        self.advance(now);
260    }
261
262    /// The fetch for request `id` produced no usable body.
263    pub fn on_failure(&mut self, now: u64, id: RequestId, failure: FetchFailure) {
264        if self.take_manifest(id) {
265            if failure.is_not_found() {
266                self.outputs.push_back(Output::ManifestNotFound);
267            }
268            return self.manifest_failed(now, ManifestFault::Fetch, failure.retry_after_us);
269        }
270        if let Some((si, chunk)) = self.take_fragment(id) {
271            self.fragment_failed(now, si, chunk, failure.status, failure.retry_after_us);
272            self.advance(now);
273        }
274    }
275
276    fn mint(&mut self) -> RequestId {
277        self.next_request += 1;
278        RequestId(self.next_request)
279    }
280
281    fn fail(&mut self, error: SmoothError) {
282        self.fatal = true;
283        self.outputs.push_back(Output::Fatal(error));
284    }
285
286    fn take_manifest(&mut self, id: RequestId) -> bool {
287        if matches!(self.manifest, ManifestCursor::InFlight(m) if m == id) {
288            self.manifest = ManifestCursor::Idle;
289            return !self.fatal;
290        }
291        false
292    }
293
294    fn take_fragment(&mut self, id: RequestId) -> Option<(usize, Chunk)> {
295        if self.fatal {
296            return None;
297        }
298        self.streams
299            .iter()
300            .position(|s| matches!(s.cursor, Cursor::InFlight { id: i, .. } if i == id))
301            .and_then(|si| match self.streams[si].cursor {
302                Cursor::InFlight { chunk, .. } => Some((si, chunk)),
303                _ => None,
304            })
305    }
306
307    /// The wait before retry number `attempt` (1-based: the first failure is attempt 1). The
308    /// shared [`RetrySchedule`] counts retries from 0, so the first failure takes its first step
309    /// exactly and every later one a jittered step.
310    fn backoff_us(&self, attempt: u32) -> u64 {
311        let delay = self.retry.delay_for_attempt(attempt.saturating_sub(1));
312        u64::try_from(delay.as_micros()).unwrap_or(u64::MAX)
313    }
314
315    fn fire_timers(&mut self, now: u64) {
316        let due: Vec<usize> = self
317            .streams
318            .iter()
319            .enumerate()
320            .filter(|(_, s)| matches!(s.cursor, Cursor::Backoff { until, .. } if until <= now))
321            .map(|(i, _)| i)
322            .collect();
323        for si in due {
324            if let Cursor::Backoff { chunk, .. } = self.streams[si].cursor {
325                self.issue_fragment(si, chunk);
326            }
327        }
328        if self.manifest_at.is_some_and(|t| t <= now)
329            && matches!(self.manifest, ManifestCursor::Idle)
330        {
331            self.manifest_at = None;
332            let id = self.mint();
333            self.manifest = ManifestCursor::InFlight(id);
334            self.actions.push_back(Action::FetchManifest {
335                id,
336                url: self.manifest_url.clone(),
337            });
338        }
339    }
340
341    fn schedule(&mut self) {
342        if !self.started {
343            return;
344        }
345        for si in 0..self.streams.len() {
346            let stream = &mut self.streams[si];
347            if stream.abandoned || !matches!(stream.cursor, Cursor::Idle) {
348                continue;
349            }
350            if let Some(chunk) = stream.plan.pop_front() {
351                self.issue_fragment(si, chunk);
352            }
353        }
354    }
355
356    fn issue_fragment(&mut self, si: usize, chunk: Chunk) {
357        let stream = &self.streams[si];
358        let relative = stream
359            .index
360            .resolve_fragment_url(stream.quality.bitrate, chunk.0);
361        let Ok(url) = url_ref::resolve(&self.manifest_url, &relative) else {
362            self.streams[si].cursor = Cursor::Idle;
363            self.outputs
364                .push_back(Output::Dropped(Dropped::FragmentUrl {
365                    stream: StreamId(si),
366                }));
367            return;
368        };
369        let id = self.mint();
370        self.streams[si].fetched = self.streams[si].quality.clone();
371        self.streams[si].cursor = Cursor::InFlight { id, chunk };
372        self.actions.push_back(Action::FetchFragment {
373            id,
374            stream: StreamId(si),
375            start: chunk.0,
376            duration: chunk.1,
377            url,
378        });
379    }
380
381    fn manifest_failed(&mut self, now: u64, fault: ManifestFault, wait_us: u64) {
382        self.manifest_failures += 1;
383        if self.manifest_failures >= self.config.max_manifest_failures {
384            return self.fail(SmoothError::Manifest(fault));
385        }
386        self.outputs
387            .push_back(Output::Dropped(Dropped::Manifest(fault)));
388        self.manifest_at =
389            Some(now.saturating_add(self.backoff_us(self.manifest_failures).max(wait_us)));
390    }
391
392    fn fragment_failed(
393        &mut self,
394        now: u64,
395        si: usize,
396        chunk: Chunk,
397        status: Option<u16>,
398        wait_us: u64,
399    ) {
400        self.streams[si].failures += 1;
401        if self.streams[si].failures >= self.config.max_fragment_failures {
402            if !self.is_live {
403                return self.fail(SmoothError::Fragment { status });
404            }
405            self.streams[si].failures = 0;
406            self.streams[si].cursor = Cursor::Idle;
407            self.outputs.push_back(Output::Dropped(Dropped::Fragment {
408                stream: StreamId(si),
409                status,
410            }));
411            self.outputs.push_back(Output::Discontinuity);
412            return;
413        }
414        let missing = FetchFailure {
415            status,
416            retry_after_us: 0,
417        }
418        .is_not_found();
419        if self.is_live && missing && matches!(self.manifest, ManifestCursor::Idle) {
420            self.manifest_at = Some(now);
421        }
422        let until = now.saturating_add(self.backoff_us(self.streams[si].failures).max(wait_us));
423        self.streams[si].cursor = Cursor::Backoff { until, chunk };
424    }
425
426    fn accept_manifest(&mut self, manifest: &SmoothManifest) {
427        self.manifest_failures = 0;
428        if self.started {
429            return self.refresh_streams(manifest);
430        }
431        match self.build_streams(manifest) {
432            Ok(true) => {
433                self.started = true;
434                self.is_live = manifest.is_live;
435                self.offer_qualities(manifest);
436            }
437            Ok(false) => self.fail(SmoothError::NoUsableStream),
438            Err(e) => self.fail(e),
439        }
440    }
441
442    fn build_streams(&mut self, manifest: &SmoothManifest) -> Result<bool, SmoothError> {
443        for index in &manifest.streams {
444            if !matches!(index.stream_type, StreamType::Video | StreamType::Audio) {
445                continue;
446            }
447            let pick = if primary_video(manifest).is_some_and(|v| std::ptr::eq(v, index)) {
448                self.choose_video_quality(manifest, index)
449            } else {
450                0
451            };
452            let Some(quality) = index.qualities.get(pick).cloned() else {
453                continue;
454            };
455            let Some(timescale) = timescale_of(manifest, index) else {
456                continue;
457            };
458            let chunks = match index.enumerate_chunks() {
459                Ok(chunks) => chunks,
460                Err(e) => {
461                    return Err(SmoothError::Timeline {
462                        stream: self.streams.len(),
463                        reason: e.to_string(),
464                    });
465                }
466            };
467            let chunks = if manifest.is_live {
468                live_join(chunks, manifest, self.config.live_join_fragments)
469            } else {
470                chunks
471            };
472            let mut stream = Stream::new(index.clone(), timescale, quality);
473            stream.plan_from(chunks);
474            self.streams.push(stream);
475        }
476        Ok(!self.streams.is_empty())
477    }
478
479    fn quality_infos(manifest: &SmoothManifest, index: &StreamIndex) -> Vec<QualityInfo> {
480        let timescale = timescale_of(manifest, index);
481        index
482            .qualities
483            .iter()
484            .map(|q| {
485                let codecs = timescale
486                    .and_then(|ts| {
487                        track_spec_from_quality_level(
488                            TrackId::new(FIRST_TRACK_ID),
489                            ts,
490                            index.stream_type,
491                            q,
492                        )
493                        .ok()
494                    })
495                    .and_then(|spec| rfc6381_track_codec_string(&spec).ok());
496                QualityInfo {
497                    bitrate: q.bitrate,
498                    resolution: q.width.zip(q.height),
499                    codecs,
500                }
501            })
502            .collect()
503    }
504
505    fn choose_video_quality(&mut self, manifest: &SmoothManifest, index: &StreamIndex) -> usize {
506        let infos = Self::quality_infos(manifest, index);
507        self.video_quality_count = infos.len();
508        self.config.video_quality.choose(&infos)
509    }
510
511    fn offer_qualities(&mut self, manifest: &SmoothManifest) {
512        let Some(video) = primary_video(manifest) else {
513            return;
514        };
515        let qualities = Self::quality_infos(manifest, video);
516        let chosen = self.config.video_quality.choose(&qualities);
517        self.outputs
518            .push_back(Output::Qualities { qualities, chosen });
519    }
520
521    fn refresh_streams(&mut self, manifest: &SmoothManifest) {
522        self.is_live = manifest.is_live;
523        let mut discontinuity = false;
524        for si in 0..self.streams.len() {
525            if self.streams[si].abandoned {
526                continue;
527            }
528            let found = manifest
529                .streams
530                .iter()
531                .find(|s| s.same_stream(&self.streams[si].index));
532            let Some(found) = found else {
533                self.streams[si].refresh_misses += 1;
534                if self.streams[si].refresh_misses >= self.config.max_refresh_misses {
535                    self.streams[si].abandoned = true;
536                    self.streams[si].plan.clear();
537                    self.outputs.push_back(Output::Dropped(Dropped::StreamLost {
538                        stream: StreamId(si),
539                    }));
540                }
541                continue;
542            };
543            self.streams[si].refresh_misses = 0;
544            match found.enumerate_chunks() {
545                Ok(chunks) => discontinuity |= self.streams[si].merge_refresh(chunks),
546                Err(_) => {
547                    self.outputs
548                        .push_back(Output::Dropped(Dropped::Manifest(ManifestFault::Timeline)));
549                }
550            }
551            self.streams[si].index = found.clone();
552        }
553        if discontinuity {
554            self.outputs.push_back(Output::Discontinuity);
555        }
556    }
557
558    fn try_announce(&mut self) {
559        let live: Vec<usize> = (0..self.streams.len())
560            .filter(|&i| !self.streams[i].abandoned)
561            .collect();
562        if self.announced || live.is_empty() {
563            return;
564        }
565        if live
566            .iter()
567            .any(|&i| self.streams[i].local_track_id.is_none())
568        {
569            return;
570        }
571        let mut specs = Vec::new();
572        for (position, &si) in live.iter().enumerate() {
573            let global = TrackId::new(
574                FIRST_TRACK_ID.saturating_add(u32::try_from(position).unwrap_or(u32::MAX)),
575            );
576            let local = self.streams[si].local_track_id.unwrap_or(FIRST_TRACK_ID);
577            let quality = self.streams[si]
578                .held
579                .first()
580                .map_or_else(|| self.streams[si].quality.clone(), |h| h.quality.clone());
581            match self.setup_track(si, local, global, &quality) {
582                Ok(spec) => {
583                    self.streams[si].track = Some(global);
584                    self.streams[si].spec = Some(spec.clone());
585                    specs.push(spec);
586                }
587                Err(reason) => return self.fail(SmoothError::TrackSetup { reason }),
588            }
589        }
590        self.announced = true;
591        self.outputs.push_back(Output::Tracks(specs));
592        for &si in &live {
593            for held in std::mem::take(&mut self.streams[si].held) {
594                self.emit_fragment(si, held.start, &held.body, &held.quality);
595            }
596        }
597    }
598
599    fn setup_track(
600        &mut self,
601        si: usize,
602        local: u32,
603        global: TrackId,
604        quality: &broadcast_smooth::QualityLevel,
605    ) -> Result<TrackSpec, String> {
606        let stream = &self.streams[si];
607        let mut spec = track_spec_from_quality_level(
608            TrackId::new(local),
609            stream.timescale,
610            stream.index.stream_type,
611            quality,
612        )
613        .map_err(|e| e.to_string())?;
614        let init = build_init_segment(std::slice::from_ref(&spec), SYNTHETIC_MOVIE_TIMESCALE)
615            .map_err(|e| e.to_string())?;
616        self.streams[si].init = Some(init);
617        self.streams[si].applied = quality.clone();
618        spec.track_id = global;
619        Ok(spec)
620    }
621
622    fn emit_fragment(
623        &mut self,
624        si: usize,
625        start: u64,
626        body: &[u8],
627        quality: &broadcast_smooth::QualityLevel,
628    ) {
629        if self.streams[si].applied != *quality && !self.reconfigure(si, quality) {
630            return self.outputs.push_back(Output::Dropped(Dropped::Fragment {
631                stream: StreamId(si),
632                status: None,
633            }));
634        }
635        self.emit_samples(si, start, body);
636    }
637
638    fn reconfigure(&mut self, si: usize, quality: &broadcast_smooth::QualityLevel) -> bool {
639        let (Some(local), Some(global)) = (self.streams[si].local_track_id, self.streams[si].track)
640        else {
641            return false;
642        };
643        let Ok(spec) = self.setup_track(si, local, global, quality) else {
644            return false;
645        };
646        if self.streams[si].spec.as_ref() != Some(&spec) {
647            self.streams[si].spec = Some(spec);
648            let specs = self.streams.iter().filter_map(|s| s.spec.clone()).collect();
649            self.outputs.push_back(Output::TracksChanged(specs));
650        }
651        true
652    }
653
654    fn emit_samples(&mut self, si: usize, start: u64, body: &[u8]) {
655        let stream = &self.streams[si];
656        let (Some(init), Some(local), Some(track)) =
657            (stream.init.as_deref(), stream.local_track_id, stream.track)
658        else {
659            return;
660        };
661        match demux_fragment(init, local, body, start) {
662            Ok(samples) => {
663                for sample in samples {
664                    self.outputs.push_back(Output::Sample { track, sample });
665                }
666            }
667            Err(_) => self.outputs.push_back(Output::Dropped(Dropped::Fragment {
668                stream: StreamId(si),
669                status: None,
670            })),
671        }
672    }
673
674    fn advance(&mut self, now: u64) {
675        if self.fatal || self.ended || !self.announced {
676            return;
677        }
678        if !self.streams.iter().all(Stream::is_drained) {
679            return;
680        }
681        if !self.is_live {
682            self.ended = true;
683            self.outputs.push_back(Output::Ended);
684        } else if self.manifest_at.is_none() && matches!(self.manifest, ManifestCursor::Idle) {
685            self.manifest_at = Some(now.saturating_add(self.refresh_wait_us()));
686        }
687    }
688
689    fn refresh_wait_us(&self) -> u64 {
690        let last = self.streams.iter().find(|s| !s.abandoned).and_then(|s| {
691            s.index
692                .enumerate_chunks()
693                .ok()?
694                .last()
695                .map(|c| (c.1, s.timescale))
696        });
697        let half = last.map_or(self.config.refresh_max_us, |(d, ts)| {
698            d.saturating_mul(MICROS_PER_SEC) / u64::from(ts).max(1) / REFRESH_DIVISOR
699        });
700        half.clamp(self.config.refresh_min_us, self.config.refresh_max_us)
701    }
702}
smooth_runtime/client/engine.rs

smooth_runtime/client/
engine.rs
