moq_json/stream/
producer.rs

1//! Publishing an ordered log over a track: an [`Encoder`] plus the track it writes to.
2
3use std::marker::PhantomData;
4use std::sync::{Arc, Mutex};
5
6use serde::Serialize;
7
8use super::{Encoder, ProducerConfig};
9use crate::Result;
10
11/// Publishes an ordered log of JSON records over a track, one record per frame in a single group.
12///
13/// An [`Encoder`] that owns its track. When something else already owns the track, use the
14/// [`Encoder`] directly.
15///
16/// Cheaply clonable: clones share one underlying track and publishing state, so multiple owners
17/// (e.g. several producers feeding one log) append into a single ordered stream.
18pub struct Producer<T> {
19	inner: Arc<Mutex<Inner<T>>>,
20	_marker: PhantomData<fn(T)>,
21}
22
23impl<T> Clone for Producer<T> {
24	fn clone(&self) -> Self {
25		Self {
26			inner: self.inner.clone(),
27			_marker: PhantomData,
28		}
29	}
30}
31
32impl<T> Producer<T> {
33	/// Create a subscriber for the underlying track.
34	pub fn consume(&self) -> moq_net::track::Subscriber {
35		self.inner.lock().unwrap().track.inner.subscribe(None)
36	}
37}
38
39impl<T: Serialize> Producer<T> {
40	/// Create a producer that publishes to the given track.
41	pub fn new(track: moq_net::track::Producer, config: ProducerConfig) -> Self {
42		Self {
43			inner: Arc::new(Mutex::new(Inner {
44				track: Track {
45					inner: track,
46					group: None,
47				},
48				encoder: Encoder::new(config),
49			})),
50			_marker: PhantomData,
51		}
52	}
53
54	/// Append one record to the log.
55	pub fn append(&mut self, value: &T) -> Result<()> {
56		self.inner.lock().unwrap().append(value)
57	}
58
59	/// Finish the track, closing the group.
60	pub fn finish(&mut self) -> Result<()> {
61		self.inner.lock().unwrap().finish()
62	}
63}
64
65/// Shared publishing state behind [`Producer`]'s `Arc<Mutex>`.
66///
67/// The track and the encoder are separate fields so a [`Pending`](super::Pending) record (which
68/// borrows the encoder) and the write that consumes it (which borrows the track) don't contend for
69/// one `&mut self`.
70struct Inner<T> {
71	track: Track,
72	encoder: Encoder<T>,
73}
74
75impl<T: Serialize> Inner<T> {
76	fn append(&mut self, value: &T) -> Result<()> {
77		// Split the borrow so `record` can hold the encoder while `track` is written through.
78		let Inner { track, encoder } = self;
79
80		// Encode first, so a value that can't be serialized doesn't publish an empty group that
81		// subscribers would advance into and wait on. Opening the group afterwards is safe because
82		// `record` guards the window: any failure below drops it uncommitted.
83		let record = encoder.encode(value)?;
84
85		let result = match track.open() {
86			Ok(()) => track.write(record.payload()),
87			Err(err) => Err(err),
88		};
89
90		if let Err(err) = result {
91			// The record never reached the wire, so dropping it desyncs a compressed encoder. `Track`
92			// has already closed the group it published, which is the group roll that recovery needs;
93			// reset the encoder to finish it, or the desync latch refuses every later record even
94			// though the fresh group could carry one.
95			drop(record);
96			encoder.reset();
97			return Err(err);
98		}
99
100		record.commit();
101		Ok(())
102	}
103
104	fn finish(&mut self) -> Result<()> {
105		self.track.finish()
106	}
107}
108
109/// The track half of [`Inner`]: the single group carrying the whole log.
110struct Track {
111	inner: moq_net::track::Producer,
112	// Opened on the first append and never rolled.
113	group: Option<moq_net::group::Producer>,
114}
115
116impl Track {
117	/// Open the log's group if it isn't already.
118	fn open(&mut self) -> Result<()> {
119		if self.group.is_none() {
120			self.group = Some(self.inner.append_group()?);
121		}
122		Ok(())
123	}
124
125	/// Append one encoded record to the log's group.
126	fn write(&mut self, payload: &bytes::Bytes) -> Result<()> {
127		let group = self.group.as_mut().expect("a group is open");
128		let Err(err) = group.write_frame(moq_net::Timestamp::now(), payload.clone()) else {
129			return Ok(());
130		};
131
132		// The group is already published and dropping the handle does not close it, so a subscriber
133		// that advanced into it would wait there with nothing to read. Close it and let a later append
134		// open a fresh one, which is what a caller recovering from the desync has to do anyway.
135		if let Some(mut group) = self.group.take() {
136			let _ = group.finish();
137		}
138		Err(err.into())
139	}
140
141	fn finish(&mut self) -> Result<()> {
142		if let Some(mut group) = self.group.take() {
143			group.finish()?;
144		}
145		self.inner.finish()?;
146		Ok(())
147	}
148}
moq_json/stream/producer.rs

moq_json/stream/
producer.rs
