laminar_core/streaming/error/
mod.rs

1//! Streaming error types.
2
3use std::fmt;
4
5/// Error type for streaming operations.
6#[derive(Debug, Clone, PartialEq, Eq)]
7pub enum StreamingError {
8    /// Channel is full and backpressure strategy is Error.
9    ChannelFull,
10
11    /// A single Arrow batch exceeds the source's entire byte budget.
12    BatchTooLarge {
13        /// Charged retained Arrow bytes.
14        bytes: usize,
15        /// Configured per-source limit.
16        limit: usize,
17    },
18
19    /// Channel is closed (all receivers dropped).
20    ChannelClosed,
21
22    /// Channel is disconnected (all senders dropped).
23    Disconnected,
24
25    /// Invalid configuration provided.
26    InvalidConfig(String),
27
28    /// Schema mismatch during `push_arrow` operation.
29    SchemaMismatch {
30        /// Expected schema field names.
31        expected: Vec<String>,
32        /// Actual schema field names.
33        actual: Vec<String>,
34    },
35
36    /// Operation timed out.
37    Timeout,
38
39    /// Internal error.
40    Internal(String),
41}
42
43impl fmt::Display for StreamingError {
44    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
45        match self {
46            Self::ChannelFull => write!(f, "channel is full"),
47            Self::BatchTooLarge { bytes, limit } => {
48                write!(
49                    f,
50                    "Arrow batch retains {bytes} bytes, exceeding source limit {limit}"
51                )
52            }
53            Self::ChannelClosed => write!(f, "channel is closed"),
54            Self::Disconnected => write!(f, "channel is disconnected"),
55            Self::InvalidConfig(msg) => write!(f, "invalid configuration: {msg}"),
56            Self::SchemaMismatch { expected, actual } => {
57                write!(f, "schema mismatch: expected {expected:?}, got {actual:?}")
58            }
59            Self::Timeout => write!(f, "operation timed out"),
60            Self::Internal(msg) => write!(f, "internal error: {msg}"),
61        }
62    }
63}
64
65impl std::error::Error for StreamingError {}
66
67/// Error returned from `try_push` operations.
68#[derive(Debug)]
69pub struct TryPushError<T> {
70    /// The value that could not be pushed.
71    pub value: T,
72    /// The error that occurred.
73    pub error: StreamingError,
74}
75
76impl<T> TryPushError<T> {
77    /// Creates a new error indicating the channel is full.
78    #[must_use]
79    pub fn full(value: T) -> Self {
80        Self {
81            value,
82            error: StreamingError::ChannelFull,
83        }
84    }
85
86    /// Creates a new error indicating the receiver was dropped.
87    #[must_use]
88    pub fn disconnected(value: T) -> Self {
89        Self {
90            value,
91            error: StreamingError::Disconnected,
92        }
93    }
94
95    /// Consumes the error and returns the value that could not be pushed.
96    #[must_use]
97    pub fn into_inner(self) -> T {
98        self.value
99    }
100}
101
102impl<T: fmt::Debug> fmt::Display for TryPushError<T> {
103    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
104        write!(f, "try_push failed: {}", self.error)
105    }
106}
107
108impl<T: fmt::Debug> std::error::Error for TryPushError<T> {}
109
110/// Error returned from `recv` operations.
111#[derive(Debug, Clone, PartialEq, Eq)]
112pub enum RecvError {
113    /// Channel is disconnected (all senders dropped).
114    Disconnected,
115
116    /// Operation timed out.
117    Timeout,
118}
119
120impl fmt::Display for RecvError {
121    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
122        match self {
123            Self::Disconnected => write!(f, "channel disconnected"),
124            Self::Timeout => write!(f, "recv timed out"),
125        }
126    }
127}
128
129impl std::error::Error for RecvError {}
130
131#[cfg(test)]
132mod tests;
laminar_core/streaming/error/mod.rs

laminar_core/streaming/error/
mod.rs
