moq_json/window/
consumer.rs

1//! Consuming a window from a track: a [`Decoder`] plus the track it reads from.
2
3use std::task::{Poll, ready};
4
5use serde::de::DeserializeOwned;
6
7use super::decoder::Codec;
8use super::{ConsumerConfig, Decoder, Event};
9use crate::Result;
10
11/// Consumes a sliding window of JSON records from a track, yielding one event per change.
12///
13/// A [`Decoder`] that owns its track: it reads groups, starts a cold DEFLATE window at each
14/// boundary, and turns each group's header into just the changes this reader has not been told
15/// about. When something else already owns the track, use the [`Decoder`] directly.
16///
17/// Group rolls never surface. A publisher rolls for compression's sake, and a header restating the
18/// window yields nothing for records already delivered, so this reads as one continuous stream of
19/// [`Event`]s regardless of how the publisher framed them.
20pub struct Consumer<T> {
21	track: moq_net::track::Subscriber,
22	group: Option<moq_net::group::Consumer>,
23	codec: Option<Codec>,
24	decoder: Decoder<T>,
25}
26
27impl<T: DeserializeOwned> Consumer<T> {
28	/// Create a consumer reading from the given track subscriber.
29	pub fn new(track: moq_net::track::Subscriber, config: ConsumerConfig) -> Self {
30		Self {
31			track,
32			group: None,
33			codec: None,
34			decoder: Decoder::new(config),
35		}
36	}
37
38	/// Absolute index of the oldest record in the window, and of the next to arrive.
39	pub fn range(&self) -> std::ops::Range<u64> {
40		self.decoder.range()
41	}
42
43	/// Get the next event, or `None` once the track ends.
44	pub async fn next(&mut self) -> Result<Option<Event<T>>>
45	where
46		T: Unpin,
47	{
48		kio::wait(|waiter| self.poll_next(waiter)).await
49	}
50
51	/// Poll for the next event, without blocking.
52	pub fn poll_next(&mut self, waiter: &kio::Waiter) -> Poll<Result<Option<Event<T>>>> {
53		loop {
54			// Drain what the frames already decoded produced before reading more.
55			if let Some(event) = self.decoder.next_event() {
56				return Poll::Ready(Ok(Some(event)));
57			}
58
59			let Some(group) = &mut self.group else {
60				match ready!(self.track.poll_next_group(waiter)?) {
61					Some(group) => {
62						self.codec = Some(Codec::new());
63						self.group = Some(group);
64						continue;
65					}
66					None => return Poll::Ready(Ok(None)),
67				}
68			};
69
70			match ready!(group.poll_read_frame(waiter)?) {
71				Some(frame) => {
72					let codec = self.codec.as_mut().expect("an open MoQ group has a window codec");
73					self.decoder.decode(codec, &frame.payload)?;
74				}
75				None => {
76					// This group is exhausted. Clear it and poll for a later one, which restates the
77					// window; the stream ends only when the track does.
78					self.group = None;
79					self.codec = None;
80				}
81			}
82		}
83	}
84}
moq_json/window/consumer.rs

moq_json/window/
consumer.rs
