moq_binary/snapshot/
producer.rs

1//! Publishing a binary value over a track.
2
3use std::sync::{Arc, Mutex};
4
5use bytes::Bytes;
6
7use crate::Result;
8
9pub use super::Config;
10
11/// Publishes a binary value over a track, one value per group.
12///
13/// Each [`update`](Self::update) rolls a new group holding the whole value, so a consumer only ever
14/// needs the newest group and older ones are dropped. For a log where every payload survives, use
15/// [`stream`](crate::stream) instead.
16///
17/// Cheaply clonable: clones share one underlying track, like other MoQ producers.
18#[derive(Clone)]
19pub struct Producer {
20	inner: Arc<Mutex<Inner>>,
21}
22
23impl Producer {
24	/// Create a producer that publishes to the given track.
25	pub fn new(track: moq_net::track::Producer, config: Config) -> Self {
26		Self {
27			inner: Arc::new(Mutex::new(Inner {
28				track,
29				compression: config.compression.is_deflate(),
30			})),
31		}
32	}
33
34	/// Create a subscriber for the underlying track.
35	pub fn consume(&self) -> moq_net::track::Subscriber {
36		self.inner.lock().unwrap().track.subscribe(None)
37	}
38
39	/// Whether any consumer for the underlying track currently exists.
40	///
41	/// The demand signal for a producer serving on request: an unused track is cached state nobody is
42	/// watching, safe to drop and recreate on the next request.
43	pub fn is_used(&self) -> bool {
44		self.inner.lock().unwrap().track.is_used()
45	}
46
47	/// Publish a new value, superseding the previous one.
48	///
49	/// Unlike [`moq-json`](https://docs.rs/moq-json), an identical value is republished rather than
50	/// skipped: comparing two opaque blobs costs a full scan, and only the caller knows whether its
51	/// bytes changed.
52	pub fn update(&mut self, payload: impl Into<Bytes>) -> Result<()> {
53		self.inner.lock().unwrap().update(payload.into())
54	}
55
56	/// Finish the track.
57	pub fn finish(&mut self) -> Result<()> {
58		self.inner.lock().unwrap().finish()
59	}
60}
61
62/// Shared publishing state behind [`Producer`]'s `Arc<Mutex>`.
63struct Inner {
64	track: moq_net::track::Producer,
65	compression: bool,
66}
67
68impl Inner {
69	fn update(&mut self, payload: Bytes) -> Result<()> {
70		// One frame per group, so the window spans a single value and starts cold every time.
71		let payload = match self.compression {
72			true => {
73				// Compression can take a large value under the group's frame limit, but every consumer
74				// decodes with moq-flate's default output cap, so publishing past it would advertise a
75				// value that always fails to read. Reject it here instead.
76				if payload.len() as u64 > moq_flate::DEFAULT_MAX_FRAME_SIZE {
77					return Err(moq_flate::Error::TooLarge(moq_flate::DEFAULT_MAX_FRAME_SIZE).into());
78				}
79				moq_flate::Encoder::new().frame(&payload)
80			}
81			false => payload,
82		};
83
84		// Check before opening a group. `append_group` publishes immediately, so discovering the limit
85		// inside `write_frame` would leave an empty newest group behind: a snapshot consumer jumps to
86		// the newest, so the previous value would be lost even though this update reported an error.
87		if payload.len() as u64 > moq_net::group::MAX_CACHE_BYTES {
88			return Err(moq_net::Error::FrameTooLarge.into());
89		}
90
91		let mut group = self.track.append_group()?;
92		if let Err(err) = group.write_frame(moq_net::Timestamp::now(), payload) {
93			// `append_group` already published this group, and a rejected frame (too large) doesn't
94			// close the track. Dropping the handle does NOT close the group, so leaving it would strand
95			// any subscriber that advanced into it with nothing to read and no end.
96			let _ = group.finish();
97			return Err(err.into());
98		}
99
100		group.finish()?;
101		Ok(())
102	}
103
104	fn finish(&mut self) -> Result<()> {
105		self.track.finish()?;
106		Ok(())
107	}
108}
moq_binary/snapshot/producer.rs

moq_binary/snapshot/
producer.rs
