Struct Producer

Source

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

Expand description

Publishes a JSON value over a track, choosing snapshots and deltas automatically.

Cheaply clonable: clones share one underlying track and publishing state, like other MoQ producers.

Implementations§

Source §

impl<T> Producer<T>

Source

pub fn consume(&self) -> TrackConsumer

Create a subscriber for the underlying track.

Source §

impl<T: Serialize> Producer<T>

Source

pub fn new(track: TrackProducer, config: ProducerConfig) -> Self

Create a producer that publishes to the given track.

Examples found in repository ?

examples/telemetry.rs (line 81)

78fn wire_bytes(config: ProducerConfig, ticks: u64) -> usize {
79	let track = moq_net::Track::new("telemetry").produce();
80	let consumer = track.consume();
81	let mut producer = Producer::<Value>::new(track, config);
82
83	for tick in 0..ticks {
84		producer.update(&telemetry(tick)).unwrap();
85	}
86	producer.finish().unwrap();
87
88	// Drain the raw stored frames (compressed if the producer compressed them) and sum their sizes.
89	let waiter = kio::Waiter::noop();
90	let mut total = 0;
91	let mut track = consumer;
92	while let Poll::Ready(Ok(Some(mut group))) = track.poll_next_group(&waiter) {
93		while let Poll::Ready(Ok(Some(frame))) = group.poll_read_frame(&waiter) {
94			total += frame.len();
95		}
96	}
97	total
98}
99
100/// Drive a producer and a live consumer in lockstep, asserting that EVERY tick reconstructs to the
101/// exact input value after decompression and delta application (not just the final one).
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 fn update(&mut self, value: &T) -> Result<()>

Publish a new value, emitting a snapshot or a delta automatically.

Does nothing if the value is unchanged from the previous publish.

Examples found in repository ?

examples/telemetry.rs (line 84)

78fn wire_bytes(config: ProducerConfig, ticks: u64) -> usize {
79	let track = moq_net::Track::new("telemetry").produce();
80	let consumer = track.consume();
81	let mut producer = Producer::<Value>::new(track, config);
82
83	for tick in 0..ticks {
84		producer.update(&telemetry(tick)).unwrap();
85	}
86	producer.finish().unwrap();
87
88	// Drain the raw stored frames (compressed if the producer compressed them) and sum their sizes.
89	let waiter = kio::Waiter::noop();
90	let mut total = 0;
91	let mut track = consumer;
92	while let Poll::Ready(Ok(Some(mut group))) = track.poll_next_group(&waiter) {
93		while let Poll::Ready(Ok(Some(frame))) = group.poll_read_frame(&waiter) {
94			total += frame.len();
95		}
96	}
97	total
98}
99
100/// Drive a producer and a live consumer in lockstep, asserting that EVERY tick reconstructs to the
101/// exact input value after decompression and delta application (not just the final one).
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 fn lock(&mut self) -> Guard<'_, T>
where T: Default + DeserializeOwned,

Lock the current value for in-place editing, publishing on drop.

The returned Guard derefs to the last-published value (or T::default() if nothing has been published yet). Editing it through DerefMut marks the guard dirty; when a dirty guard drops it publishes the result, a no-op if unchanged.

This is the counterpart to a callback: hold the guard, mutate, drop. The guard holds the producer’s lock for its lifetime, so independent owners are serialized: each one starts from the latest value and their changes compose instead of clobbering. Don’t hold a guard across an .await, since that keeps the lock held while suspended.

Source

pub fn finish(&mut self) -> Result<()>

Finish the track, closing any open group.

Examples found in repository ?

examples/telemetry.rs (line 86)

78fn wire_bytes(config: ProducerConfig, ticks: u64) -> usize {
79	let track = moq_net::Track::new("telemetry").produce();
80	let consumer = track.consume();
81	let mut producer = Producer::<Value>::new(track, config);
82
83	for tick in 0..ticks {
84		producer.update(&telemetry(tick)).unwrap();
85	}
86	producer.finish().unwrap();
87
88	// Drain the raw stored frames (compressed if the producer compressed them) and sum their sizes.
89	let waiter = kio::Waiter::noop();
90	let mut total = 0;
91	let mut track = consumer;
92	while let Poll::Ready(Ok(Some(mut group))) = track.poll_next_group(&waiter) {
93		while let Poll::Ready(Ok(Some(frame))) = group.poll_read_frame(&waiter) {
94			total += frame.len();
95		}
96	}
97	total
98}
99
100/// Drive a producer and a live consumer in lockstep, asserting that EVERY tick reconstructs to the
101/// exact input value after decompression and delta application (not just the final one).
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}