codecs/decoding/framing/
character_delimited.rs

1use bytes::{Buf, Bytes, BytesMut};
2use memchr::memchr;
3use tokio_util::codec::Decoder;
4use tracing::{trace, warn};
5use vector_config::configurable_component;
6
7use super::BoxedFramingError;
8
9/// The behavior to apply when a frame exceeds the maximum allowed byte size.
10#[configurable_component]
11#[derive(Copy, Clone, Debug, Default, PartialEq, Eq, PartialOrd, Ord, Hash)]
12#[serde(rename_all = "snake_case")]
13pub enum OversizedAction {
14    /// Drop the entire oversized frame.
15    #[default]
16    Drop,
17
18    /// Truncate the frame to the maximum allowed size and emit the partial content.
19    ///
20    /// The remainder of the oversized frame is discarded up to the next delimiter.
21    Truncate,
22}
23
24/// Config used to build a `CharacterDelimitedDecoder`.
25#[configurable_component]
26#[derive(Debug, Clone)]
27pub struct CharacterDelimitedDecoderConfig {
28    /// Options for the character delimited decoder.
29    pub character_delimited: CharacterDelimitedDecoderOptions,
30}
31
32impl CharacterDelimitedDecoderConfig {
33    /// Creates a `CharacterDelimitedDecoderConfig` with the specified delimiter and default max length.
34    pub const fn new(delimiter: u8) -> Self {
35        Self {
36            character_delimited: CharacterDelimitedDecoderOptions::new(delimiter, None),
37        }
38    }
39    /// Build the `CharacterDelimitedDecoder` from this configuration.
40    pub const fn build(&self) -> CharacterDelimitedDecoder {
41        let oversized_action = self.character_delimited.oversized_action;
42        if let Some(max_length) = self.character_delimited.max_length {
43            CharacterDelimitedDecoder::new_with_max_length(
44                self.character_delimited.delimiter,
45                max_length,
46            )
47            .with_oversized_action(oversized_action)
48        } else {
49            CharacterDelimitedDecoder::new(self.character_delimited.delimiter)
50        }
51    }
52}
53
54/// Options for building a `CharacterDelimitedDecoder`.
55#[configurable_component]
56#[derive(Clone, Debug, PartialEq, Eq)]
57pub struct CharacterDelimitedDecoderOptions {
58    /// The character that delimits byte sequences.
59    #[configurable(metadata(docs::type_override = "ascii_char"))]
60    #[serde(with = "vector_core::serde::ascii_char")]
61    pub delimiter: u8,
62
63    /// The maximum length of the byte buffer.
64    ///
65    /// This length does *not* include the trailing delimiter.
66    ///
67    /// By default, no maximum length is enforced. If events are malformed, this can lead to
68    /// additional resource usage as events continue to be buffered in memory, and can potentially
69    /// lead to memory exhaustion in extreme cases.
70    ///
71    /// If there is a risk of processing malformed data, such as logs with user-controlled input,
72    /// consider setting the maximum length to a reasonably large value as a safety net. This
73    /// prevents processing from being unbounded.
74    #[serde(skip_serializing_if = "vector_core::serde::is_default")]
75    pub max_length: Option<usize>,
76
77    /// The behavior when a frame exceeds `max_length`.
78    ///
79    /// When set to `drop` (the default), the entire oversized frame is discarded.
80    /// When set to `truncate`, the frame is truncated to `max_length` bytes and the
81    /// remainder is discarded up to the next delimiter.
82    ///
83    /// This option has no effect if `max_length` is not set.
84    #[serde(default, skip_serializing_if = "vector_core::serde::is_default")]
85    pub oversized_action: OversizedAction,
86}
87
88impl CharacterDelimitedDecoderOptions {
89    /// Create a `CharacterDelimitedDecoderOptions` with a delimiter and optional max_length.
90    pub const fn new(delimiter: u8, max_length: Option<usize>) -> Self {
91        Self {
92            delimiter,
93            max_length,
94            oversized_action: OversizedAction::Drop,
95        }
96    }
97}
98
99/// A decoder for handling bytes that are delimited by (a) chosen character(s).
100#[derive(Copy, Clone, Debug, Eq, PartialEq, Ord, PartialOrd, Hash)]
101pub struct CharacterDelimitedDecoder {
102    /// The delimiter used to separate byte sequences.
103    pub delimiter: u8,
104    /// The maximum length of the byte buffer.
105    pub max_length: usize,
106    /// The behavior when a frame exceeds `max_length`.
107    pub oversized_action: OversizedAction,
108}
109
110impl CharacterDelimitedDecoder {
111    /// Creates a `CharacterDelimitedDecoder` with the specified delimiter.
112    pub const fn new(delimiter: u8) -> Self {
113        CharacterDelimitedDecoder {
114            delimiter,
115            max_length: usize::MAX,
116            oversized_action: OversizedAction::Drop,
117        }
118    }
119
120    /// Creates a `CharacterDelimitedDecoder` with a maximum frame length limit.
121    ///
122    /// Any frames longer than `max_length` bytes will be discarded entirely.
123    pub const fn new_with_max_length(delimiter: u8, max_length: usize) -> Self {
124        CharacterDelimitedDecoder {
125            max_length,
126            ..CharacterDelimitedDecoder::new(delimiter)
127        }
128    }
129
130    /// Sets the behavior when a frame exceeds `max_length`.
131    pub const fn with_oversized_action(mut self, action: OversizedAction) -> Self {
132        self.oversized_action = action;
133        self
134    }
135
136    /// Returns the maximum frame length when decoding.
137    pub const fn max_length(&self) -> usize {
138        self.max_length
139    }
140}
141
142impl Decoder for CharacterDelimitedDecoder {
143    type Item = Bytes;
144    type Error = BoxedFramingError;
145
146    fn decode(&mut self, buf: &mut BytesMut) -> Result<Option<Bytes>, Self::Error> {
147        loop {
148            // This function has the following goal: we are searching for
149            // sub-buffers delimited by `self.delimiter` with size no more than
150            // `self.max_length`. If a sub-buffer is found that exceeds
151            // `self.max_length` we either discard it or truncate it (depending
152            // on `self.oversized_action`), else we return it. At the end of
153            // the buffer if the delimiter is not present the remainder of the
154            // buffer is discarded.
155            match memchr(self.delimiter, buf) {
156                None => return Ok(None),
157                Some(next_delimiter_idx) => {
158                    if next_delimiter_idx > self.max_length {
159                        match self.oversized_action {
160                            OversizedAction::Drop => {
161                                warn!(
162                                    message = "Discarding frame larger than max_length.",
163                                    buf_len = buf.len(),
164                                    max_length = self.max_length,
165                                );
166                                buf.advance(next_delimiter_idx + 1);
167                            }
168                            OversizedAction::Truncate => {
169                                warn!(
170                                    message = "Truncating frame larger than max_length.",
171                                    original_len = next_delimiter_idx,
172                                    max_length = self.max_length,
173                                );
174                                let frame = buf.split_to(self.max_length).freeze();
175                                // Discard the remaining bytes up to and including the delimiter.
176                                buf.advance(next_delimiter_idx - self.max_length + 1);
177                                return Ok(Some(frame));
178                            }
179                        }
180                    } else {
181                        let frame = buf.split_to(next_delimiter_idx).freeze();
182                        trace!(
183                            message = "Decoding the frame.",
184                            bytes_processed = frame.len()
185                        );
186                        buf.advance(1); // scoot past the delimiter
187                        return Ok(Some(frame));
188                    }
189                }
190            }
191        }
192    }
193
194    fn decode_eof(&mut self, buf: &mut BytesMut) -> Result<Option<Bytes>, Self::Error> {
195        match self.decode(buf)? {
196            Some(frame) => Ok(Some(frame)),
197            None => {
198                if buf.is_empty() {
199                    Ok(None)
200                } else if buf.len() > self.max_length {
201                    match self.oversized_action {
202                        OversizedAction::Drop => {
203                            warn!(
204                                message = "Discarding frame larger than max_length.",
205                                buf_len = buf.len(),
206                                max_length = self.max_length,
207                            );
208                            buf.clear();
209                            Ok(None)
210                        }
211                        OversizedAction::Truncate => {
212                            warn!(
213                                message = "Truncating frame larger than max_length.",
214                                original_len = buf.len(),
215                                max_length = self.max_length,
216                            );
217                            let frame = buf.split_to(self.max_length).freeze();
218                            buf.clear();
219                            Ok(Some(frame))
220                        }
221                    }
222                } else {
223                    let bytes: Bytes = buf.split_to(buf.len()).freeze();
224                    Ok(Some(bytes))
225                }
226            }
227        }
228    }
229}
230
231#[cfg(test)]
232mod tests {
233    use std::collections::HashMap;
234
235    use bytes::BufMut;
236    use indoc::indoc;
237
238    use super::*;
239
240    #[test]
241    fn decode() {
242        let mut codec = CharacterDelimitedDecoder::new(b'\n');
243        let buf = &mut BytesMut::new();
244        buf.put_slice(b"abc\n");
245        assert_eq!(Some("abc".into()), codec.decode(buf).unwrap());
246    }
247
248    #[test]
249    fn decode_max_length() {
250        const MAX_LENGTH: usize = 6;
251
252        let mut codec = CharacterDelimitedDecoder::new_with_max_length(b'\n', MAX_LENGTH);
253        let buf = &mut BytesMut::new();
254
255        // limit is 6 so it will skip longer lines
256        buf.put_slice(b"1234567\n123456\n123412314\n123");
257
258        assert_eq!(codec.decode(buf).unwrap(), Some(Bytes::from("123456")));
259        assert_eq!(codec.decode(buf).unwrap(), None);
260
261        let buf = &mut BytesMut::new();
262
263        // limit is 6 so it will skip longer lines
264        buf.put_slice(b"1234567\n123456\n123412314\n123");
265
266        assert_eq!(codec.decode_eof(buf).unwrap(), Some(Bytes::from("123456")));
267        assert_eq!(codec.decode_eof(buf).unwrap(), Some(Bytes::from("123")));
268        assert_eq!(codec.decode_eof(buf).unwrap(), None);
269    }
270
271    // Regression test for [infinite loop bug](https://github.com/vectordotdev/vector/issues/2564)
272    // Derived from https://github.com/tokio-rs/tokio/issues/1483
273    #[test]
274    fn decode_discard_repeat() {
275        const MAX_LENGTH: usize = 1;
276
277        let mut codec = CharacterDelimitedDecoder::new_with_max_length(b'\n', MAX_LENGTH);
278        let buf = &mut BytesMut::new();
279
280        buf.reserve(200);
281        buf.put(&b"aa"[..]);
282        assert!(codec.decode(buf).unwrap().is_none());
283        buf.put(&b"a"[..]);
284        assert!(codec.decode(buf).unwrap().is_none());
285    }
286
287    #[test]
288    fn decode_json_escaped() {
289        let mut input = HashMap::new();
290        input.insert("key", "value");
291        input.insert("new", "li\nne");
292
293        let mut bytes = serde_json::to_vec(&input).unwrap();
294        bytes.push(b'\n');
295
296        let mut codec = CharacterDelimitedDecoder::new(b'\n');
297        let buf = &mut BytesMut::new();
298
299        buf.reserve(bytes.len());
300        buf.extend(bytes);
301
302        let result = codec.decode(buf).unwrap();
303
304        assert!(result.is_some());
305        assert!(buf.is_empty());
306    }
307
308    #[test]
309    fn decode_json_multiline() {
310        let events = indoc! {r#"
311            {"log":"\u0009at org.springframework.security.web.context.SecurityContextPersistenceFilter.doFilter(SecurityContextPersistenceFilter.java:105)\n","stream":"stdout","time":"2019-01-18T07:49:27.374616758Z"}
312            {"log":"\u0009at org.springframework.security.web.FilterChainProxy$VirtualFilterChain.doFilter(FilterChainProxy.java:334)\n","stream":"stdout","time":"2019-01-18T07:49:27.374640288Z"}
313            {"log":"\u0009at org.springframework.security.web.context.request.async.WebAsyncManagerIntegrationFilter.doFilterInternal(WebAsyncManagerIntegrationFilter.java:56)\n","stream":"stdout","time":"2019-01-18T07:49:27.374655505Z"}
314            {"log":"\u0009at org.springframework.web.filter.OncePerRequestFilter.doFilter(OncePerRequestFilter.java:107)\n","stream":"stdout","time":"2019-01-18T07:49:27.374671955Z"}
315            {"log":"\u0009at org.springframework.security.web.FilterChainProxy$VirtualFilterChain.doFilter(FilterChainProxy.java:334)\n","stream":"stdout","time":"2019-01-18T07:49:27.374690312Z"}
316            {"log":"\u0009at org.springframework.security.web.FilterChainProxy.doFilterInternal(FilterChainProxy.java:215)\n","stream":"stdout","time":"2019-01-18T07:49:27.374704522Z"}
317            {"log":"\u0009at org.springframework.security.web.FilterChainProxy.doFilter(FilterChainProxy.java:178)\n","stream":"stdout","time":"2019-01-18T07:49:27.374718459Z"}
318            {"log":"\u0009at org.springframework.web.filter.DelegatingFilterProxy.invokeDelegate(DelegatingFilterProxy.java:357)\n","stream":"stdout","time":"2019-01-18T07:49:27.374732919Z"}
319            {"log":"\u0009at org.springframework.web.filter.DelegatingFilterProxy.doFilter(DelegatingFilterProxy.java:270)\n","stream":"stdout","time":"2019-01-18T07:49:27.374750799Z"}
320            {"log":"\u0009at org.apache.catalina.core.ApplicationFilterChain.internalDoFilter(ApplicationFilterChain.java:193)\n","stream":"stdout","time":"2019-01-18T07:49:27.374764819Z"}
321            {"log":"\u0009at org.apache.catalina.core.ApplicationFilterChain.doFilter(ApplicationFilterChain.java:166)\n","stream":"stdout","time":"2019-01-18T07:49:27.374778682Z"}
322            {"log":"\u0009at org.springframework.web.filter.RequestContextFilter.doFilterInternal(RequestContextFilter.java:99)\n","stream":"stdout","time":"2019-01-18T07:49:27.374792429Z"}
323            {"log":"\u0009at org.springframework.web.filter.OncePerRequestFilter.doFilter(OncePerRequestFilter.java:107)\n","stream":"stdout","time":"2019-01-18T07:49:27.374805985Z"}
324            {"log":"\u0009at org.apache.catalina.core.ApplicationFilterChain.internalDoFilter(ApplicationFilterChain.java:193)\n","stream":"stdout","time":"2019-01-18T07:49:27.374819625Z"}
325            {"log":"\u0009at org.apache.catalina.core.ApplicationFilterChain.doFilter(ApplicationFilterChain.java:166)\n","stream":"stdout","time":"2019-01-18T07:49:27.374833335Z"}
326            {"log":"\u0009at org.springframework.web.filter.HttpPutFormContentFilter.doFilterInternal(HttpPutFormContentFilter.java:109)\n","stream":"stdout","time":"2019-01-18T07:49:27.374847845Z"}
327            {"log":"\u0009at org.springframework.web.filter.OncePerRequestFilter.doFilter(OncePerRequestFilter.java:107)\n","stream":"stdout","time":"2019-01-18T07:49:27.374861925Z"}
328            {"log":"\u0009at org.apache.catalina.core.ApplicationFilterChain.internalDoFilter(ApplicationFilterChain.java:193)\n","stream":"stdout","time":"2019-01-18T07:49:27.37487589Z"}
329            {"log":"\u0009at org.apache.catalina.core.ApplicationFilterChain.doFilter(ApplicationFilterChain.java:166)\n","stream":"stdout","time":"2019-01-18T07:49:27.374890043Z"}
330            {"log":"\u0009at org.springframework.web.filter.HiddenHttpMethodFilter.doFilterInternal(HiddenHttpMethodFilter.java:93)\n","stream":"stdout","time":"2019-01-18T07:49:27.374903813Z"}
331            {"log":"\u0009at org.springframework.web.filter.OncePerRequestFilter.doFilter(OncePerRequestFilter.java:107)\n","stream":"stdout","time":"2019-01-18T07:49:27.374917793Z"}
332            {"log":"\u0009at org.apache.catalina.core.ApplicationFilterChain.internalDoFilter(ApplicationFilterChain.java:193)\n","stream":"stdout","time":"2019-01-18T07:49:27.374931586Z"}
333            {"log":"\u0009at org.apache.catalina.core.ApplicationFilterChain.doFilter(ApplicationFilterChain.java:166)\n","stream":"stdout","time":"2019-01-18T07:49:27.374946006Z"}
334            {"log":"\u0009at org.springframework.boot.actuate.metrics.web.servlet.WebMvcMetricsFilter.filterAndRecordMetrics(WebMvcMetricsFilter.java:117)\n","stream":"stdout","time":"2019-01-18T07:49:27.37496104Z"}
335            {"log":"\u0009at org.springframework.boot.actuate.metrics.web.servlet.WebMvcMetricsFilter.doFilterInternal(WebMvcMetricsFilter.java:106)\n","stream":"stdout","time":"2019-01-18T07:49:27.37498773Z"}
336            {"log":"\u0009at org.springframework.web.filter.OncePerRequestFilter.doFilter(OncePerRequestFilter.java:107)\n","stream":"stdout","time":"2019-01-18T07:49:27.375003113Z"}
337            {"log":"\u0009at org.apache.catalina.core.ApplicationFilterChain.internalDoFilter(ApplicationFilterChain.java:193)\n","stream":"stdout","time":"2019-01-18T07:49:27.375017063Z"}
338            {"log":"\u0009at org.apache.catalina.core.ApplicationFilterChain.doFilter(ApplicationFilterChain.java:166)\n","stream":"stdout","time":"2019-01-18T07:49:27.37503086Z"}
339            {"log":"\u0009at org.springframework.web.filter.CharacterEncodingFilter.doFilterInternal(CharacterEncodingFilter.java:200)\n","stream":"stdout","time":"2019-01-18T07:49:27.3750454Z"}
340            {"log":"\u0009at org.springframework.web.filter.OncePerRequestFilter.doFilter(OncePerRequestFilter.java:107)\n","stream":"stdout","time":"2019-01-18T07:49:27.37505928Z"}
341            {"log":"\u0009at org.apache.catalina.core.ApplicationFilterChain.internalDoFilter(ApplicationFilterChain.java:193)\n","stream":"stdout","time":"2019-01-18T07:49:27.37507306Z"}
342            {"log":"\u0009at org.apache.catalina.core.ApplicationFilterChain.doFilter(ApplicationFilterChain.java:166)\n","stream":"stdout","time":"2019-01-18T07:49:27.375086726Z"}
343            {"log":"\u0009at org.apache.catalina.core.StandardWrapperValve.invoke(StandardWrapperValve.java:198)\n","stream":"stdout","time":"2019-01-18T07:49:27.375100817Z"}
344            {"log":"\u0009at org.apache.catalina.core.StandardContextValve.invoke(StandardContextValve.java:96)\n","stream":"stdout","time":"2019-01-18T07:49:27.375115354Z"}
345            {"log":"\u0009at org.apache.catalina.authenticator.AuthenticatorBase.invoke(AuthenticatorBase.java:493)\n","stream":"stdout","time":"2019-01-18T07:49:27.375129454Z"}
346            {"log":"\u0009at org.apache.catalina.core.StandardHostValve.invoke(StandardHostValve.java:140)\n","stream":"stdout","time":"2019-01-18T07:49:27.375144001Z"}
347            {"log":"\u0009at org.apache.catalina.valves.ErrorReportValve.invoke(ErrorReportValve.java:81)\n","stream":"stdout","time":"2019-01-18T07:49:27.375157464Z"}
348            {"log":"\u0009at org.apache.catalina.core.StandardEngineValve.invoke(StandardEngineValve.java:87)\n","stream":"stdout","time":"2019-01-18T07:49:27.375170981Z"}
349            {"log":"\u0009at org.apache.catalina.connector.CoyoteAdapter.service(CoyoteAdapter.java:342)\n","stream":"stdout","time":"2019-01-18T07:49:27.375184417Z"}
350            {"log":"\u0009at org.apache.coyote.http11.Http11Processor.service(Http11Processor.java:800)\n","stream":"stdout","time":"2019-01-18T07:49:27.375198024Z"}
351            {"log":"\u0009at org.apache.coyote.AbstractProcessorLight.process(AbstractProcessorLight.java:66)\n","stream":"stdout","time":"2019-01-18T07:49:27.375211594Z"}
352            {"log":"\u0009at org.apache.coyote.AbstractProtocol$ConnectionHandler.process(AbstractProtocol.java:806)\n","stream":"stdout","time":"2019-01-18T07:49:27.375225237Z"}
353            {"log":"\u0009at org.apache.tomcat.util.net.NioEndpoint$SocketProcessor.doRun(NioEndpoint.java:1498)\n","stream":"stdout","time":"2019-01-18T07:49:27.375239487Z"}
354            {"log":"\u0009at org.apache.tomcat.util.net.SocketProcessorBase.run(SocketProcessorBase.java:49)\n","stream":"stdout","time":"2019-01-18T07:49:27.375253464Z"}
355            {"log":"\u0009at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)\n","stream":"stdout","time":"2019-01-18T07:49:27.375323255Z"}
356            {"log":"\u0009at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)\n","stream":"stdout","time":"2019-01-18T07:49:27.375345642Z"}
357            {"log":"\u0009at org.apache.tomcat.util.threads.TaskThread$WrappingRunnable.run(TaskThread.java:61)\n","stream":"stdout","time":"2019-01-18T07:49:27.375363208Z"}
358            {"log":"\u0009at java.lang.Thread.run(Thread.java:748)\n","stream":"stdout","time":"2019-01-18T07:49:27.375377695Z"}
359            {"log":"\n","stream":"stdout","time":"2019-01-18T07:49:27.375391335Z"}
360            {"log":"\n","stream":"stdout","time":"2019-01-18T07:49:27.375416915Z"}
361            {"log":"2019-01-18 07:53:06.419 [               ]  INFO 1 --- [vent-bus.prod-1] c.t.listener.CommonListener              : warehousing Dailywarehousing.daily\n","stream":"stdout","time":"2019-01-18T07:53:06.420527437Z"}
362        "#};
363
364        let mut codec = CharacterDelimitedDecoder::new(b'\n');
365        let buf = &mut BytesMut::new();
366
367        buf.extend(events.to_string().as_bytes());
368
369        let mut i = 0;
370        while codec.decode(buf).unwrap().is_some() {
371            i += 1;
372        }
373
374        assert_eq!(i, 51);
375    }
376
377    #[test]
378    fn decode_truncate_oversized() {
379        const MAX_LENGTH: usize = 6;
380
381        let mut codec = CharacterDelimitedDecoder::new_with_max_length(b'\n', MAX_LENGTH)
382            .with_oversized_action(OversizedAction::Truncate);
383        let buf = &mut BytesMut::new();
384
385        buf.put_slice(b"1234567890\n123456\n");
386
387        // Oversized frame is truncated to 6 bytes
388        assert_eq!(codec.decode(buf).unwrap(), Some(Bytes::from("123456")));
389        // Next frame is correctly parsed
390        assert_eq!(codec.decode(buf).unwrap(), Some(Bytes::from("123456")));
391        assert_eq!(codec.decode(buf).unwrap(), None);
392    }
393
394    #[test]
395    fn decode_truncate_exact_boundary() {
396        const MAX_LENGTH: usize = 6;
397
398        let mut codec = CharacterDelimitedDecoder::new_with_max_length(b'\n', MAX_LENGTH)
399            .with_oversized_action(OversizedAction::Truncate);
400        let buf = &mut BytesMut::new();
401
402        // Exactly at max_length — should be emitted normally, not truncated
403        buf.put_slice(b"123456\n");
404        assert_eq!(codec.decode(buf).unwrap(), Some(Bytes::from("123456")));
405        assert_eq!(codec.decode(buf).unwrap(), None);
406    }
407
408    #[test]
409    fn decode_truncate_next_frame_intact() {
410        const MAX_LENGTH: usize = 3;
411
412        let mut codec = CharacterDelimitedDecoder::new_with_max_length(b'\n', MAX_LENGTH)
413            .with_oversized_action(OversizedAction::Truncate);
414        let buf = &mut BytesMut::new();
415
416        buf.put_slice(b"toolong\nok\nanother_too_long\nfin\n");
417
418        assert_eq!(codec.decode(buf).unwrap(), Some(Bytes::from("too")));
419        assert_eq!(codec.decode(buf).unwrap(), Some(Bytes::from("ok")));
420        assert_eq!(codec.decode(buf).unwrap(), Some(Bytes::from("ano")));
421        assert_eq!(codec.decode(buf).unwrap(), Some(Bytes::from("fin")));
422        assert_eq!(codec.decode(buf).unwrap(), None);
423    }
424
425    #[test]
426    fn decode_eof_truncate() {
427        const MAX_LENGTH: usize = 4;
428
429        let mut codec = CharacterDelimitedDecoder::new_with_max_length(b'\n', MAX_LENGTH)
430            .with_oversized_action(OversizedAction::Truncate);
431        let buf = &mut BytesMut::new();
432
433        // No trailing delimiter — decode_eof should truncate
434        buf.put_slice(b"abcdefgh");
435        assert_eq!(codec.decode_eof(buf).unwrap(), Some(Bytes::from("abcd")));
436        assert_eq!(codec.decode_eof(buf).unwrap(), None);
437    }
438
439    #[test]
440    fn decode_truncate_mixed_with_drop() {
441        const MAX_LENGTH: usize = 6;
442
443        // Default (Drop) behavior — oversized frames are dropped entirely
444        let mut codec_drop = CharacterDelimitedDecoder::new_with_max_length(b'\n', MAX_LENGTH);
445        let buf = &mut BytesMut::new();
446        buf.put_slice(b"1234567\n123456\n123412314\n123");
447
448        assert_eq!(codec_drop.decode(buf).unwrap(), Some(Bytes::from("123456")));
449        assert_eq!(codec_drop.decode(buf).unwrap(), None);
450
451        // Truncate behavior — oversized frames are truncated
452        let mut codec_trunc = CharacterDelimitedDecoder::new_with_max_length(b'\n', MAX_LENGTH)
453            .with_oversized_action(OversizedAction::Truncate);
454        let buf = &mut BytesMut::new();
455        buf.put_slice(b"1234567\n123456\n123412314\n123");
456
457        assert_eq!(
458            codec_trunc.decode(buf).unwrap(),
459            Some(Bytes::from("123456"))
460        );
461        assert_eq!(
462            codec_trunc.decode(buf).unwrap(),
463            Some(Bytes::from("123456"))
464        );
465        assert_eq!(
466            codec_trunc.decode(buf).unwrap(),
467            Some(Bytes::from("123412"))
468        );
469        assert_eq!(codec_trunc.decode(buf).unwrap(), None);
470    }
471}
codecs/decoding/framing/character_delimited.rs

codecs/decoding/framing/
character_delimited.rs
