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::{ConsumerConfig, Decoder};
8use crate::Result;
9
10/// Consumes an ordered log of JSON records from a track, yielding every record in order.
11///
12/// A [`Decoder`] that owns its track: it reads one record per frame, in order, and starts the
13/// decoder on a cold window at each group boundary. A [`Producer`](super::Producer) writes the whole
14/// log into one group, but a publisher that rolls its own (the way an
15/// [`Encoder`](super::Encoder) desync is cleared) is read here too. When something else already owns
16/// the track, use the [`Decoder`] directly.
17pub struct Consumer<T> {
18	track: moq_net::track::Subscriber,
19	group: Option<moq_net::group::Consumer>,
20	decoder: Decoder<T>,
21}
22
23impl<T: DeserializeOwned> Consumer<T> {
24	/// Create a consumer reading from the given track subscriber.
25	///
26	/// Set [`ConsumerConfig::compression`] to read a track written by a producer with
27	/// [`ProducerConfig::compression`](super::ProducerConfig::compression) on.
28	pub fn new(track: moq_net::track::Subscriber, config: ConsumerConfig) -> Self {
29		Self {
30			track,
31			group: None,
32			decoder: Decoder::new(config),
33		}
34	}
35
36	/// Get the next record, 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 record, without blocking.
45	pub fn poll_next(&mut self, waiter: &kio::Waiter) -> Poll<Result<Option<T>>> {
46		loop {
47			let Some(group) = &mut self.group else {
48				match self.track.poll_next_group(waiter)? {
49					Poll::Ready(Some(group)) => {
50						// Each group is its own compressed stream, so the window starts cold.
51						self.decoder.reset();
52						self.group = Some(group);
53						continue;
54					}
55					Poll::Ready(None) => return Poll::Ready(Ok(None)),
56					Poll::Pending => return Poll::Pending,
57				}
58			};
59
60			match group.poll_read_frame(waiter)? {
61				Poll::Ready(Some(frame)) => return Poll::Ready(Ok(Some(self.decoder.decode(&frame.payload)?))),
62				Poll::Ready(None) => {
63					// This group is exhausted. Clear it and poll for a later one, which starts its own
64					// window; the stream ends only when the track does.
65					self.group = None;
66				}
67				Poll::Pending => return Poll::Pending,
68			}
69		}
70	}
71}
moq_json/stream/consumer.rs

moq_json/stream/
consumer.rs
