Struct DecoderFramedRead 

pub struct DecoderFramedRead<T, D> { /* private fields */ }

A tokio_util::codec::FramedRead wrapper that continues decoding after recoverable decoder errors.

Problem

The standard tokio_util::codec::FramedRead terminates the stream when a decoder returns an error. This is problematic for Vector because:

• Vector decoders classify some errors as recoverable (e.g., malformed JSON in one line shouldn’t stop processing subsequent valid lines)
• Sources need to continue processing data even after encountering decode errors
• Metrics and observability require tracking both successful and failed decode attempts

Solution

DecoderFramedRead wraps the decoder in a DecoderResultWrapper that transforms decoder errors into successful results containing the error. This allows:

• The stream to continue after errors
• Callers to inspect errors and decide whether to continue (via StreamDecodingError::can_continue())
• Proper error metrics and logging

When to Use

Use DecoderFramedRead when:

• You’re using a Vector Decoder that implements error recovery logic
• You need to continue processing after decode errors
• You’re processing line-delimited or record-based formats where one bad record shouldn’t stop processing

Use standard FramedRead when:

• You’re using simple decoders (e.g., CharacterDelimitedDecoder) that don’t need error recovery
• Any decode error should terminate the stream
• You’re working with binary protocols where errors indicate corruption

Example

use vector_lib::codecs::{DecoderFramedRead, Decoder};
use futures::StreamExt;

let decoder = Decoder::new(
    Framer::NewlineDelimited(NewlineDelimitedDecoder::new()),
    Deserializer::Json(JsonDeserializer::default()),
);

let mut stream = DecoderFramedRead::new(reader, decoder);

while let Some(result) = stream.next().await {
    match result {
        Ok(events) => process_events(events),
        Err(error) if error.can_continue() => {
            // Log the error but continue processing
            warn!("Decode error (continuing): {}", error);
        }
        Err(error) => {
            // Fatal error, stop processing
            error!("Fatal decode error: {}", error);
            break;
        }
    }
}

Implementations

impl<T, D> DecoderFramedRead<T, D>

where T: AsyncRead, D: Decoder,

pub fn new(inner: T, decoder: D) -> Self

Creates a new DecoderFramedRead with the given decoder.

This wraps the provided decoder to enable error recovery, allowing the stream to continue processing after recoverable decode errors.

Arguments

• inner - The async reader to read from
• decoder - The decoder to use for parsing data

pub fn with_capacity(inner: T, decoder: D, capacity: usize) -> Self

Creates a new DecoderFramedRead with a specific buffer capacity.

Use this when you know the expected message size to optimize memory usage.

Arguments

• inner - The async reader to read from
• decoder - The decoder to use for parsing data
• capacity - The initial buffer capacity in bytes

pub fn get_ref(&self) -> &T

Returns a reference to the underlying I/O stream.

This is useful for accessing the underlying reader’s properties or state without consuming the DecoderFramedRead.

pub fn get_mut(&mut self) -> &mut T

Returns a mutable reference to the underlying I/O stream.

This allows modifying the underlying reader’s state, though care should be taken not to interfere with ongoing decoding operations.

pub fn read_buffer(&self) -> &BytesMut

Returns a reference to the internal read buffer.

This provides access to any buffered but not yet decoded data. Useful for debugging or implementing custom recovery logic.

Trait Implementations

impl<T, D> Stream for DecoderFramedRead<T, D>

where T: AsyncRead, D: Decoder, D::Error: From<Error>,

type Item = Result<<D as Decoder>::Item, <D as Decoder>::Error>

Values yielded by the stream.

fn poll_next( self: Pin<&mut Self>, cx: &mut Context<'_>, ) -> Poll<Option<Self::Item>>

Attempt to pull out the next value of this stream, registering the current task for wakeup if the value is not yet available, and returning None if the stream is exhausted.

fn size_hint(&self) -> (usize, Option<usize>)

Returns the bounds on the remaining length of the stream.

impl<'pin, T, D> Unpin for DecoderFramedRead<T, D>

where PinnedFieldsOf<__DecoderFramedRead<'pin, T, D>>: Unpin,