moq_json/stream/
consumer.rs

1//! Consuming an ordered log from a track: a [`Decoder`] plus the track it reads from.
2
3use std::task::Poll;
4
5use serde::de::DeserializeOwned;
6
7use super::Decoder;
8
9pub use super::Config;
10use crate::Result;
11
12/// Consumes an ordered log of JSON records from a track, yielding every record in order.
13///
14/// A [`Decoder`] that owns its track: it reads one record per frame, in order. The log is a single
15/// group, which is what makes the mode lossless: rolling to a second group means the records that
16/// would have completed the first are gone, so a [`Producer`](super::Producer) that cannot write
17/// ends the track instead. A second group is therefore a broken publisher, and reading it would
18/// present a gap as a continuous log, so it fails with [`Error::Rolled`](crate::Error::Rolled)
19/// rather than yielding the remainder. When something else already owns the track, use the
20/// [`Decoder`] directly.
21///
22/// The failure does not wait for the first group to end: whatever has already arrived in it is
23/// yielded, and the read then fails rather than blocking on a group a broken publisher may never
24/// finish.
25pub struct Consumer<T> {
26	track: moq_net::track::Subscriber,
27	group: Option<moq_net::group::Consumer>,
28	/// Whether the log's one group has been taken, so a second is a rolled log rather than the first.
29	taken: bool,
30	/// Sticky once a second group is seen: the records it displaced are gone, so every later read
31	/// fails too rather than reporting the rest of the log as a whole one.
32	rolled: bool,
33	decoder: Decoder<T>,
34}
35
36impl<T: DeserializeOwned> Consumer<T> {
37	/// Create a consumer reading from the given track subscriber.
38	///
39	/// Set [`Config::compression`] to match the encoder that wrote the track.
40	pub fn new(track: moq_net::track::Subscriber, config: Config) -> Self {
41		Self {
42			track,
43			group: None,
44			taken: false,
45			rolled: false,
46			decoder: Decoder::new(config),
47		}
48	}
49
50	/// Get the next record, or `None` once the track ends.
51	pub async fn next(&mut self) -> Result<Option<T>>
52	where
53		T: Unpin,
54	{
55		kio::wait(|waiter| self.poll_next(waiter)).await
56	}
57
58	/// Poll for the next record, without blocking.
59	pub fn poll_next(&mut self, waiter: &kio::Waiter) -> Poll<Result<Option<T>>> {
60		loop {
61			if self.rolled {
62				return Poll::Ready(Err(crate::Error::Rolled));
63			}
64
65			let Some(group) = &mut self.group else {
66				// Arrival order rather than sequence order, because there is only ever one group to
67				// take and a second one has to be seen whatever its sequence. The monotonic
68				// `poll_next_group` would drop a late lower sequence, which is the very loss this
69				// has to report.
70				match self.track.poll_recv_group(waiter)? {
71					Poll::Ready(Some(_)) if self.taken => self.rolled = true,
72					Poll::Ready(Some(group)) => {
73						self.taken = true;
74						self.decoder.reset();
75						self.group = Some(group);
76					}
77					Poll::Ready(None) => return Poll::Ready(Ok(None)),
78					Poll::Pending => return Poll::Pending,
79				}
80				continue;
81			};
82
83			match group.poll_read_frame(waiter)? {
84				Poll::Ready(Some(frame)) => return Poll::Ready(Ok(Some(self.decoder.decode(&frame.payload)?))),
85				Poll::Ready(None) => {
86					// The log's one group is exhausted. Keep polling the track so a clean end still
87					// reports the log as complete, and so a second group is caught as `Rolled`.
88					self.group = None;
89				}
90				// Nothing more in the group yet, so ask the track before parking on it. A publisher
91				// that opens a second group and leaves the first open would otherwise hold this read
92				// open forever, on a log that already lost the records the second one displaced.
93				// Both polls register, so either source wakes this read.
94				Poll::Pending => match self.track.poll_recv_group(waiter)? {
95					Poll::Ready(Some(_)) => self.rolled = true,
96					// A finished track does not truncate the group in hand; its frames may still arrive.
97					Poll::Ready(None) | Poll::Pending => return Poll::Pending,
98				},
99			}
100		}
101	}
102}
moq_json/stream/consumer.rs

moq_json/stream/
consumer.rs
