Struct StreamingQueueCompressor 

pub struct StreamingQueueCompressor { /* private fields */ }

Streaming compressor with queue-based API

Example

use ragc_core::{StreamingQueueCompressor, StreamingQueueConfig};
use ahash::AHashSet;

let config = StreamingQueueConfig::default();
let splitters = AHashSet::new(); // Normally from reference
let mut compressor = StreamingQueueCompressor::with_splitters(
    "output.agc",
    config,
    splitters
)?;

// Push sequences (blocks when queue is full - automatic backpressure!)
for (sample, contig_name, data) in sequences {
    compressor.push(sample, contig_name, data)?;
}

// Finalize - waits for all compression to complete
compressor.finalize()?;

Implementations

impl StreamingQueueCompressor

pub fn with_splitters( output_path: impl AsRef<Path>, config: StreamingQueueConfig, splitters: AHashSet<u64>, ) -> Result<Self>

Create a new streaming compressor with pre-computed splitters

Use this when you already have splitters (e.g., from a reference genome)

Arguments

• output_path - Path to output AGC archive
• config - Compression configuration
• splitters - Pre-computed splitter k-mers

pub fn with_full_splitter_data( output_path: impl AsRef<Path>, config: StreamingQueueConfig, splitters: AHashSet<u64>, singletons: Vec<u64>, duplicates: AHashSet<u64>, ) -> Result<Self>

Create a new streaming compressor with full splitter data for dynamic discovery

This is the preferred constructor when using adaptive mode. It accepts:

• splitters: Pre-computed splitter k-mers from reference (for initial segmentation)
• singletons: All singleton k-mers from reference (for exclusion in find_new_splitters)
• duplicates: All duplicate k-mers from reference (for exclusion in find_new_splitters)

Arguments

• output_path - Path to output AGC archive
• config - Compression configuration
• splitters - Pre-computed splitter k-mers
• singletons - Reference singleton k-mers (sorted Vec for binary search)
• duplicates - Reference duplicate k-mers

pub fn new( output_path: impl AsRef<Path>, config: StreamingQueueConfig, ) -> Result<Self>

Create compressor and determine splitters from first contig

Note: This requires at least one contig to be pushed before workers start. Consider using with_splitters() instead if you have a reference genome.

pub fn push( &mut self, sample_name: String, contig_name: String, data: Contig, ) -> Result<()>

Push a contig to the compression queue

BLOCKS if the queue is full (automatic backpressure!)

Arguments

• sample_name - Name of the sample
• contig_name - Name of the contig
• data - Contig sequence data (Vec)

Example

compressor.push("sample1".to_string(), "chr1".to_string(), vec![b'A', b'T', b'G', b'C'])?;

pub fn drain(&self) -> Result<()>

Finalize compression

This will:

1. Close the queue (no more pushes allowed)
2. Wait for all worker threads to finish processing
3. Write metadata to the archive
4. Close the archive file

Example

// ... push sequences ...
compressor.finalize()?;

pub fn sync_and_flush(&self, sample_name: &str) -> Result<()>

Insert sync tokens to trigger incremental compression of buffered segments. Call this after pushing a batch of samples to process them incrementally instead of waiting for finalize().

pub fn finalize(self) -> Result<()>

pub fn queue_stats(&self) -> QueueStats

Get current queue statistics