codecs/decoding/framing/
mod.rs

1//! A collection of framing methods that can be used to convert from byte frames
2//! with defined boundaries to byte chunks.
3
4#![deny(missing_docs)]
5
6mod bytes;
7mod character_delimited;
8mod chunked_gelf;
9mod length_delimited;
10mod newline_delimited;
11mod octet_counting;
12mod varint_length_delimited;
13
14use std::{any::Any, fmt::Debug};
15
16use ::bytes::Bytes;
17pub use character_delimited::{
18    CharacterDelimitedDecoder, CharacterDelimitedDecoderConfig, CharacterDelimitedDecoderOptions,
19    OversizedAction,
20};
21pub use chunked_gelf::{ChunkedGelfDecoder, ChunkedGelfDecoderConfig, ChunkedGelfDecoderOptions};
22use dyn_clone::DynClone;
23pub use length_delimited::{LengthDelimitedDecoder, LengthDelimitedDecoderConfig};
24pub use newline_delimited::{
25    NewlineDelimitedDecoder, NewlineDelimitedDecoderConfig, NewlineDelimitedDecoderOptions,
26};
27pub use octet_counting::{
28    OctetCountingDecoder, OctetCountingDecoderConfig, OctetCountingDecoderOptions,
29};
30use tokio_util::codec::LinesCodecError;
31pub use varint_length_delimited::{
32    VarintLengthDelimitedDecoder, VarintLengthDelimitedDecoderConfig,
33};
34
35pub use self::bytes::{BytesDecoder, BytesDecoderConfig};
36use super::StreamDecodingError;
37
38/// An error that occurred while producing byte frames from a byte stream / byte
39/// message.
40///
41/// It requires conformance to `TcpError` so that we can determine whether the
42/// error is recoverable or if trying to continue will lead to hanging up the
43/// TCP source indefinitely.
44pub trait FramingError: std::error::Error + StreamDecodingError + Send + Sync + Any {
45    /// Coerces the error to a `dyn Any`.
46    /// This is useful for downcasting the error to a concrete type
47    fn as_any(&self) -> &dyn Any;
48}
49
50impl std::error::Error for BoxedFramingError {}
51
52impl FramingError for std::io::Error {
53    fn as_any(&self) -> &dyn Any {
54        self as &dyn Any
55    }
56}
57
58impl FramingError for LinesCodecError {
59    fn as_any(&self) -> &dyn Any {
60        self as &dyn Any
61    }
62}
63
64impl<T> From<T> for BoxedFramingError
65where
66    T: FramingError + 'static,
67{
68    fn from(value: T) -> Self {
69        Box::new(value)
70    }
71}
72
73/// A `Box` containing a `FramingError`.
74pub type BoxedFramingError = Box<dyn FramingError>;
75
76impl StreamDecodingError for BoxedFramingError {
77    fn can_continue(&self) -> bool {
78        self.as_ref().can_continue()
79    }
80}
81
82/// Produce byte frames from a byte stream / byte message.
83pub trait Framer:
84    tokio_util::codec::Decoder<Item = Bytes, Error = BoxedFramingError> + DynClone + Debug + Send + Sync
85{
86}
87
88/// Default implementation for `Framer`s that implement
89/// `tokio_util::codec::Decoder`.
90impl<Decoder> Framer for Decoder where
91    Decoder: tokio_util::codec::Decoder<Item = Bytes, Error = BoxedFramingError>
92        + Clone
93        + Debug
94        + Send
95        + Sync
96{
97}
98
99dyn_clone::clone_trait_object!(Framer);
100
101/// A `Box` containing a `Framer`.
102pub type BoxedFramer = Box<dyn Framer>;
codecs/decoding/framing/mod.rs

codecs/decoding/framing/
mod.rs
