laminar_core/streaming/config/
mod.rs

1//! Configuration types for channels, sources, and sinks.
2
3/// Default buffer size for channels.
4pub const DEFAULT_BUFFER_SIZE: usize = 2048;
5
6/// Minimum buffer size.
7pub const MIN_BUFFER_SIZE: usize = 4;
8
9/// Maximum buffer size.
10pub const MAX_BUFFER_SIZE: usize = 1 << 20; // 1M entries
11
12/// Backpressure strategy when buffer is full.
13///
14/// Stored in source/sink configs and exposed in SQL DDL (`BACKPRESSURE = '...'`).
15/// Source pushes are always nonblocking and reject on count or Arrow-byte saturation.
16/// Snapshot history evicts its oldest entries independently of this setting.
17#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
18pub enum BackpressureStrategy {
19    /// Block until space is available (default).
20    #[default]
21    Block,
22    /// Drop the oldest item to make room.
23    DropOldest,
24    /// Reject the push immediately.
25    Reject,
26}
27
28impl std::str::FromStr for BackpressureStrategy {
29    type Err = String;
30
31    fn from_str(s: &str) -> Result<Self, Self::Err> {
32        match s.to_lowercase().as_str() {
33            "block" | "blocking" => Ok(Self::Block),
34            "drop" | "drop_oldest" | "dropoldest" => Ok(Self::DropOldest),
35            "reject" | "error" => Ok(Self::Reject),
36            _ => Err(format!(
37                "invalid backpressure strategy: '{s}'. Valid values: block, drop_oldest, reject"
38            )),
39        }
40    }
41}
42
43/// Wait strategy — parsed from SQL DDL for forward-compatibility.
44/// Currently unused; crossfire handles its own backoff internally.
45#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
46pub enum WaitStrategy {
47    /// Spin-loop.
48    Spin,
49    /// Spin with yields (default).
50    #[default]
51    SpinYield,
52    /// Park the thread.
53    Park,
54}
55
56impl std::str::FromStr for WaitStrategy {
57    type Err = String;
58
59    fn from_str(s: &str) -> Result<Self, Self::Err> {
60        match s.to_lowercase().as_str() {
61            "spin" => Ok(Self::Spin),
62            "spin_yield" | "spinyield" | "yield" => Ok(Self::SpinYield),
63            "park" | "parking" => Ok(Self::Park),
64            _ => Err(format!(
65                "invalid wait strategy: '{s}'. Valid values: spin, spin_yield, park"
66            )),
67        }
68    }
69}
70
71/// Channel configuration. Only `buffer_size` is used by the channel;
72/// `backpressure` and `wait_strategy` are stored for higher-level consumers.
73#[derive(Debug, Clone)]
74pub struct ChannelConfig {
75    /// Buffer size.
76    pub buffer_size: usize,
77    /// Backpressure strategy (used by catalog/snapshot ring, not the channel).
78    pub backpressure: BackpressureStrategy,
79    /// Wait strategy (reserved for future use).
80    pub wait_strategy: WaitStrategy,
81    /// Whether to track statistics (reserved for future use).
82    pub track_stats: bool,
83}
84
85impl Default for ChannelConfig {
86    fn default() -> Self {
87        Self {
88            buffer_size: DEFAULT_BUFFER_SIZE,
89            backpressure: BackpressureStrategy::Block,
90            wait_strategy: WaitStrategy::SpinYield,
91            track_stats: false,
92        }
93    }
94}
95
96impl ChannelConfig {
97    /// Creates a config with the specified buffer size.
98    #[must_use]
99    pub fn with_buffer_size(buffer_size: usize) -> Self {
100        Self {
101            buffer_size: buffer_size.clamp(MIN_BUFFER_SIZE, MAX_BUFFER_SIZE),
102            ..Default::default()
103        }
104    }
105}
106
107/// Configuration for a Source.
108#[derive(Debug, Clone)]
109pub struct SourceConfig {
110    /// Channel configuration.
111    pub channel: ChannelConfig,
112    /// Name of the source (for debugging/metrics).
113    pub name: Option<String>,
114    /// Arrow-byte limit shared by cloned producers, the input ring and queued broadcast data.
115    /// Defaults to 64 MiB. Generic `Record<T>` pushes remain count bounded; their arbitrary
116    /// heap storage is not measured. Invalid limits make `push_arrow` return `InvalidConfig`.
117    pub max_queued_bytes: usize,
118}
119
120impl Default for SourceConfig {
121    fn default() -> Self {
122        Self {
123            channel: ChannelConfig::default(),
124            name: None,
125            max_queued_bytes: super::DEFAULT_SOURCE_MAX_QUEUED_BYTES,
126        }
127    }
128}
129
130impl SourceConfig {
131    /// Creates a source config with the specified buffer size.
132    #[must_use]
133    pub fn with_buffer_size(buffer_size: usize) -> Self {
134        Self {
135            channel: ChannelConfig::with_buffer_size(buffer_size),
136            ..Self::default()
137        }
138    }
139
140    /// Creates a named source config.
141    #[must_use]
142    pub fn named(name: impl Into<String>) -> Self {
143        Self {
144            name: Some(name.into()),
145            ..Self::default()
146        }
147    }
148}
149
150#[cfg(test)]
151mod tests;
laminar_core/streaming/config/mod.rs

laminar_core/streaming/config/
mod.rs
