pub fn read_file_task<Ctor, I, Err>(
    path: PathBuf,
    tx: std::sync::mpsc::SyncSender<Result<MessagePayload, anyhow::Error>>,
    activator: Option<SyncActivator>,
    read_style: FileReadStyle,
    compression: Compression,
    iter_ctor: Ctor,
) where
    I: IntoIterator<Item = Result<MessagePayload, Err>> + Send + 'static,
    Ctor: FnOnce(Box<dyn AvroRead + Send>) -> Result<I, Err>,
    Err: Into<anyhow::Error>,
{
    log::trace!("reading file {}", path.display());
    let file = match std::fs::File::open(&path).with_context(|| {
        format!(
            "file source: unable to open file at path {}",
            path.to_string_lossy(),
        )
    }) {
        Ok(file) => file,
        Err(err) => {
            // If we fail to send an error, it's likely due to a race condition
            // with the source being closed.
            let _ = tx.send(Err(err));
            return;
        }
    };

    let file_stream = match open_file_stream(path.clone(), file, read_style) {
        Ok(f) => f,
        Err(err) => {
            // If we fail to send an error, it's likely due to a race condition
            // with the source being closed.
            let _ = tx.send(Err(err));
            return;
        }
    };

    let file_stream: Box<dyn AvroRead + Send> = match compression {
        Compression::Gzip => Box::new(MultiGzDecoder::new(file_stream)),
        Compression::None => Box::new(file_stream),
    };

    let iter = iter_ctor(file_stream);

    match iter.map_err(Into::into).with_context(|| {
        format!(
            "Failed to obtain records from file at path {}",
            path.to_string_lossy(),
        )
    }) {
        Ok(i) => send_records(i, tx, activator),
        Err(e) => {
            // If we fail to send an error, it's likely due to a race condition
            // with the source being closed.
            let _ = tx.send(Err(e));
        }
    };
}
/// Blocking logic to read from a file, intended for its own thread.
Blocking logic to read from a file, intended for its own thread.
Blocking logic to read from a file intended for its own thread .
pub fn read_file_task<Ctor, I, Err>(
    path: PathBuf,
    tx: std::sync::mpsc::SyncSender<Result<MessagePayload, anyhow::Error>>,
    activator: Option<SyncActivator>,
    read_style: FileReadStyle,
    compression: Compression,
    iter_ctor: Ctor,
) where
    I: IntoIterator<Item = Result<MessagePayload, Err>> + Send + 'static,
    Ctor: FnOnce(Box<dyn AvroRead + Send>) -> Result<I, Err>,
    Err: Into<anyhow::Error>,
{
    log::trace!("reading file {}", path.display());
    let file = match std::fs::File::open(&path).with_context(|| {
        format!(
            "file source: unable to open file at path {}",
            path.to_string_lossy(),
        )
    }) {
        Ok(file) => file,
        Err(err) => {
            // If we fail to send an error, it's likely due to a race condition
            // with the source being closed.
            let _ = tx.send(Err(err));
            return;
        }
    };
    let file_stream = match open_file_stream(path.clone(), file, read_style) {
        Ok(f) => f,
        Err(err) => {
            // If we fail to send an error, it's likely due to a race condition
            // with the source being closed.
            let _ = tx.send(Err(err));
            return;
        }
    };
    let file_stream: Box<dyn AvroRead + Send> = match compression {
        Compression::Gzip => Box::new(MultiGzDecoder::new(file_stream)),
        Compression::None => Box::new(file_stream),
    };
    let iter = iter_ctor(file_stream);
    match iter.map_err(Into::into).with_context(|| {
        format!(
            "Failed to obtain records from file at path {}",
            path.to_string_lossy(),
        )
    }) {
        Ok(i) => send_records(i, tx, activator),
        Err(e) => {
            // If we fail to send an error, it's likely due to a race condition
            // with the source being closed.
            let _ = tx.send(Err(e));
        }
    };
}
pub fn read_file_task < Ctor , I , Err > ( path : PathBuf , tx : std :: sync :: mpsc :: SyncSender < Result < MessagePayload , anyhow :: Error > > , activator : Option < SyncActivator > , read_style : FileReadStyle , compression : Compression , iter_ctor : Ctor , ) where I : IntoIterator < Item = Result < MessagePayload , Err > > + Send + ' static , Ctor : FnOnce ( Box < dyn AvroRead + Send > ) -> Result < I , Err > , Err : Into < anyhow :: Error > , { log :: trace ! ( " " path display ( ) ) ; let file = match std :: fs :: File :: open ( & path ) . with_context ( | | { format ! ( " " path to_string_lossy ( ) ) } ) { Ok ( file ) => file , Err ( err ) => { // If we fail to send an error, it's likely due to a race condition // with the source being closed. let _ = tx . send ( Err ( err ) ) ; return ; } } ; let file_stream = match open_file_stream ( path . clone ( ) , file , read_style ) { Ok ( f ) => f , Err ( err ) => { // If we fail to send an error, it's likely due to a race condition // with the source being closed. let _ = tx . send ( Err ( err ) ) ; return ; } } ; let file_stream : Box < dyn AvroRead + Send > = match compression { Compression :: Gzip => Box :: new ( MultiGzDecoder :: new ( file_stream ) ) , Compression :: None => Box :: new ( file_stream ) , } ; let iter = iter_ctor ( file_stream ) ; match iter . map_err ( Into :: into ) . with_context ( | | { format ! ( " " path to_string_lossy ( ) ) } ) { Ok ( i ) => send_records ( i , tx , activator ) , Err ( e ) => { // If we fail to send an error, it's likely due to a race condition // with the source being closed. let _ = tx . send ( Err ( e ) ) ; } } ; }
Blocking logic to read from a file, intended for its own thread.
Blocking logic to read from a file intended for its own thread .
path
PathBuf
tx
std::sync::mpsc::SyncSender<Result<MessagePayload, anyhow::Error>>
activator
Option<SyncActivator>
read_style
FileReadStyle
compression
Compression
iter_ctor
Ctor
path
PathBuf
tx
std::sync::mpsc::SyncSender<Result<MessagePayload, anyhow::Error>>
activator
Option<SyncActivator>
read_style
FileReadStyle
compression
Compression
iter_ctor
Ctor