feat(iori): add definition for discontinuity

Part of #45

Author: Yesterday17

crates/iori/src/hls/segment.rs

@@ -23,6 +23,8 @@ pub struct M3u8Segment {
     /// Media sequence id from the m3u8 file
     pub media_sequence: u64,
 
+    pub part_index: u64,
+
     pub duration: f64,
     pub format: SegmentFormat,
 }
@@ -59,6 +61,10 @@ impl StreamingSegment for M3u8Segment {
     fn format(&self) -> SegmentFormat {
         self.format.clone()
     }
+
+    fn part_index(&self) -> u64 {
+        self.part_index
+    }
 }
 
 impl RemoteStreamingSegment for M3u8Segment {

crates/iori/src/hls/source.rs

@@ -166,6 +166,7 @@ impl HlsMediaPlaylistSource {
                 initial_segment: initial_segment.clone(),
                 sequence: self.sequence.fetch_add(1, Ordering::Relaxed),
                 media_sequence,
+                part_index: segment.part_index,
                 byte_range: segment.byte_range.as_ref().map(|r| crate::ByteRange {
                     offset: r.offset.unwrap_or(next_range_start),
                     length: Some(r.length),

crates/iori/src/lib.rs

@@ -58,6 +58,7 @@ pub trait StreamingSource {
     ) -> impl Future<Output = IoriResult<impl Stream<Item = IoriResult<Vec<Self::Segment>>>>>;
 }
 
+/// A segment of a streaming source
 pub trait StreamingSegment {
     /// Stream id
     fn stream_id(&self) -> u64;
@@ -66,8 +67,17 @@ pub trait StreamingSegment {
     fn stream_type(&self) -> StreamType;
 
     /// Sequence ID of the segment, starts from 0
+    ///
+    /// It MUST be contiguous and monotonically increasing for a given stream.
     fn sequence(&self) -> u64;
 
+    /// Part index of the segment.
+    ///
+    /// For segments without discontinuities, this defaults to 0.
+    fn part_index(&self) -> u64 {
+        0
+    }
+
     /// File name of the segment
     fn file_name(&self) -> &str;
 

crates/iori/src/merge/pipe.rs

@@ -22,6 +22,7 @@ type SendSegment = (
 /// PipeMerger is a merger that pipes the segments directly to the output.
 ///
 /// If there are any missing segments, it will skip them.
+/// PipeMerger does not and can not handle discontinuities.
 pub struct PipeMerger {
     recycle: bool,
 

crates/iori/src/segment.rs

@@ -74,6 +74,7 @@ pub struct SegmentInfo {
     pub key: Option<std::sync::Arc<IoriKey>>,
     pub duration: Option<f64>,
     pub format: SegmentFormat,
+    pub part_index: u64,
 }
 
 impl<T> From<&T> for SegmentInfo
@@ -90,6 +91,7 @@ where
             duration: segment.duration(),
             stream_type: segment.stream_type(),
             format: segment.format(),
+            part_index: segment.part_index(),
         }
     }
 }
@@ -126,6 +128,10 @@ impl StreamingSegment for Box<dyn StreamingSegment + Send + Sync + '_> {
     fn format(&self) -> SegmentFormat {
         self.as_ref().format()
     }
+
+    fn part_index(&self) -> u64 {
+        self.as_ref().part_index()
+    }
 }
 
 impl StreamingSegment for &Box<dyn StreamingSegment + Send + Sync + '_> {
@@ -160,6 +166,10 @@ impl StreamingSegment for &Box<dyn StreamingSegment + Send + Sync + '_> {
     fn format(&self) -> SegmentFormat {
         self.as_ref().format()
     }
+
+    fn part_index(&self) -> u64 {
+        self.as_ref().part_index()
+    }
 }
 
 #[derive(Debug, Clone, Copy, PartialEq, Eq, Default, Serialize, Deserialize)]