#![warn(missing_docs)]
#![warn(clippy::missing_panics_doc, clippy::missing_errors_doc)]
#![doc(html_root_url = "https://docs.rs/segment-buffer/0.5.3")]
#![cfg_attr(docsrs, feature(doc_cfg))]
mod cipher;
mod error;
mod segment;
mod store;
#[cfg(feature = "encryption")]
#[cfg_attr(docsrs, doc(cfg(feature = "encryption")))]
pub use cipher::{AesGcmCipher, XChaCha20Poly1305Cipher};
pub use cipher::{CipherError, SegmentCipher};
pub use error::{IoSite, Result, SegmentError};
#[cfg(feature = "loom")]
pub use segment::SegmentRange;
#[cfg(feature = "loom")]
pub use store::{RealStore, SegmentStore};
#[cfg(any(test, feature = "fuzz"))]
pub mod fuzz_hooks {
pub use crate::segment::{
filename, parse_filename, unwrap_envelope, wrap_envelope, SegmentRange,
};
}
use std::path::PathBuf;
use std::sync::Arc;
use std::time::Instant;
use parking_lot::Mutex;
use serde::de::DeserializeOwned;
use serde::Serialize;
use tracing::{debug, info};
const LOCK_FILE_NAME: &str = ".segment-buffer.lock";
#[derive(Debug, Clone, PartialEq, Eq)]
#[non_exhaustive]
pub enum FlushPolicy {
Batch(usize),
Interval(std::time::Duration),
BatchOrInterval {
batch_size: usize,
interval: std::time::Duration,
},
Manual,
}
impl Default for FlushPolicy {
fn default() -> Self {
FlushPolicy::BatchOrInterval {
batch_size: 256,
interval: std::time::Duration::from_secs(5),
}
}
}
impl FlushPolicy {
fn should_flush(&self, pending_len: usize, time_since_last_flush: std::time::Duration) -> bool {
match self {
FlushPolicy::Batch(n) => pending_len >= *n,
FlushPolicy::Interval(d) => time_since_last_flush >= *d,
FlushPolicy::BatchOrInterval {
batch_size,
interval,
} => pending_len >= *batch_size || time_since_last_flush >= *interval,
FlushPolicy::Manual => false,
}
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
#[non_exhaustive]
pub enum DurabilityPolicy {
Maximal,
#[default]
Segment,
Throughput,
}
#[non_exhaustive]
#[derive(Clone)]
pub struct SegmentConfig {
pub flush_policy: FlushPolicy,
pub max_size_bytes: u64,
pub compression_level: i32,
pub durability: DurabilityPolicy,
pub cipher: Option<Arc<dyn SegmentCipher + Send + Sync>>,
}
impl std::fmt::Debug for SegmentConfig {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("SegmentConfig")
.field("flush_policy", &self.flush_policy)
.field("max_size_bytes", &self.max_size_bytes)
.field("compression_level", &self.compression_level)
.field("durability", &self.durability)
.field("cipher", &self.cipher.as_ref().map(|_| "[set]"))
.finish()
}
}
impl Default for SegmentConfig {
fn default() -> Self {
Self {
flush_policy: FlushPolicy::default(),
max_size_bytes: 10 * 1024 * 1024 * 1024,
compression_level: 3,
durability: DurabilityPolicy::default(),
cipher: None,
}
}
}
#[derive(Debug, Clone)]
pub struct SegmentConfigBuilder {
inner: SegmentConfig,
}
impl SegmentConfigBuilder {
pub fn flush_policy(mut self, policy: FlushPolicy) -> Self {
self.inner.flush_policy = policy;
self
}
pub fn flush_at_batch_size(self, batch_size: usize) -> Self {
self.flush_policy(FlushPolicy::Batch(batch_size))
}
pub fn flush_at_interval(self, interval: std::time::Duration) -> Self {
self.flush_policy(FlushPolicy::Interval(interval))
}
pub fn flush_at_batch_or_interval(
self,
batch_size: usize,
interval: std::time::Duration,
) -> Self {
self.flush_policy(FlushPolicy::BatchOrInterval {
batch_size,
interval,
})
}
pub fn flush_manually(self) -> Self {
self.flush_policy(FlushPolicy::Manual)
}
pub fn max_size_bytes(mut self, max_size_bytes: u64) -> Self {
self.inner.max_size_bytes = max_size_bytes;
self
}
pub fn compression_level(mut self, compression_level: i32) -> Self {
self.inner.compression_level = compression_level;
self
}
pub fn durability(mut self, policy: DurabilityPolicy) -> Self {
self.inner.durability = policy;
self
}
pub fn cipher(mut self, cipher: Arc<dyn SegmentCipher + Send + Sync>) -> Self {
self.inner.cipher = Some(cipher);
self
}
#[cfg(feature = "encryption")]
#[cfg_attr(docsrs, doc(cfg(feature = "encryption")))]
pub fn recommended_cipher(self, key: [u8; 32]) -> Self {
self.cipher(Arc::new(XChaCha20Poly1305Cipher::new(&key)))
}
pub fn build(self) -> SegmentConfig {
self.inner
}
}
impl SegmentConfig {
#[must_use = "the builder is meaningless if discarded"]
pub fn builder() -> SegmentConfigBuilder {
SegmentConfigBuilder {
inner: SegmentConfig::default(),
}
}
}
#[derive(Debug, Clone)]
#[non_exhaustive]
pub struct BufferStats {
pub pending_count: u64,
pub latest_sequence: u64,
pub head_sequence: u64,
pub next_sequence: u64,
pub approx_disk_bytes: u64,
pub max_size_bytes: u64,
pub store_pressure: f32,
}
#[derive(Debug, Clone, PartialEq, Eq)]
#[non_exhaustive]
pub struct RecoveryReport {
pub segment_count: usize,
pub head_seq: u64,
pub next_seq: u64,
pub disk_bytes: u64,
pub removed_tmp_files: usize,
}
struct BufferInner<T> {
unflushed: Vec<T>,
next_seq: u64,
head_seq: u64,
last_flush: Instant,
}
#[doc(alias = "queue")]
#[doc(alias = "spool")]
#[doc(alias = "wal")]
#[doc(alias = "writeahead")]
#[doc(alias = "log")]
pub struct SegmentBuffer<T> {
dir: PathBuf,
config: SegmentConfig,
inner: Mutex<BufferInner<T>>,
approx_disk_bytes: std::sync::atomic::AtomicU64,
scan_cache: Mutex<Option<Vec<segment::SegmentRange>>>,
iteration_in_progress: std::sync::atomic::AtomicBool,
compressor: Mutex<zstd::bulk::Compressor<'static>>,
decompressor: Mutex<zstd::bulk::Decompressor<'static>>,
store: Arc<dyn store::SegmentStore + Send + Sync>,
lock_file: Option<std::fs::File>,
mtime_supported: bool,
last_dir_mtime: Mutex<Option<std::time::SystemTime>>,
}
impl<T> std::fmt::Debug for SegmentBuffer<T>
where
T: Serialize + DeserializeOwned + Clone + Send,
{
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
let stats = self.stats();
f.debug_struct("SegmentBuffer")
.field("dir", &self.dir)
.field("pending_count", &stats.pending_count)
.field("latest_sequence", &stats.latest_sequence)
.field("head_sequence", &stats.head_sequence)
.field("next_sequence", &stats.next_sequence)
.field("approx_disk_bytes", &stats.approx_disk_bytes)
.field("max_size_bytes", &stats.max_size_bytes)
.field("store_pressure", &stats.store_pressure)
.finish()
}
}
impl<T> SegmentBuffer<T>
where
T: Serialize + DeserializeOwned + Clone + Send,
{
pub fn open(dir: impl Into<PathBuf>, config: SegmentConfig) -> Result<Self> {
let (buffer, _report) = Self::open_with_report(dir, config)?;
Ok(buffer)
}
pub fn open_with_report(
dir: impl Into<PathBuf>,
config: SegmentConfig,
) -> Result<(Self, RecoveryReport)> {
let dir = dir.into();
let store: Arc<dyn store::SegmentStore + Send + Sync> =
Arc::new(store::RealStore::new(dir.clone()));
store.create_dir_all().map_err(|e| e.with_dir())?;
let lock_path = dir.join(LOCK_FILE_NAME);
let lock_file = std::fs::OpenOptions::new()
.create(true)
.read(true)
.write(true)
.truncate(false)
.open(&lock_path)
.map_err(|source| SegmentError::Io {
site: IoSite::Segment(lock_path.clone()),
source,
})?;
if fs4::FileExt::try_lock(&lock_file).is_err() {
return Err(SegmentError::Locked { path: lock_path });
}
Self::open_internal(dir, config, store, Some(lock_file))
}
#[cfg(feature = "loom")]
pub fn open_with_store(
dir: impl Into<PathBuf>,
config: SegmentConfig,
store: Arc<dyn store::SegmentStore + Send + Sync>,
) -> Result<Self> {
let dir = dir.into();
let (buffer, _report) = Self::open_internal(dir, config, store, None)?;
Ok(buffer)
}
fn open_internal(
dir: PathBuf,
config: SegmentConfig,
store: Arc<dyn store::SegmentStore + Send + Sync>,
lock_file: Option<std::fs::File>,
) -> Result<(Self, RecoveryReport)> {
store.create_dir_all().map_err(|e| e.with_dir())?;
let compressor = zstd::bulk::Compressor::new(config.compression_level)?;
let decompressor = zstd::bulk::Decompressor::new()?;
let mtime_supported = probe_mtime_capability(&dir);
let initial_mtime = std::fs::metadata(&dir).and_then(|m| m.modified()).ok();
let buffer = Self {
dir,
config,
inner: Mutex::new(BufferInner {
unflushed: Vec::new(),
next_seq: 0,
head_seq: 0,
last_flush: Instant::now(),
}),
approx_disk_bytes: std::sync::atomic::AtomicU64::new(0),
scan_cache: Mutex::new(None),
iteration_in_progress: std::sync::atomic::AtomicBool::new(false),
compressor: Mutex::new(compressor),
decompressor: Mutex::new(decompressor),
store,
lock_file,
mtime_supported,
last_dir_mtime: Mutex::new(initial_mtime),
};
let report = buffer.recover()?;
Ok((buffer, report))
}
#[track_caller]
pub fn append(&self, event: T) -> Result<u64> {
self.assert_not_reentered("append");
let (should_flush, seq) = {
let mut inner = self.inner.lock();
inner.unflushed.push(event);
inner.next_seq += 1;
let seq = inner.next_seq - 1;
let should_flush = self
.config
.flush_policy
.should_flush(inner.unflushed.len(), inner.last_flush.elapsed());
(should_flush, seq)
};
if should_flush {
self.flush()?;
}
Ok(seq)
}
#[track_caller]
pub fn flush(&self) -> Result<()> {
self.assert_not_reentered("flush");
let (events, start_seq, end_seq) = {
let mut inner = self.inner.lock();
inner.last_flush = Instant::now();
if inner.unflushed.is_empty() {
return Ok(());
}
let events = std::mem::take(&mut inner.unflushed);
inner.unflushed.reserve(events.capacity());
let count = events.len() as u64;
let end_seq = inner.next_seq - 1;
let start_seq = end_seq + 1 - count;
(events, start_seq, end_seq)
};
let compressed_len = self.write_segment(start_seq, end_seq, &events)?;
self.approx_disk_bytes
.fetch_add(compressed_len, std::sync::atomic::Ordering::Relaxed);
self.invalidate_scan_cache();
debug!(
path = self.segment_path(start_seq, end_seq).display().to_string(),
seq = start_seq,
end_seq,
count = events.len(),
bytes = compressed_len,
"Flushed segment"
);
Ok(())
}
#[track_caller]
pub fn read_from(&self, start_seq: u64, limit: usize) -> Result<Vec<T>> {
self.assert_not_reentered("read_from");
if limit == 0 {
return Ok(Vec::new());
}
let mut result: Vec<T> = Vec::with_capacity(limit.min(1024));
let segments = self.scan_segments()?;
for seg in &segments {
if result.len() >= limit {
break;
}
if seg.end < start_seq {
continue;
}
let events = self.read_segment(*seg)?;
let skip = if seg.start < start_seq {
(start_seq - seg.start) as usize
} else {
0
};
for event in events.into_iter().skip(skip) {
if result.len() >= limit {
break;
}
result.push(event);
}
}
if result.len() < limit {
let inner = self.inner.lock();
let pending_start = inner.next_seq.saturating_sub(inner.unflushed.len() as u64);
for (i, event) in inner.unflushed.iter().enumerate() {
let seq = pending_start + i as u64;
if seq < start_seq {
continue;
}
if result.len() >= limit {
break;
}
result.push(event.clone());
}
}
Ok(result)
}
pub fn for_each_from<F>(&self, start_seq: u64, limit: usize, mut f: F) -> Result<usize>
where
F: FnMut(u64, &T),
{
if limit == 0 {
return Ok(0);
}
self.iteration_in_progress
.store(true, std::sync::atomic::Ordering::Relaxed);
let _guard = IterationGuard(&self.iteration_in_progress);
let mut visited = 0usize;
let segments = self.scan_segments()?;
for seg in &segments {
if visited >= limit {
break;
}
if seg.end < start_seq {
continue;
}
let events = self.read_segment(*seg)?;
let skip = if seg.start < start_seq {
(start_seq - seg.start) as usize
} else {
0
};
for (offset, event) in events.iter().enumerate().skip(skip) {
if visited >= limit {
break;
}
let seq = seg.start + offset as u64;
f(seq, event);
visited += 1;
}
}
if visited < limit {
let inner = self.inner.lock();
let pending_start = inner.next_seq.saturating_sub(inner.unflushed.len() as u64);
for (i, event) in inner.unflushed.iter().enumerate() {
if visited >= limit {
break;
}
let seq = pending_start + i as u64;
if seq < start_seq {
continue;
}
f(seq, event);
visited += 1;
}
}
Ok(visited)
}
#[track_caller]
pub fn delete_acked(&self, acked_seq: u64) -> Result<usize> {
self.assert_not_reentered("delete_acked");
let segments = self.scan_segments()?;
let mut deleted = 0;
let mut freed_bytes: u64 = 0;
let mut new_head = None;
for seg in &segments {
if seg.end <= acked_seq {
let path = self.segment_path(seg.start, seg.end);
let file_bytes = self.store.segment_size(*seg);
freed_bytes += file_bytes;
if self.store.remove_segment(*seg)? {
deleted += 1;
debug!(
path = path.display().to_string(),
seq = seg.start,
end_seq = seg.end,
bytes = file_bytes,
"Deleted acked segment"
);
}
} else if new_head.is_none() {
new_head = Some(seg.start);
}
}
self.approx_disk_bytes
.fetch_sub(freed_bytes, std::sync::atomic::Ordering::Relaxed);
self.invalidate_scan_cache();
{
let mut inner = self.inner.lock();
let pending_start = inner.next_seq.saturating_sub(inner.unflushed.len() as u64);
inner.head_seq = new_head.unwrap_or(inner.next_seq).min(pending_start);
}
if deleted > 0 {
info!(
path = self.dir.display().to_string(),
deleted,
bytes = freed_bytes,
seq = acked_seq,
"Deleted acked segments"
);
}
Ok(deleted)
}
#[must_use = "the sequence number is meaningless if discarded"]
#[track_caller]
pub fn latest_sequence(&self) -> u64 {
self.assert_not_reentered("latest_sequence");
let inner = self.inner.lock();
if inner.next_seq == 0 {
0
} else {
inner.next_seq - 1
}
}
#[must_use = "the backlog size is meaningless if discarded"]
#[track_caller]
pub fn pending_count(&self) -> u64 {
self.assert_not_reentered("pending_count");
let inner = self.inner.lock();
inner.next_seq.saturating_sub(inner.head_seq)
}
#[must_use = "the backlog size is meaningless if discarded"]
pub fn len(&self) -> u64 {
self.pending_count()
}
#[must_use = "the emptiness flag is meaningless if discarded"]
pub fn is_empty(&self) -> bool {
self.pending_count() == 0
}
#[must_use = "the pressure value is meaningless if discarded"]
pub fn store_pressure(&self) -> f32 {
if self.config.max_size_bytes == 0 {
return 0.0;
}
let bytes = self
.approx_disk_bytes
.load(std::sync::atomic::Ordering::Relaxed);
(bytes as f32 / self.config.max_size_bytes as f32).min(1.0)
}
#[must_use = "the overload flag is meaningless if discarded"]
pub fn is_overloaded(&self) -> bool {
self.store_pressure() > 0.9
}
#[must_use = "the snapshot is meaningless if discarded"]
#[track_caller]
pub fn stats(&self) -> BufferStats {
self.assert_not_reentered("stats");
let inner = self.inner.lock();
let pending_count = inner.next_seq.saturating_sub(inner.head_seq);
let latest_sequence = if inner.next_seq == 0 {
0
} else {
inner.next_seq - 1
};
let approx_disk_bytes = self
.approx_disk_bytes
.load(std::sync::atomic::Ordering::Relaxed);
let store_pressure = if self.config.max_size_bytes == 0 {
0.0
} else {
(approx_disk_bytes as f32 / self.config.max_size_bytes as f32).min(1.0)
};
BufferStats {
pending_count,
latest_sequence,
head_sequence: inner.head_seq,
next_sequence: inner.next_seq,
approx_disk_bytes,
max_size_bytes: self.config.max_size_bytes,
store_pressure,
}
}
#[must_use = "the path is meaningless if discarded"]
pub fn path(&self) -> &std::path::Path {
&self.dir
}
#[must_use = "the config is meaningless if discarded"]
pub fn config(&self) -> &SegmentConfig {
&self.config
}
#[track_caller]
pub fn sync_disk_bytes(&self) -> Result<u64> {
self.assert_not_reentered("sync_disk_bytes");
let segments = self.scan_segments()?;
let total: u64 = segments.iter().map(|s| self.store.segment_size(*s)).sum();
self.approx_disk_bytes
.store(total, std::sync::atomic::Ordering::Relaxed);
Ok(total)
}
#[track_caller]
pub fn append_all<I>(&self, items: I) -> Result<u64>
where
I: IntoIterator<Item = T>,
{
self.assert_not_reentered("append_all");
let (should_flush, last_seq, count) = {
let mut inner = self.inner.lock();
let mut count = 0u64;
let mut last_seq = inner.next_seq.saturating_sub(1);
for item in items {
inner.unflushed.push(item);
inner.next_seq = inner.next_seq.wrapping_add(1);
last_seq = inner.next_seq - 1;
count += 1;
}
if count == 0 {
return Ok(inner.next_seq.saturating_sub(1));
}
let should_flush = self
.config
.flush_policy
.should_flush(inner.unflushed.len(), inner.last_flush.elapsed());
(should_flush, last_seq, count)
};
debug_assert!(count > 0);
if should_flush {
self.flush()?;
}
Ok(last_seq)
}
#[track_caller]
pub fn iter_from(&self, start_seq: u64, limit: usize) -> Result<SegmentIter<'_, T>> {
self.assert_not_reentered("iter_from");
if limit == 0 {
return Ok(SegmentIter {
inner: Vec::new().into_iter(),
_phantom: std::marker::PhantomData,
});
}
let items = self.read_from(start_seq, limit)?;
let indexed: Vec<(u64, T)> = items
.into_iter()
.enumerate()
.map(|(i, item)| (start_seq + i as u64, item))
.collect();
Ok(SegmentIter {
inner: indexed.into_iter(),
_phantom: std::marker::PhantomData,
})
}
#[track_caller]
fn assert_not_reentered(&self, method: &'static str) {
if self
.iteration_in_progress
.load(std::sync::atomic::Ordering::Relaxed)
{
panic!(
"{method}: cannot call from within a for_each_from callback \
(the buffer mutex is held; re-entry would deadlock)"
);
}
}
fn recover(&self) -> Result<RecoveryReport> {
let removed_tmp_files = self.store.clean_tmp()?;
let segments = self.scan_segments()?;
let total_bytes: u64 = segments.iter().map(|s| self.store.segment_size(*s)).sum();
let (head_seq, next_seq) = match (segments.first(), segments.last()) {
(Some(first), Some(last)) => (first.start, last.end + 1),
_ => (0, 0),
};
let segment_count = segments.len();
{
let mut inner = self.inner.lock();
inner.head_seq = head_seq;
inner.next_seq = next_seq;
}
self.approx_disk_bytes
.store(total_bytes, std::sync::atomic::Ordering::Relaxed);
*self.scan_cache.lock() = Some(segments.clone());
info!(
path = self.dir.display().to_string(),
segments = segment_count,
seq = head_seq,
end_seq = next_seq,
bytes = total_bytes,
removed_tmp = removed_tmp_files,
"Segment buffer recovered"
);
Ok(RecoveryReport {
segment_count,
head_seq,
next_seq,
disk_bytes: total_bytes,
removed_tmp_files,
})
}
fn write_segment(&self, start: u64, end: u64, events: &[T]) -> Result<u64> {
let path = self.segment_path(start, end);
let range = segment::SegmentRange::new(start, end);
let mut compressor = self.compressor.lock();
let bytes = segment::encode_segment(
self.config.cipher.as_deref(),
&mut compressor,
&path,
events,
)?;
drop(compressor);
self.store
.write_atomic(range, &bytes, self.config.durability)
.map_err(|e| e.with_path(&path))
}
fn read_segment(&self, seg: segment::SegmentRange) -> Result<Vec<T>> {
let path = self.segment_path(seg.start, seg.end);
let raw = self.store.read_bytes(seg).map_err(|e| e.with_path(&path))?;
let mut decompressor = self.decompressor.lock();
segment::decode_segment(
self.config.cipher.as_deref(),
&mut decompressor,
&raw,
&path,
)
.map_err(|e| e.with_path(&path))
}
fn scan_segments(&self) -> Result<Vec<segment::SegmentRange>> {
{
let cache = self.scan_cache.lock();
if let Some(ref segments) = *cache {
if !self.mtime_supported || !self.dir_mtime_changed() {
return Ok(segments.clone());
}
}
}
let segments = self
.store
.scan()
.map_err(|e| e.with_dir())
.map_err(|e| e.with_path(&self.dir))?;
let fresh_mtime = std::fs::metadata(&self.dir).and_then(|m| m.modified()).ok();
let mut cache = self.scan_cache.lock();
*cache = Some(segments.clone());
*self.last_dir_mtime.lock() = fresh_mtime;
Ok(segments)
}
fn dir_mtime_changed(&self) -> bool {
let current = match std::fs::metadata(&self.dir).and_then(|m| m.modified()) {
Ok(t) => t,
Err(_) => return true, };
let cached = *self.last_dir_mtime.lock();
match cached {
Some(prev) => prev != current,
None => true,
}
}
fn invalidate_scan_cache(&self) {
let mut cache = self.scan_cache.lock();
*cache = None;
}
fn segment_path(&self, start: u64, end: u64) -> PathBuf {
self.dir.join(segment::filename(start, end))
}
}
struct IterationGuard<'a>(&'a std::sync::atomic::AtomicBool);
impl Drop for IterationGuard<'_> {
fn drop(&mut self) {
self.0.store(false, std::sync::atomic::Ordering::Relaxed);
}
}
pub struct SegmentIter<'a, T> {
inner: std::vec::IntoIter<(u64, T)>,
_phantom: std::marker::PhantomData<&'a SegmentBuffer<T>>,
}
impl<T> Iterator for SegmentIter<'_, T> {
type Item = (u64, T);
fn next(&mut self) -> Option<Self::Item> {
self.inner.next()
}
fn size_hint(&self) -> (usize, Option<usize>) {
self.inner.size_hint()
}
}
impl<T> std::iter::FusedIterator for SegmentIter<'_, T> {}
impl<T> Drop for SegmentBuffer<T> {
fn drop(&mut self) {
if let Some(lock_file) = self.lock_file.take() {
let _ = fs4::FileExt::unlock(&lock_file);
drop(lock_file);
}
}
}
fn probe_mtime_capability(dir: &std::path::Path) -> bool {
let sentinel = dir.join(".segment-buffer.mtime-probe");
let _ = std::fs::write(&sentinel, b"a");
let t1 = std::fs::metadata(&sentinel).and_then(|m| m.modified()).ok();
std::thread::sleep(std::time::Duration::from_millis(15));
let _ = std::fs::write(&sentinel, b"b");
let t2 = std::fs::metadata(&sentinel).and_then(|m| m.modified()).ok();
let _ = std::fs::remove_file(&sentinel);
matches!((t1, t2), (Some(a), Some(b)) if a != b)
}
const _: () = {
const fn assert_send_sync<T: Send + Sync>() {}
assert_send_sync::<SegmentBuffer<()>>();
};
#[cfg(test)]
mod tests;
#[cfg(test)]
mod property_tests;
#[cfg(doctest)]
mod example_doctests {
#[doc = concat!("```rust\n", include_str!("../examples/basic_usage.rs"), "\n```")]
const BASIC_USAGE: () = ();
#[doc = concat!("```rust\n", include_str!("../examples/backpressure.rs"), "\n```")]
const BACKPRESSURE: () = ();
#[doc = concat!("```rust\n", include_str!("../examples/crash_recovery.rs"), "\n```")]
const CRASH_RECOVERY: () = ();
#[doc = concat!("```rust\n", include_str!("../examples/mpmc.rs"), "\n```")]
const MPMC: () = ();
#[cfg(feature = "encryption")]
#[doc = concat!("```rust\n", include_str!("../examples/encrypted.rs"), "\n```")]
const ENCRYPTED: () = ();
}