use std::fmt;
use std::fs::File;
use std::io::{self, BufReader, Read};
use std::path::Path;
use crate::error::{Error, ParseError, Result};
use crate::format::{Compression, Format};
use crate::qual::{self, QualityEncoding};
use crate::record::Sequence;
pub const DEFAULT_BUFFER_SIZE: usize = 128 * 1024;
const MIN_BUFFER_SIZE: usize = 4 * 1024;
pub struct FastxReader<R: Read> {
inner: R,
buf: Vec<u8>,
pos: usize,
end: usize,
eof: bool,
format: Option<Format>,
line: u64,
quality_encoding: QualityEncoding,
max_line_length: Option<usize>,
max_record_length: Option<usize>,
}
impl<R: Read> FastxReader<R> {
pub fn new(inner: R) -> FastxReader<R> {
FastxReader::with_capacity(inner, DEFAULT_BUFFER_SIZE)
}
pub fn with_format(inner: R, format: Format) -> FastxReader<R> {
let mut reader = FastxReader::new(inner);
reader.format = Some(format);
reader
}
pub fn with_capacity(inner: R, capacity: usize) -> FastxReader<R> {
FastxReader {
inner,
buf: vec![0; capacity.max(MIN_BUFFER_SIZE)],
pos: 0,
end: 0,
eof: false,
format: None,
line: 0,
quality_encoding: QualityEncoding::Phred33,
max_line_length: None,
max_record_length: None,
}
}
pub fn format(&self) -> Option<Format> {
self.format
}
pub fn quality_encoding(&self) -> QualityEncoding {
self.quality_encoding
}
pub fn line_number(&self) -> u64 {
self.line
}
pub fn into_inner(self) -> R {
self.inner
}
pub fn read_into(&mut self, record: &mut Sequence) -> Result<bool> {
let format = match self.format {
Some(format) => {
if !self.skip_blank_lines()? {
return Ok(false);
}
format
}
None => match self.detect_format()? {
Some(format) => {
self.format = Some(format);
format
}
None => return Ok(false),
},
};
record.clear();
match format {
Format::Fasta => self.read_fasta_into(record)?,
Format::Fastq => self.read_fastq_into(record)?,
}
Ok(true)
}
pub fn read_record(&mut self) -> Result<Option<Sequence>> {
let mut record = Sequence::default();
if self.read_into(&mut record)? {
Ok(Some(record))
} else {
Ok(None)
}
}
pub fn records(&mut self) -> Records<'_, R> {
Records { reader: self }
}
pub fn for_each_record<F>(&mut self, mut f: F) -> Result<()>
where
F: FnMut(&Sequence) -> Result<()>,
{
let mut record = Sequence::default();
while self.read_into(&mut record)? {
f(&record)?;
}
Ok(())
}
pub fn count_records(&mut self) -> Result<u64> {
let mut n = 0;
let mut record = Sequence::default();
while self.read_into(&mut record)? {
n += 1;
}
Ok(n)
}
fn read_fasta_into(&mut self, record: &mut Sequence) -> Result<()> {
let (start, end) = match self.read_line()? {
Some(range) => range,
None => {
return Err(Error::parse(
self.line,
ParseError::UnexpectedEof {
expected: "a FASTA header",
},
))
}
};
if self.buf[start] != b'>' && self.buf[start] != b';' {
return Err(Error::parse(
self.line,
ParseError::ExpectedHeader {
found: self.buf[start],
},
));
}
record.set_header(&self.buf[start + 1..end]);
if record.id.is_empty() {
return Err(Error::parse(self.line, ParseError::EmptyId));
}
loop {
match self.peek_byte()? {
None | Some(b'>') => break,
_ => {
let (start, end) = self.read_line()?.expect("peeked byte is available");
record.seq.extend_from_slice(&self.buf[start..end]);
self.check_record_limit(record.seq.len(), "sequence")?;
}
}
}
Ok(())
}
fn read_fastq_into(&mut self, record: &mut Sequence) -> Result<()> {
let (start, end) = match self.read_line()? {
Some(range) => range,
None => {
return Err(Error::parse(
self.line,
ParseError::UnexpectedEof {
expected: "a FASTQ header",
},
))
}
};
if self.buf[start] != b'@' {
return Err(Error::parse(
self.line,
ParseError::ExpectedHeader {
found: self.buf[start],
},
));
}
record.set_header(&self.buf[start + 1..end]);
if record.id.is_empty() {
return Err(Error::parse(self.line, ParseError::EmptyId));
}
loop {
match self.peek_byte()? {
None => {
return Err(Error::parse(
self.line,
ParseError::UnexpectedEof {
expected: "a FASTQ '+' separator",
},
))
}
Some(b'+') => {
self.read_line()?;
break;
}
_ => {
let (start, end) = self.read_line()?.expect("peeked byte is available");
record.seq.extend_from_slice(&self.buf[start..end]);
self.check_record_limit(record.seq.len(), "sequence")?;
}
}
}
let quality = record.quality.get_or_insert_with(Vec::new);
while quality.len() < record.seq.len() {
match self.read_line()? {
Some((start, end)) => quality.extend_from_slice(&self.buf[start..end]),
None => {
return Err(Error::LengthMismatch {
id: record.id.clone(),
seq: record.seq.len(),
quality: quality.len(),
})
}
}
}
if quality.len() != record.seq.len() {
return Err(Error::LengthMismatch {
id: record.id.clone(),
seq: record.seq.len(),
quality: quality.len(),
});
}
if self.quality_encoding != QualityEncoding::Phred33 {
let from = self.quality_encoding.offset();
for c in quality.iter_mut() {
*c = qual::encode(qual::score(*c, from), qual::PHRED33);
}
}
Ok(())
}
fn check_line_limit(&self, length: usize) -> Result<()> {
match self.max_line_length {
Some(limit) if length > limit => Err(Error::TooLarge {
line: self.line + 1,
what: "line",
limit,
}),
_ => Ok(()),
}
}
fn check_record_limit(&self, length: usize, what: &'static str) -> Result<()> {
match self.max_record_length {
Some(limit) if length > limit => Err(Error::TooLarge {
line: self.line,
what,
limit,
}),
_ => Ok(()),
}
}
fn skip_blank_lines(&mut self) -> Result<bool> {
loop {
match self.peek_byte()? {
None => return Ok(false),
Some(b'\n') => {
self.pos += 1;
self.line += 1;
}
Some(b'\r') => self.pos += 1,
Some(_) => return Ok(true),
}
}
}
fn detect_format(&mut self) -> Result<Option<Format>> {
if !self.skip_blank_lines()? {
return Ok(None);
}
let byte = self.buf[self.pos];
match Format::from_first_byte(byte) {
Some(format) => Ok(Some(format)),
None => Err(Error::parse(
self.line + 1,
ParseError::ExpectedHeader { found: byte },
)),
}
}
fn read_line(&mut self) -> Result<Option<(usize, usize)>> {
let mut search_from = self.pos;
loop {
if let Some(offset) = memchr::memchr(b'\n', &self.buf[search_from..self.end]) {
let newline = search_from + offset;
let start = self.pos;
let mut stop = newline;
if stop > start && self.buf[stop - 1] == b'\r' {
stop -= 1;
}
self.check_line_limit(stop - start)?;
self.pos = newline + 1;
self.line += 1;
return Ok(Some((start, stop)));
}
if self.eof {
if self.pos == self.end {
return Ok(None);
}
let start = self.pos;
let mut stop = self.end;
if stop > start && self.buf[stop - 1] == b'\r' {
stop -= 1;
}
self.check_line_limit(stop - start)?;
self.pos = self.end;
self.line += 1;
return Ok(Some((start, stop)));
}
self.check_line_limit(self.end - self.pos)?;
let previous_end = self.end;
let shift = self.refill()?;
search_from = previous_end - shift;
}
}
fn peek_byte(&mut self) -> Result<Option<u8>> {
while self.pos == self.end && !self.eof {
self.refill()?;
}
if self.pos == self.end {
Ok(None)
} else {
Ok(Some(self.buf[self.pos]))
}
}
fn refill(&mut self) -> Result<usize> {
let mut shift = 0;
if self.pos > 0 {
self.buf.copy_within(self.pos..self.end, 0);
shift = self.pos;
self.end -= self.pos;
self.pos = 0;
}
if self.end == self.buf.len() {
let grown = self.buf.len().saturating_mul(2).max(MIN_BUFFER_SIZE);
self.buf.resize(grown, 0);
}
loop {
match self.inner.read(&mut self.buf[self.end..]) {
Ok(0) => {
self.eof = true;
break;
}
Ok(n) => {
self.end += n;
break;
}
Err(e) if e.kind() == io::ErrorKind::Interrupted => continue,
Err(e) => return Err(Error::Io(e)),
}
}
Ok(shift)
}
}
impl<R: Read> fmt::Debug for FastxReader<R> {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
f.debug_struct("FastxReader")
.field("format", &self.format)
.field("buffer_size", &self.buf.len())
.field("buffered", &(self.end - self.pos))
.field("line", &self.line)
.field("eof", &self.eof)
.finish_non_exhaustive()
}
}
impl<R: Read> Iterator for FastxReader<R> {
type Item = Result<Sequence>;
fn next(&mut self) -> Option<Self::Item> {
match self.read_record() {
Ok(Some(record)) => Some(Ok(record)),
Ok(None) => None,
Err(e) => Some(Err(e)),
}
}
}
pub struct Records<'a, R: Read> {
reader: &'a mut FastxReader<R>,
}
impl<R: Read> Iterator for Records<'_, R> {
type Item = Result<Sequence>;
fn next(&mut self) -> Option<Self::Item> {
match self.reader.read_record() {
Ok(Some(record)) => Some(Ok(record)),
Ok(None) => None,
Err(e) => Some(Err(e)),
}
}
}
#[derive(Debug, Clone)]
pub struct ReaderBuilder {
format: Option<Format>,
buffer_size: usize,
quality_encoding: QualityEncoding,
max_line_length: Option<usize>,
max_record_length: Option<usize>,
}
impl Default for ReaderBuilder {
fn default() -> Self {
ReaderBuilder {
format: None,
buffer_size: DEFAULT_BUFFER_SIZE,
quality_encoding: QualityEncoding::Phred33,
max_line_length: None,
max_record_length: None,
}
}
}
impl ReaderBuilder {
pub fn new() -> ReaderBuilder {
ReaderBuilder::default()
}
pub fn format(mut self, format: Format) -> Self {
self.format = Some(format);
self
}
pub fn buffer_size(mut self, bytes: usize) -> Self {
self.buffer_size = bytes;
self
}
pub fn quality_encoding(mut self, encoding: QualityEncoding) -> Self {
self.quality_encoding = encoding;
self
}
pub fn max_line_length(mut self, bytes: usize) -> Self {
self.max_line_length = Some(bytes);
self
}
pub fn max_record_length(mut self, bytes: usize) -> Self {
self.max_record_length = Some(bytes);
self
}
pub fn build<R: Read>(&self, inner: R) -> FastxReader<R> {
let mut reader = FastxReader::with_capacity(inner, self.buffer_size);
reader.format = self.format;
reader.quality_encoding = self.quality_encoding;
reader.max_line_length = self.max_line_length;
reader.max_record_length = self.max_record_length;
reader
}
pub fn open<P: AsRef<Path>>(&self, path: P) -> Result<FastxReader<Box<dyn Read + Send>>> {
let path = path.as_ref();
let mut builder = self.clone();
if builder.format.is_none() {
builder.format = Format::from_path(path);
}
Ok(builder.build(open_reader(path)?))
}
}
pub type BoxedReader = FastxReader<Box<dyn Read + Send>>;
pub fn open<P: AsRef<Path>>(path: P) -> Result<BoxedReader> {
ReaderBuilder::default().open(path)
}
pub fn from_stdin() -> Result<BoxedReader> {
let stream = decompress(Box::new(io::stdin()))?;
Ok(FastxReader::new(stream))
}
fn open_reader(path: &Path) -> Result<Box<dyn Read + Send>> {
let file = File::open(path)
.map_err(|e| Error::Io(io::Error::new(e.kind(), format!("{}: {e}", path.display()))))?;
decompress(Box::new(BufReader::with_capacity(64 * 1024, file)))
}
fn decompress(mut stream: Box<dyn Read + Send>) -> Result<Box<dyn Read + Send>> {
let mut magic = [0u8; 4];
let mut filled = 0;
while filled < magic.len() {
match stream.read(&mut magic[filled..]) {
Ok(0) => break,
Ok(n) => filled += n,
Err(e) if e.kind() == io::ErrorKind::Interrupted => continue,
Err(e) => return Err(Error::Io(e)),
}
}
let head = io::Cursor::new(magic[..filled].to_vec());
let rejoined = head.chain(stream);
match Compression::from_magic(&magic[..filled]) {
Compression::None => Ok(Box::new(rejoined)),
Compression::Gzip | Compression::Bgzf => gunzip(rejoined),
Compression::Zstd => unzstd(rejoined),
}
}
#[cfg(feature = "gzip")]
fn gunzip<R: Read + Send + 'static>(stream: R) -> Result<Box<dyn Read + Send>> {
Ok(Box::new(flate2::read::MultiGzDecoder::new(stream)))
}
#[cfg(not(feature = "gzip"))]
fn gunzip<R: Read + Send + 'static>(_stream: R) -> Result<Box<dyn Read + Send>> {
Err(Error::FeatureDisabled("gzip"))
}
#[cfg(feature = "zstd")]
fn unzstd<R: Read + Send + 'static>(stream: R) -> Result<Box<dyn Read + Send>> {
Ok(Box::new(zstd::stream::read::Decoder::new(stream)?))
}
#[cfg(not(feature = "zstd"))]
fn unzstd<R: Read + Send + 'static>(_stream: R) -> Result<Box<dyn Read + Send>> {
Err(Error::FeatureDisabled("zstd"))
}
#[cfg(test)]
mod tests {
use super::*;
fn ids(data: &[u8]) -> Vec<String> {
FastxReader::new(data).map(|r| r.unwrap().id).collect()
}
#[test]
fn reads_simple_fasta() {
let data = b">a desc here\nACGT\n>b\nTTTT\nGGGG\n";
let records: Vec<_> = FastxReader::new(&data[..])
.collect::<Result<Vec<_>>>()
.unwrap();
assert_eq!(records.len(), 2);
assert_eq!(records[0].id, "a");
assert_eq!(records[0].description.as_deref(), Some("desc here"));
assert_eq!(records[0].seq, b"ACGT");
assert_eq!(records[1].seq, b"TTTTGGGG");
assert!(records[1].quality.is_none());
}
#[test]
fn reads_simple_fastq() {
let data = b"@a\nACGT\n+\nIIII\n@b desc\nTT\n+b desc\n!!\n";
let records: Vec<_> = FastxReader::new(&data[..])
.collect::<Result<Vec<_>>>()
.unwrap();
assert_eq!(records.len(), 2);
assert_eq!(records[0].quality.as_deref(), Some(&b"IIII"[..]));
assert_eq!(records[1].id, "b");
assert_eq!(records[1].description.as_deref(), Some("desc"));
assert_eq!(records[1].quality.as_deref(), Some(&b"!!"[..]));
}
#[test]
fn detects_format() {
let mut reader = FastxReader::new(&b">a\nAC\n"[..]);
assert_eq!(reader.format(), None);
reader.next().unwrap().unwrap();
assert_eq!(reader.format(), Some(Format::Fasta));
let mut reader = FastxReader::new(&b"@a\nAC\n+\nII\n"[..]);
reader.next().unwrap().unwrap();
assert_eq!(reader.format(), Some(Format::Fastq));
}
#[test]
fn handles_crlf_and_missing_final_newline() {
let data = b">a\r\nACGT\r\nAC\r\n>b\r\nTT";
let records: Vec<_> = FastxReader::new(&data[..])
.collect::<Result<Vec<_>>>()
.unwrap();
assert_eq!(records[0].seq, b"ACGTAC");
assert_eq!(records[1].seq, b"TT");
}
#[test]
fn handles_blank_lines_between_records() {
let data = b"\n\n>a\nACGT\n\n\n>b\nTT\n\n";
assert_eq!(ids(&data[..]), ["a", "b"]);
let records: Vec<_> = FastxReader::new(&data[..])
.collect::<Result<Vec<_>>>()
.unwrap();
assert_eq!(records[0].seq, b"ACGT");
}
#[test]
fn handles_empty_input() {
assert_eq!(FastxReader::new(&b""[..]).count(), 0);
assert_eq!(FastxReader::new(&b"\n\n\n"[..]).count(), 0);
}
#[test]
fn multi_line_fastq() {
let data = b"@a\nACGT\nACGT\n+\nIIII\nJJJJ\n@b\nTT\n+\n!!\n";
let records: Vec<_> = FastxReader::new(&data[..])
.collect::<Result<Vec<_>>>()
.unwrap();
assert_eq!(records[0].seq, b"ACGTACGT");
assert_eq!(records[0].quality.as_deref(), Some(&b"IIIIJJJJ"[..]));
assert_eq!(records[1].id, "b");
}
#[test]
fn quality_starting_with_at_sign() {
let data = b"@a\nACGT\n+\n@@@@\n@b\nTTTT\n+\nIIII\n";
let records: Vec<_> = FastxReader::new(&data[..])
.collect::<Result<Vec<_>>>()
.unwrap();
assert_eq!(records.len(), 2);
assert_eq!(records[0].quality.as_deref(), Some(&b"@@@@"[..]));
assert_eq!(records[1].id, "b");
}
#[test]
fn tiny_buffer_still_parses() {
let long = "A".repeat(50_000);
let data = format!(">a\n{long}\n>b\nACGT\n");
let mut reader = FastxReader::with_capacity(data.as_bytes(), 1);
let records: Vec<_> = reader.records().collect::<Result<Vec<_>>>().unwrap();
assert_eq!(records.len(), 2);
assert_eq!(records[0].seq.len(), 50_000);
assert_eq!(records[1].seq, b"ACGT");
}
#[test]
fn read_into_reuses_allocations() {
let data = b">a\nACGT\n>b\nTT\n";
let mut reader = FastxReader::new(&data[..]);
let mut record = Sequence::default();
assert!(reader.read_into(&mut record).unwrap());
assert_eq!(record.id, "a");
assert!(reader.read_into(&mut record).unwrap());
assert_eq!(record.id, "b");
assert_eq!(record.seq, b"TT");
assert!(!reader.read_into(&mut record).unwrap());
}
#[test]
fn empty_fasta_record_is_allowed() {
let data = b">a\n>b\nACGT\n";
let records: Vec<_> = FastxReader::new(&data[..])
.collect::<Result<Vec<_>>>()
.unwrap();
assert_eq!(records[0].seq, b"");
assert_eq!(records[1].seq, b"ACGT");
}
#[test]
fn rejects_garbage() {
let err = FastxReader::new(&b"not a sequence file\n"[..])
.next()
.unwrap()
.unwrap_err();
assert!(matches!(
err,
Error::Parse {
kind: ParseError::ExpectedHeader { found: b'n' },
..
}
));
}
#[test]
fn rejects_truncated_fastq() {
let err = FastxReader::new(&b"@a\nACGT\n"[..])
.next()
.unwrap()
.unwrap_err();
assert!(matches!(
err,
Error::Parse {
kind: ParseError::UnexpectedEof { .. },
..
}
));
let err = FastxReader::new(&b"@a\nACGT\n+\nII\n"[..])
.next()
.unwrap()
.unwrap_err();
assert!(matches!(
err,
Error::LengthMismatch {
seq: 4,
quality: 2,
..
}
));
}
#[test]
fn rejects_empty_id() {
let err = FastxReader::new(&b">\nACGT\n"[..])
.next()
.unwrap()
.unwrap_err();
assert!(matches!(
err,
Error::Parse {
kind: ParseError::EmptyId,
..
}
));
}
#[test]
fn reports_line_numbers() {
let data = b">a\nACGT\n>b\nACGT\nnope";
let mut reader = FastxReader::with_format(&data[..], Format::Fasta);
reader.next().unwrap().unwrap();
assert_eq!(reader.line_number(), 2);
}
#[test]
fn phred64_input_is_normalised_to_phred33() {
let data = b"@old\nACGT\n+\nhhhB\n";
let record = ReaderBuilder::new()
.quality_encoding(QualityEncoding::Phred64)
.build(&data[..])
.read_record()
.unwrap()
.unwrap();
assert_eq!(record.quality.as_deref(), Some(&b"III#"[..]));
assert_eq!(record.quality_scores().unwrap(), vec![40, 40, 40, 2]);
let record = FastxReader::new(&data[..]).read_record().unwrap().unwrap();
assert_eq!(record.quality.as_deref(), Some(&b"hhhB"[..]));
}
#[test]
fn line_length_limit_is_enforced() {
let long = format!(">a\n{}\n", "A".repeat(10_000));
let err = ReaderBuilder::new()
.max_line_length(1_000)
.build(long.as_bytes())
.read_record()
.unwrap_err();
assert!(
matches!(
err,
Error::TooLarge {
what: "line",
limit: 1_000,
..
}
),
"{err}"
);
let record = ReaderBuilder::new()
.max_line_length(1_000_000)
.build(long.as_bytes())
.read_record()
.unwrap()
.unwrap();
assert_eq!(record.seq.len(), 10_000);
}
#[test]
fn record_length_limit_catches_many_short_lines() {
let mut data = String::from(">a\n");
for _ in 0..200 {
data.push_str(&"A".repeat(50));
data.push('\n');
}
let err = ReaderBuilder::new()
.max_line_length(1_000)
.max_record_length(5_000)
.build(data.as_bytes())
.read_record()
.unwrap_err();
assert!(
matches!(
err,
Error::TooLarge {
what: "sequence",
limit: 5_000,
..
}
),
"{err}"
);
}
#[test]
fn limits_are_unlimited_by_default() {
let long = format!(">chrom\n{}\n", "ACGT".repeat(50_000));
let record = FastxReader::with_capacity(long.as_bytes(), 4096)
.read_record()
.unwrap()
.unwrap();
assert_eq!(record.seq.len(), 200_000);
}
#[test]
fn forced_format_reads_fasta_as_written() {
let data = b">a\nACGT\n";
let mut reader = FastxReader::with_format(&data[..], Format::Fasta);
assert_eq!(reader.format(), Some(Format::Fasta));
assert_eq!(reader.next().unwrap().unwrap().seq, b"ACGT");
}
}