moq_json/stream/
encoder.rs

1//! The track-free half of stream publishing: records in, frame payloads out.
2
3use std::marker::PhantomData;
4
5use bytes::Bytes;
6use serde::Serialize;
7
8use crate::{Error, Result};
9
10/// Configuration for an [`Encoder`], and so for the [`Producer`](super::Producer) wrapping one.
11///
12/// Build from [`Default`] and override fields (the struct is `#[non_exhaustive]`, so new
13/// options stay additive).
14#[derive(Debug, Clone, Default)]
15#[non_exhaustive]
16pub struct ProducerConfig {
17	/// Compress the group as one sync-flushed DEFLATE stream, so each record reuses the earlier
18	/// ones as context and shrinks sharply.
19	///
20	/// `false` (the default) emits plaintext JSON frames. A [`Decoder`](super::Decoder) reading them
21	/// must set [`ConsumerConfig::compression`](super::ConsumerConfig::compression) to match.
22	pub compression: bool,
23}
24
25impl ProducerConfig {
26	/// Set [`compression`](Self::compression) (a builder, since the struct is `#[non_exhaustive]`).
27	pub fn with_compression(mut self, compression: bool) -> Self {
28		self.compression = compression;
29		self
30	}
31}
32
33/// An encoded record the caller has not yet acknowledged writing.
34///
35/// Returned by [`Encoder::encode`]. Write the [`payload`](Self::payload), then
36/// [`commit`](Self::commit).
37///
38/// A frame that is never committed never reached the wire. With compression on that is
39/// unrecoverable within the group: the window is ahead of what the consumer holds, and a log has no
40/// keyframe to resynchronize on the way [`snapshot`](crate::snapshot) does. So the encoder refuses
41/// to encode anything further ([`Error::Desync`]) until the caller rolls a new group and calls
42/// [`Encoder::reset`]. Without compression each record stands alone, so a dropped one leaves a gap
43/// in the log but nothing undecodable, and encoding continues.
44#[must_use = "write and commit the record; an uncommitted compressed record stops the encoder"]
45pub struct Pending<'a, T> {
46	encoder: &'a mut Encoder<T>,
47	payload: Bytes,
48	committed: bool,
49}
50
51impl<T> Pending<'_, T> {
52	/// The frame payload to write.
53	pub fn payload(&self) -> &Bytes {
54		&self.payload
55	}
56
57	/// Acknowledge that the record reached the wire, keeping the encoder's window.
58	///
59	/// Only call this once the write has actually succeeded.
60	pub fn commit(mut self) {
61		self.committed = true;
62	}
63}
64
65impl<T> Drop for Pending<'_, T> {
66	fn drop(&mut self) {
67		if !self.committed {
68			self.encoder.desync();
69		}
70	}
71}
72
73/// Encodes JSON records into frame payloads, sharing one DEFLATE window across the log.
74///
75/// The track-free core of [`Producer`](super::Producer). Unlike
76/// [`snapshot::Encoder`](crate::snapshot::Encoder) there are no group boundaries to report: a log is
77/// an unbroken sequence of self-contained records, so every payload is simply the next frame.
78///
79/// The window spans everything encoded so far, so payloads must reach the wire in order and be
80/// decoded in the same order. If the caller does roll a group, call [`reset`](Self::reset) so the
81/// next record starts a cold window that the new group's decoder can follow.
82pub struct Encoder<T> {
83	/// The DEFLATE encoder (one window for the whole log), `Some` while compressing.
84	flate: Option<moq_flate::Encoder>,
85	compression: bool,
86
87	/// Set when a compressed record was encoded but never written. The window is then ahead of the
88	/// consumer for the rest of the group, so encoding stops until the caller rolls a new one.
89	desynced: bool,
90
91	_marker: PhantomData<fn(T)>,
92}
93
94impl<T> Encoder<T> {
95	/// Create an encoder with a cold window.
96	pub fn new(config: ProducerConfig) -> Self {
97		Self {
98			flate: config.compression.then(moq_flate::Encoder::new),
99			compression: config.compression,
100			desynced: false,
101			_marker: PhantomData,
102		}
103	}
104
105	/// Start a cold DEFLATE window, for a caller that has just rolled a group.
106	///
107	/// This is also how a caller clears an [`Error::Desync`]: roll a new group so the consumer starts
108	/// its own cold window, then reset.
109	pub fn reset(&mut self) {
110		self.flate = self.compression.then(moq_flate::Encoder::new);
111		self.desynced = false;
112	}
113
114	/// Mark the window as ahead of the consumer, after a record that was never written.
115	///
116	/// Only meaningful while compressing: an uncompressed record carries no shared state, so losing
117	/// one leaves a gap in the log rather than an undecodable stream.
118	fn desync(&mut self) {
119		self.desynced = self.compression;
120	}
121}
122
123impl<T: Serialize> Encoder<T> {
124	/// Encode one record into the next frame payload.
125	///
126	/// The record comes back as a [`Pending`] the caller writes and then
127	/// [`commit`](Pending::commit)s. Errors with [`Error::Desync`] if a previous compressed record
128	/// was left uncommitted, since every frame after it would be undecodable.
129	pub fn encode(&mut self, value: &T) -> Result<Pending<'_, T>> {
130		if self.desynced {
131			return Err(Error::Desync);
132		}
133
134		let bytes = serde_json::to_vec(value)?;
135		let payload = match self.flate.as_mut() {
136			Some(flate) => flate.frame(&bytes),
137			None => Bytes::from(bytes),
138		};
139
140		Ok(Pending {
141			encoder: self,
142			payload,
143			committed: false,
144		})
145	}
146}
moq_json/stream/encoder.rs

moq_json/stream/
encoder.rs
