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

moq_json/stream/
encoder.rs
