moq_json/snapshot/
consumer.rs

1//! Consuming a JSON value from a track: a [`Decoder`] plus the track it reads from.
2
3use std::task::Poll;
4
5use serde::de::DeserializeOwned;
6
7use super::{ConsumerConfig, Decoder};
8use crate::Result;
9
10/// Consumes a JSON value from a track, reconstructing it from snapshots and deltas.
11///
12/// A [`Decoder`] that owns its track: it reads groups, routes each frame by its position, and
13/// yields the reconstructed value. When something else already owns the track, use the [`Decoder`]
14/// directly.
15pub struct Consumer<T> {
16	track: moq_net::track::Subscriber,
17	group: Option<moq_net::group::Consumer>,
18	decoder: Decoder<T>,
19	frames_read: usize,
20}
21
22impl<T: DeserializeOwned> Consumer<T> {
23	/// Create a consumer reading from the given track subscriber.
24	///
25	/// Set [`ConsumerConfig::compression`] to read a track written by a producer with
26	/// [`ProducerConfig::compression`](super::ProducerConfig::compression) on.
27	pub fn new(track: moq_net::track::Subscriber, config: ConsumerConfig) -> Self {
28		Self {
29			track,
30			group: None,
31			decoder: Decoder::new(config),
32			frames_read: 0,
33		}
34	}
35
36	/// Get the next reconstructed value, or `None` once the track ends.
37	pub async fn next(&mut self) -> Result<Option<T>>
38	where
39		T: Unpin,
40	{
41		kio::wait(|waiter| self.poll_next(waiter)).await
42	}
43
44	/// Poll for the next reconstructed value, without blocking.
45	///
46	/// Jumps to the newest group, reads its snapshot, and applies deltas in order. All frames already
47	/// buffered in the group are applied in one poll but only the resulting *latest* value is yielded:
48	/// the intermediate reconstructions are stale, so a late joiner (or any consumer that has fallen
49	/// behind) catches up to the head in a single step instead of replaying every superseded state.
50	/// Frames must still be decoded in order (the DEFLATE window and merge patches are sequential);
51	/// only the per-frame deserialize and yield are skipped. Switching to a newer group discards the
52	/// older one.
53	pub fn poll_next(&mut self, waiter: &kio::Waiter) -> Poll<Result<Option<T>>> {
54		// Drain to the newest group, resetting reconstruction state whenever we switch.
55		let track_finished = loop {
56			match self.track.poll_next_group(waiter)? {
57				Poll::Ready(Some(group)) => {
58					self.group = Some(group);
59					// The next frame is the new group's snapshot, which also restarts the decoder's window.
60					self.frames_read = 0;
61				}
62				Poll::Ready(None) => break true,
63				Poll::Pending => break false,
64			}
65		};
66
67		// Apply every frame currently buffered in the group, tracking whether any moved us forward and
68		// whether the group is still open with nothing buffered yet (vs. exhausted).
69		// `poll_read_frame` returns an owned `Poll`, so the borrow of `self.group` ends before the
70		// match arms, leaving `apply` (and clearing the group) free to take `&mut self`.
71		let mut advanced = false;
72		let mut group_pending = false;
73		while let Some(group) = &mut self.group {
74			match group.poll_read_frame(waiter)? {
75				Poll::Ready(Some(frame)) => {
76					self.apply(&frame.payload)?;
77					advanced = true;
78				}
79				// The current group is exhausted; wait for a newer one.
80				Poll::Ready(None) => {
81					self.group = None;
82					break;
83				}
84				// The group is still open but has nothing buffered yet.
85				Poll::Pending => {
86					group_pending = true;
87					break;
88				}
89			}
90		}
91
92		if advanced {
93			// Deserialize once, from the head of the backlog we just drained.
94			return Poll::Ready(Ok(self.decoder.decode()?));
95		}
96
97		// An open group may still deliver frames even after the track finishes (it was appended before
98		// the finish), so wait on it rather than ending the stream.
99		if group_pending {
100			return Poll::Pending;
101		}
102
103		if track_finished {
104			Poll::Ready(Ok(None))
105		} else {
106			Poll::Pending
107		}
108	}
109
110	/// Apply one frame: frame 0 of a group is a snapshot, the rest are merge patches.
111	fn apply(&mut self, payload: &[u8]) -> Result<()> {
112		match self.frames_read {
113			0 => self.decoder.snapshot(payload)?,
114			_ => self.decoder.delta(payload)?,
115		}
116		self.frames_read += 1;
117		Ok(())
118	}
119}
moq_json/snapshot/consumer.rs

moq_json/snapshot/
consumer.rs
