playout_runtime/
schedule.rs

1//! The channel schedule: an ordered list of entries (programme / ad / slate).
2//!
3//! The wire/data format here is deliberately **ours to define** — issue
4//! [#748](https://github.com/fishloa/rust-broadcast/issues/748)'s
5//! design-decision comment: SCTE 224 (ESNI) covers policy/blackout
6//! signalling, not channel assembly, and adopting it would be cargo-culting a
7//! standard that does not answer this question. There is no applicable open
8//! standard for "ordered list of sources with planned starts" to transcribe,
9//! so [`Schedule`] is a plain in-memory model, not a wire type: it has no
10//! `Parse`/`Serialize` pair, and it is deliberately minimal — enough to plan
11//! [transitions](crate::transition) and the SCTE-35 cues they imply, nothing
12//! about rights, geography, or blackout rules.
13
14use crate::error::{Error, Result};
15use alloc::string::String;
16use alloc::vec::Vec;
17
18/// What kind of content a [`ScheduleEntry`] carries.
19#[derive(Debug, Clone, Copy, PartialEq, Eq)]
20#[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))]
21#[non_exhaustive]
22pub enum EntryKind {
23    /// Regular programme content.
24    Programme,
25    /// An advertisement — a scheduled ad-break asset.
26    Ad,
27    /// Filler/slate content played when nothing else is scheduled (e.g. a
28    /// hold slide, or gap-filler between programme and ad).
29    Slate,
30}
31
32impl EntryKind {
33    /// Stable label.
34    pub fn name(&self) -> &'static str {
35        match self {
36            EntryKind::Programme => "programme",
37            EntryKind::Ad => "ad",
38            EntryKind::Slate => "slate",
39        }
40    }
41}
42broadcast_common::impl_spec_display!(EntryKind);
43
44/// Caller-supplied opaque identity for a source's codec configuration (its
45/// `avcC`/`hvcC`/`esds`, etc.), used only for equality comparison.
46///
47/// This crate never parses codec-configuration bitstreams — the "no
48/// transcoding or conforming" non-goal in issue #748's design decision — so
49/// it cannot compute this itself. The caller (whatever demuxed the source,
50/// e.g. `transmux`'s init-segment parsing) computes a stable fingerprint of
51/// its own config and passes it here; a change in fingerprint between
52/// consecutive entries is exactly what
53/// [`TransitionPlan::discontinuity`](crate::transition::TransitionPlan::discontinuity)
54/// treats as a discontinuity to signal downstream (e.g. via
55/// `broadcast_hls::mark_init_discontinuities`) — never something this crate
56/// re-encodes or conforms.
57#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Default)]
58#[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))]
59pub struct CodecConfigId(pub u64);
60
61/// One entry in the schedule: a source, its kind, and when it is planned to
62/// start on the shared channel timeline.
63#[derive(Debug, Clone, PartialEq, Eq)]
64#[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))]
65pub struct ScheduleEntry {
66    /// Caller-chosen identity for the source this entry plays (e.g. an asset
67    /// id or URI) — opaque to this crate beyond equality/display.
68    pub id: String,
69    /// What kind of content this entry is.
70    pub kind: EntryKind,
71    /// The channel-timeline instant (caller's clock unit — SCTE-35 cues use
72    /// 90 kHz ticks, so that's the natural choice) this entry is scheduled to
73    /// start at.
74    pub planned_start: u64,
75    /// The PTS value the source's own first presented sample carries, in the
76    /// same clock unit. Needed to compute the PTS-rebase offset at the
77    /// transition into this entry — see [`crate::transition`].
78    pub source_start_pts: u64,
79    /// Codec-config identity, for discontinuity detection at the transition
80    /// into this entry.
81    pub codec_config: CodecConfigId,
82}
83
84/// An ordered schedule: entries strictly increasing by `planned_start`.
85///
86/// This crate does not sort entries for you — [`Schedule::push`] rejects an
87/// out-of-order entry immediately, so a caller building a schedule sees
88/// ordering mistakes at construction time rather than as a silently
89/// misordered playout later.
90#[derive(Debug, Clone, Default, PartialEq, Eq)]
91#[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))]
92pub struct Schedule {
93    entries: Vec<ScheduleEntry>,
94}
95
96impl Schedule {
97    /// An empty schedule.
98    #[must_use]
99    pub fn new() -> Self {
100        Schedule {
101            entries: Vec::new(),
102        }
103    }
104
105    /// Append `entry`. Errors with [`Error::OutOfOrder`] if `entry` does not
106    /// start strictly after the schedule's current last entry.
107    pub fn push(&mut self, entry: ScheduleEntry) -> Result<()> {
108        if let Some(last) = self.entries.last()
109            && entry.planned_start <= last.planned_start
110        {
111            return Err(Error::OutOfOrder {
112                prev: last.planned_start,
113                next: entry.planned_start,
114            });
115        }
116        self.entries.push(entry);
117        Ok(())
118    }
119
120    /// The entries, in schedule order.
121    #[must_use]
122    pub fn entries(&self) -> &[ScheduleEntry] {
123        &self.entries
124    }
125
126    /// Number of entries.
127    #[must_use]
128    pub fn len(&self) -> usize {
129        self.entries.len()
130    }
131
132    /// Whether the schedule has no entries.
133    #[must_use]
134    pub fn is_empty(&self) -> bool {
135        self.entries.is_empty()
136    }
137
138    /// The entry active at `pts` — the last entry whose `planned_start <=
139    /// pts` — or `None` if `pts` precedes every entry (or the schedule is
140    /// empty).
141    #[must_use]
142    pub fn active_at(&self, pts: u64) -> Option<&ScheduleEntry> {
143        self.active_index_at(pts).map(|i| &self.entries[i])
144    }
145
146    /// Index form of [`Self::active_at`], for [`crate::transition`] to also
147    /// reach the *next* entry without a second lookup.
148    pub(crate) fn active_index_at(&self, pts: u64) -> Option<usize> {
149        if self.entries.is_empty() {
150            return None;
151        }
152        let idx = self.entries.partition_point(|e| e.planned_start <= pts);
153        if idx == 0 { None } else { Some(idx - 1) }
154    }
155}
156
157#[cfg(test)]
158mod tests {
159    use super::*;
160
161    fn entry(id: &str, kind: EntryKind, planned_start: u64) -> ScheduleEntry {
162        ScheduleEntry {
163            id: id.into(),
164            kind,
165            planned_start,
166            source_start_pts: 0,
167            codec_config: CodecConfigId(1),
168        }
169    }
170
171    #[test]
172    fn push_accepts_strictly_increasing_starts() {
173        let mut sched = Schedule::new();
174        sched
175            .push(entry("prog-1", EntryKind::Programme, 0))
176            .unwrap();
177        sched.push(entry("ad-1", EntryKind::Ad, 1_000)).unwrap();
178        assert_eq!(sched.len(), 2);
179    }
180
181    #[test]
182    fn push_rejects_non_increasing_start() {
183        let mut sched = Schedule::new();
184        sched
185            .push(entry("prog-1", EntryKind::Programme, 1_000))
186            .unwrap();
187        let err = sched.push(entry("ad-1", EntryKind::Ad, 1_000)).unwrap_err();
188        match err {
189            Error::OutOfOrder { prev, next } => {
190                assert_eq!(prev, 1_000);
191                assert_eq!(next, 1_000);
192            }
193            other => panic!("expected OutOfOrder, got {other:?}"),
194        }
195
196        let err = sched.push(entry("ad-2", EntryKind::Ad, 500)).unwrap_err();
197        assert!(matches!(
198            err,
199            Error::OutOfOrder {
200                prev: 1_000,
201                next: 500
202            }
203        ));
204    }
205
206    #[test]
207    fn active_at_finds_the_last_entry_starting_at_or_before_pts() {
208        let mut sched = Schedule::new();
209        sched
210            .push(entry("prog-1", EntryKind::Programme, 0))
211            .unwrap();
212        sched.push(entry("ad-1", EntryKind::Ad, 1_000)).unwrap();
213        sched
214            .push(entry("prog-2", EntryKind::Programme, 2_000))
215            .unwrap();
216
217        assert_eq!(sched.active_at(0).unwrap().id, "prog-1");
218        assert_eq!(sched.active_at(999).unwrap().id, "prog-1");
219        assert_eq!(sched.active_at(1_000).unwrap().id, "ad-1");
220        assert_eq!(sched.active_at(1_999).unwrap().id, "ad-1");
221        assert_eq!(sched.active_at(2_000).unwrap().id, "prog-2");
222        assert_eq!(sched.active_at(1_000_000).unwrap().id, "prog-2");
223    }
224
225    #[test]
226    fn active_at_before_the_first_entry_is_none() {
227        let mut sched = Schedule::new();
228        sched
229            .push(entry("prog-1", EntryKind::Programme, 1_000))
230            .unwrap();
231        assert!(sched.active_at(999).is_none());
232    }
233
234    #[test]
235    fn empty_schedule_has_no_active_entry() {
236        let sched = Schedule::new();
237        assert!(sched.is_empty());
238        assert!(sched.active_at(0).is_none());
239    }
240
241    #[test]
242    fn label_convention() {
243        assert_eq!(EntryKind::Programme.name(), "programme");
244        assert_eq!(alloc::format!("{}", EntryKind::Ad), "ad");
245        assert_eq!(EntryKind::Slate.name(), "slate");
246    }
247}
playout_runtime/schedule.rs

playout_runtime/
schedule.rs
