http_streams_core/
protobuf_format.rs

1//! Length-prefixed protobuf, both directions.
2
3use crate::content_type::ContentType;
4use crate::error::StreamError;
5use crate::format::{
6    DecodeOptions, DefaultFormat, IdentityParser, ItemEncoder, StreamFormat, StreamFormatDecode,
7    StreamFormatEncode,
8};
9use crate::protobuf_len_codec::ProtobufLenPrefixCodec;
10use bytes::BytesMut;
11
12const PROTOBUF_CONTENT_TYPE: &str = "application/x-protobuf-stream";
13/// The stream type this crate emits, plus the two names used for a bare protobuf payload.
14///
15/// Deliberately **not** `application/octet-stream`: accepting it would mean decoding any
16/// unlabelled binary body as protobuf frames.
17const PROTOBUF_ALIASES: &[&str] = &[
18    "application/x-protobuf-stream",
19    "application/x-protobuf",
20    "application/protobuf",
21];
22
23/// Protobuf messages, each preceded by its length as a LEB128 varint.
24#[derive(Debug, Clone, Copy, Default)]
25pub struct ProtobufStreamFormat;
26
27impl ProtobufStreamFormat {
28    /// A length-prefixed protobuf format.
29    pub fn new() -> Self {
30        Self
31    }
32}
33
34impl DefaultFormat for ProtobufStreamFormat {
35    fn default_format() -> Self {
36        Self
37    }
38}
39
40impl StreamFormat for ProtobufStreamFormat {
41    fn format_name(&self) -> &'static str {
42        "protobuf"
43    }
44
45    fn default_content_type(&self) -> &'static str {
46        PROTOBUF_CONTENT_TYPE
47    }
48
49    fn accepts_content_type(&self, ct: &ContentType<'_>) -> bool {
50        ct.matches_any(PROTOBUF_ALIASES)
51    }
52}
53
54/// Per-stream state for [`ProtobufStreamFormat`]. There is none: every frame is self-describing.
55#[derive(Debug, Clone, Copy, Default)]
56pub struct ProtobufEncoder;
57
58impl<T> ItemEncoder<T> for ProtobufEncoder
59where
60    T: prost::Message,
61{
62    fn encode(&mut self, item: &T, _index: u64, buf: &mut BytesMut) -> Result<(), StreamError> {
63        let encoded = item.encode_to_vec();
64        let mut frame = Vec::with_capacity(encoded.len() + 10);
65        prost::encoding::encode_varint(encoded.len() as u64, &mut frame);
66        frame.extend(encoded);
67        buf.extend_from_slice(&frame);
68        Ok(())
69    }
70}
71
72impl<T> StreamFormatEncode<T> for ProtobufStreamFormat
73where
74    T: prost::Message,
75{
76    type Encoder = ProtobufEncoder;
77
78    fn encoder(&self) -> Self::Encoder {
79        ProtobufEncoder
80    }
81}
82
83impl<T> StreamFormatDecode<T> for ProtobufStreamFormat
84where
85    T: prost::Message + Default,
86{
87    type Frame = Result<T, StreamError>;
88    type Framer = ProtobufLenPrefixCodec<T>;
89    type Parser = IdentityParser;
90
91    fn framer(&self, options: &DecodeOptions) -> Self::Framer {
92        ProtobufLenPrefixCodec::new_with_max_length(options.max_obj_len)
93    }
94
95    fn parser(&self) -> Self::Parser {
96        IdentityParser
97    }
98}
http_streams_core/protobuf_format.rs

http_streams_core/
protobuf_format.rs
