Skip to content

Commit

Permalink
fix: properly flush and sync all writers
Browse files Browse the repository at this point in the history
  • Loading branch information
holtgrewe committed Oct 24, 2023
1 parent f4f01b1 commit 9c1658d
Show file tree
Hide file tree
Showing 3 changed files with 58 additions and 1 deletion.
21 changes: 21 additions & 0 deletions src/annotate/seqvars/mod.rs
Original file line number Diff line number Diff line change
Expand Up @@ -50,6 +50,7 @@ use crate::annotate::seqvars::provider::MehariProvider;
use crate::common::GenomeRelease;

use crate::db::create::txs::data::TxSeqDatabase;
use crate::finalize_buf_writer;
use crate::ped::{PedigreeByName, Sex};

use self::ann::{AnnField, Consequence, FeatureBiotype};
Expand Down Expand Up @@ -653,6 +654,12 @@ impl VarFishSeqvarTsvWriter {
}
}

/// Flush buffers.
pub fn flush(&mut self) -> Result<(), anyhow::Error> {
self.inner.flush()?;
Ok(())
}

/// Fill `record` coordinate fields.
///
/// # Returns
Expand Down Expand Up @@ -1560,11 +1567,21 @@ pub fn run(_common: &crate::common::Args, args: &Args) -> Result<(), anyhow::Err
.map(BufWriter::new)
.map(BgzfWriter::new)?,
);

run_with_writer(&mut writer, args)?;

let bgzf_writer = writer.into_inner();
let mut buf_writer = bgzf_writer

Check warning on line 1574 in src/annotate/seqvars/mod.rs

View check run for this annotation

Codecov / codecov/patch

src/annotate/seqvars/mod.rs#L1573-L1574

Added lines #L1573 - L1574 were not covered by tests
.finish()
.map_err(|e| anyhow::anyhow!("problem finishing BGZF: {}", e))?;
finalize_buf_writer!(buf_writer);

Check warning on line 1577 in src/annotate/seqvars/mod.rs

View check run for this annotation

Codecov / codecov/patch

src/annotate/seqvars/mod.rs#L1576-L1577

Added lines #L1576 - L1577 were not covered by tests
} else {
let mut writer = VcfWriter::new(File::create(path_output_vcf).map(BufWriter::new)?);

run_with_writer(&mut writer, args)?;

let mut buf_writer = writer.into_inner();
finalize_buf_writer!(buf_writer);
}
} else {
// Load the HGNC xlink map.
Expand Down Expand Up @@ -1601,6 +1618,10 @@ pub fn run(_common: &crate::common::Args, args: &Args) -> Result<(), anyhow::Err

writer.set_hgnc_map(hgnc_map);
run_with_writer(&mut writer, args)?;

writer
.flush()
.map_err(|e| anyhow::anyhow!("problem flushing file: {}", e))?;

Check warning on line 1624 in src/annotate/seqvars/mod.rs

View check run for this annotation

Codecov / codecov/patch

src/annotate/seqvars/mod.rs#L1624

Added line #L1624 was not covered by tests
}

Ok(())
Expand Down
24 changes: 23 additions & 1 deletion src/annotate/strucvars/mod.rs
Original file line number Diff line number Diff line change
Expand Up @@ -12,6 +12,7 @@ use std::{fs::File, io::BufWriter};

use crate::common::noodles::{open_vcf_reader, AsyncVcfReader};
use crate::common::GenomeRelease;
use crate::finalize_buf_writer;
use crate::ped::PedigreeByName;
use annonars::common::cli::CANONICAL;
use annonars::freqs::cli::import::reading::guess_assembly;
Expand Down Expand Up @@ -923,7 +924,7 @@ impl AnnotatedVcfWriter for VarFishStrucvarTsvWriter {
}

impl VarFishStrucvarTsvWriter {
// Create new TSV writer from path.
/// Create new TSV writer from path.
pub fn with_path<P>(p: P) -> Self
where
P: AsRef<Path>,
Expand All @@ -942,6 +943,12 @@ impl VarFishStrucvarTsvWriter {
header: None,
}
}

/// Flush buffers.
pub fn flush(&mut self) -> Result<(), anyhow::Error> {
self.inner.flush()?;
Ok(())
}
}

/// Enumeration for describing the orientation of a paired-end read.
Expand Down Expand Up @@ -2990,13 +2997,24 @@ pub async fn run(_common: &crate::common::Args, args: &Args) -> Result<(), anyho
);
writer.set_assembly(assembly);
writer.set_pedigree(&pedigree);

run_with_writer(&mut writer, &args, &pedigree, &header).await?;

let bgzf_writer = writer.into_inner();
let mut buf_writer = bgzf_writer

Check warning on line 3004 in src/annotate/strucvars/mod.rs

View check run for this annotation

Codecov / codecov/patch

src/annotate/strucvars/mod.rs#L3003-L3004

Added lines #L3003 - L3004 were not covered by tests
.finish()
.map_err(|e| anyhow::anyhow!("problem finishing BGZF: {}", e))?;
finalize_buf_writer!(buf_writer);

Check warning on line 3007 in src/annotate/strucvars/mod.rs

View check run for this annotation

Codecov / codecov/patch

src/annotate/strucvars/mod.rs#L3006-L3007

Added lines #L3006 - L3007 were not covered by tests
} else {
let mut writer =
noodles_vcf::Writer::new(File::create(path_output_vcf).map(BufWriter::new)?);
writer.set_assembly(assembly);
writer.set_pedigree(&pedigree);

run_with_writer(&mut writer, &args, &pedigree, &header).await?;

let mut buf_writer = writer.into_inner();
finalize_buf_writer!(buf_writer);
}
} else {
let path_output_tsv = args
Expand All @@ -3009,6 +3027,10 @@ pub async fn run(_common: &crate::common::Args, args: &Args) -> Result<(), anyho
writer.set_pedigree(&pedigree);

run_with_writer(&mut writer, &args, &pedigree, &header).await?;

writer
.flush()
.map_err(|e| anyhow::anyhow!("problem flushing file: {}", e))?;

Check warning on line 3033 in src/annotate/strucvars/mod.rs

View check run for this annotation

Codecov / codecov/patch

src/annotate/strucvars/mod.rs#L3033

Added line #L3033 was not covered by tests
}

Ok(())
Expand Down
14 changes: 14 additions & 0 deletions src/common/io/std.rs
Original file line number Diff line number Diff line change
Expand Up @@ -94,6 +94,20 @@ where
Ok(buffer)
}

/// Given a `BufWriter<File>`, flush buffers and sync the file.
#[macro_export]
macro_rules! finalize_buf_writer {
($a:expr) => {
$a.flush()
.map_err(|e| anyhow::anyhow!("problem flushing buffers: {}", e))?;

Check warning on line 102 in src/common/io/std.rs

View check run for this annotation

Codecov / codecov/patch

src/common/io/std.rs#L102

Added line #L102 was not covered by tests
let file = $a
.into_inner()
.map_err(|e| anyhow::anyhow!("problem getting inner file: {}", e))?;

Check warning on line 105 in src/common/io/std.rs

View check run for this annotation

Codecov / codecov/patch

src/common/io/std.rs#L105

Added line #L105 was not covered by tests
file.sync_all()
.map_err(|e| anyhow::anyhow!("problem syncing file: {}", e))?;

Check warning on line 107 in src/common/io/std.rs

View check run for this annotation

Codecov / codecov/patch

src/common/io/std.rs#L107

Added line #L107 was not covered by tests
};
}

#[cfg(test)]
mod test {
use std::io::Read;
Expand Down

0 comments on commit 9c1658d

Please sign in to comment.