Struct Consumer

Source

pub struct Consumer<T> { /* private fields */ }

Expand description

Consumes a JSON value from a track, reconstructing it from snapshots and deltas.

Implementations§

Source §

impl<T: DeserializeOwned> Consumer<T>

Source

pub fn new(track: TrackConsumer, config: ConsumerConfig) -> Self

Create a consumer reading from the given track subscriber.

Set ConsumerConfig::compression to read a track written by a producer with ProducerConfig::compression on.

Examples found in repository ?

examples/telemetry.rs (line 109)

102fn verify(producer_config: ProducerConfig, ticks: u64) {
103	let track = moq_net::Track::new("telemetry").produce();
104	let consumer = track.consume();
105	let mut producer = Producer::<Value>::new(track, producer_config.clone());
106
107	let mut consumer_config = ConsumerConfig::default();
108	consumer_config.compression = producer_config.compression;
109	let mut consumer = moq_json::Consumer::<Value>::new(consumer, consumer_config);
110	let waiter = kio::Waiter::noop();
111
112	for tick in 0..ticks {
113		let expected = telemetry(tick);
114		producer.update(&expected).unwrap();
115		// The producer emits exactly one frame per update, so the live consumer yields exactly one
116		// reconstructed value: it must match the input byte-for-byte after decompression + patching.
117		match consumer.poll_next(&waiter) {
118			Poll::Ready(Ok(Some(value))) => assert_eq!(value, expected, "tick {tick} reconstruction mismatch"),
119			other => panic!("tick {tick}: expected a value, got {other:?}"),
120		}
121	}
122	producer.finish().unwrap();
123
124	// Drain: nothing left and the stream ends cleanly.
125	assert!(
126		matches!(consumer.poll_next(&waiter), Poll::Ready(Ok(None))),
127		"stream did not end cleanly"
128	);
129}
130
131/// A consumer that joins only after the whole stream exists must still rebuild the latest value from
132/// the newest group's snapshot + deltas. For the compressed path this exercises the lazy decoder
133/// replaying the group's already-stored slices to warm its window before decoding the final frame.
134///
135/// Returns how many values the late joiner surfaced to the application: with backlog collapsing this
136/// is far below `ticks`, since stale intermediate reconstructions are applied internally but skipped.
137fn verify_late_joiner(producer_config: ProducerConfig, ticks: u64) -> usize {
138	let track = moq_net::Track::new("telemetry").produce();
139	let consumer = track.consume();
140	let mut producer = Producer::<Value>::new(track, producer_config.clone());
141	for tick in 0..ticks {
142		producer.update(&telemetry(tick)).unwrap();
143	}
144	producer.finish().unwrap();
145
146	let mut consumer_config = ConsumerConfig::default();
147	consumer_config.compression = producer_config.compression;
148	let mut consumer = moq_json::Consumer::<Value>::new(consumer, consumer_config);
149	let waiter = kio::Waiter::noop();
150	let mut last = None;
151	let mut yielded = 0;
152	while let Poll::Ready(Ok(Some(value))) = consumer.poll_next(&waiter) {
153		last = Some(value);
154		yielded += 1;
155	}
156	assert_eq!(
157		last.as_ref(),
158		Some(&telemetry(ticks - 1)),
159		"late joiner reconstruction mismatch"
160	);
161	yielded
162}

Source

pub async fn next(&mut self) -> Result<Option<T>>
where T: Unpin,

Get the next reconstructed value, or None once the track ends.

Source

pub fn poll_next(&mut self, waiter: &Waiter) -> Poll<Result<Option<T>>>

Poll for the next reconstructed value, without blocking.

Jumps to the newest group, reads its snapshot, and applies deltas in order. All frames already buffered in the group are applied in one poll but only the resulting latest value is yielded: the intermediate reconstructions are stale, so a late joiner (or any consumer that has fallen behind) catches up to the head in a single step instead of replaying every superseded state. Frames must still be decoded in order (the DEFLATE window and merge patches are sequential); only the per-frame deserialize and yield are skipped. Switching to a newer group discards the older one.

Examples found in repository ?

examples/telemetry.rs (line 117)

102fn verify(producer_config: ProducerConfig, ticks: u64) {
103	let track = moq_net::Track::new("telemetry").produce();
104	let consumer = track.consume();
105	let mut producer = Producer::<Value>::new(track, producer_config.clone());
106
107	let mut consumer_config = ConsumerConfig::default();
108	consumer_config.compression = producer_config.compression;
109	let mut consumer = moq_json::Consumer::<Value>::new(consumer, consumer_config);
110	let waiter = kio::Waiter::noop();
111
112	for tick in 0..ticks {
113		let expected = telemetry(tick);
114		producer.update(&expected).unwrap();
115		// The producer emits exactly one frame per update, so the live consumer yields exactly one
116		// reconstructed value: it must match the input byte-for-byte after decompression + patching.
117		match consumer.poll_next(&waiter) {
118			Poll::Ready(Ok(Some(value))) => assert_eq!(value, expected, "tick {tick} reconstruction mismatch"),
119			other => panic!("tick {tick}: expected a value, got {other:?}"),
120		}
121	}
122	producer.finish().unwrap();
123
124	// Drain: nothing left and the stream ends cleanly.
125	assert!(
126		matches!(consumer.poll_next(&waiter), Poll::Ready(Ok(None))),
127		"stream did not end cleanly"
128	);
129}
130
131/// A consumer that joins only after the whole stream exists must still rebuild the latest value from
132/// the newest group's snapshot + deltas. For the compressed path this exercises the lazy decoder
133/// replaying the group's already-stored slices to warm its window before decoding the final frame.
134///
135/// Returns how many values the late joiner surfaced to the application: with backlog collapsing this
136/// is far below `ticks`, since stale intermediate reconstructions are applied internally but skipped.
137fn verify_late_joiner(producer_config: ProducerConfig, ticks: u64) -> usize {
138	let track = moq_net::Track::new("telemetry").produce();
139	let consumer = track.consume();
140	let mut producer = Producer::<Value>::new(track, producer_config.clone());
141	for tick in 0..ticks {
142		producer.update(&telemetry(tick)).unwrap();
143	}
144	producer.finish().unwrap();
145
146	let mut consumer_config = ConsumerConfig::default();
147	consumer_config.compression = producer_config.compression;
148	let mut consumer = moq_json::Consumer::<Value>::new(consumer, consumer_config);
149	let waiter = kio::Waiter::noop();
150	let mut last = None;
151	let mut yielded = 0;
152	while let Poll::Ready(Ok(Some(value))) = consumer.poll_next(&waiter) {
153		last = Some(value);
154		yielded += 1;
155	}
156	assert_eq!(
157		last.as_ref(),
158		Some(&telemetry(ticks - 1)),
159		"late joiner reconstruction mismatch"
160	);
161	yielded
162}