moq_binary/snapshot/
consumer.rs

1//! Consuming a binary value from a track.
2
3use std::task::Poll;
4
5use bytes::Bytes;
6
7use crate::Result;
8
9pub use super::Config;
10
11/// Consumes a binary value from a track, yielding the newest one.
12///
13/// Jumps to the newest group and reads the value out of it, so a late joiner starts at the current
14/// value rather than replaying superseded ones.
15pub struct Consumer {
16	track: moq_net::track::Ordered,
17	group: Option<moq_net::group::Consumer>,
18	/// The DEFLATE decoder for the current group, `Some` while decompressing. A snapshot group is
19	/// normally one frame, but the window is per group either way.
20	flate: Option<moq_flate::Decoder>,
21	compression: bool,
22}
23
24impl Consumer {
25	/// Create a consumer reading from the given track subscriber.
26	///
27	/// Set [`Config::compression`] to match the producer that wrote the track.
28	pub fn new(track: moq_net::track::Subscriber, config: Config) -> Self {
29		Self {
30			track: track.ordered(),
31			group: None,
32			flate: None,
33			compression: config.compression.is_deflate(),
34		}
35	}
36
37	/// Get the next value, or `None` once the track ends.
38	pub async fn next(&mut self) -> Result<Option<Bytes>> {
39		kio::wait(|waiter| self.poll_next(waiter)).await
40	}
41
42	/// Poll for the next value, without blocking.
43	///
44	/// Jumps to the newest group and drains everything buffered in it, yielding only the last value:
45	/// the earlier ones are already superseded, so a consumer that has fallen behind catches up to
46	/// the head in a single step. A compressed group's frames are still decoded in order, since they
47	/// share one window; only the yield is skipped. Switching to a newer group discards the older one.
48	///
49	/// A group the transport can no longer serve is discarded the same way, not reported: on a
50	/// snapshot track its content is superseded by definition, so the reader waits for the
51	/// replacement. Only a failure of the track itself ends the stream.
52	pub fn poll_next(&mut self, waiter: &kio::Waiter) -> Poll<Result<Option<Bytes>>> {
53		// Drain to the newest group, starting a cold window whenever we switch.
54		let track_finished = loop {
55			match self.track.poll_next_group(waiter)? {
56				Poll::Ready(Some(group)) => {
57					self.group = Some(group);
58					self.flate = self.compression.then(moq_flate::Decoder::new);
59				}
60				Poll::Ready(None) => break true,
61				Poll::Pending => break false,
62			}
63		};
64
65		// Decode every frame currently buffered in the group, keeping only the last.
66		// `poll_read_frame` returns an owned `Poll`, so the borrow of `self.group` ends before the
67		// match arms, leaving `decode` (and clearing the group) free to take `&mut self`.
68		let mut latest = None;
69		let mut group_pending = false;
70		while let Some(group) = &mut self.group {
71			match group.poll_read_frame(waiter) {
72				Poll::Ready(Ok(Some(frame))) => latest = Some(self.decode(&frame.payload)?),
73				// The current group is exhausted; wait for a newer one.
74				Poll::Ready(Ok(None)) => {
75					self.group = None;
76					break;
77				}
78				// The transport can no longer serve the rest of this group: it was superseded and
79				// reclaimed (`Old`), dropped under memory pressure (`Evicted`), or read past the
80				// drift budget (`Lagged`). A snapshot reader only ever wants the newest value, so a
81				// group whose content is gone is never fatal: drop it and wait for its replacement.
82				// A track- or session-level failure still arrives through `poll_next_group` above.
83				Poll::Ready(Err(err)) => {
84					let sequence = group.sequence;
85					self.group = None;
86					tracing::warn!(
87						track = self.track.name(),
88						group = sequence,
89						error = ?err,
90						"snapshot group lost; waiting for a newer one"
91					);
92					break;
93				}
94				// The group is still open but has nothing buffered yet.
95				Poll::Pending => {
96					group_pending = true;
97					break;
98				}
99			}
100		}
101
102		if let Some(payload) = latest {
103			return Poll::Ready(Ok(Some(payload)));
104		}
105
106		// An open group may still deliver frames even after the track finishes (it was appended before
107		// the finish), so wait on it rather than ending the stream.
108		if group_pending {
109			return Poll::Pending;
110		}
111
112		match track_finished {
113			true => Poll::Ready(Ok(None)),
114			false => Poll::Pending,
115		}
116	}
117
118	/// Decompress one frame, if the track is compressed.
119	fn decode(&mut self, payload: &Bytes) -> Result<Bytes> {
120		Ok(match self.flate.as_mut() {
121			Some(flate) => flate.frame(payload)?,
122			None => payload.clone(),
123		})
124	}
125}
moq_binary/snapshot/consumer.rs

moq_binary/snapshot/
consumer.rs
