1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
//! Error type returned by Redis broker operations.
use std::error::Error as StdError;
use thiserror::Error;
/// Errors surfaced by the Redis broker implementation.
#[derive(Debug, Error)]
#[non_exhaustive]
pub enum RedisError {
/// Failed to establish or use the underlying `fred` connection.
#[error("redis connection error: {0}")]
Connect(#[source] Box<dyn StdError + Send + Sync>),
/// Failed to publish (`XADD`) a message to a stream.
#[error("redis publish error: {0}")]
Publish(#[source] Box<dyn StdError + Send + Sync>),
/// Failed to open a subscription (consumer-group creation or the first read).
#[error("redis subscribe error: {0}")]
Subscribe(#[source] Box<dyn StdError + Send + Sync>),
/// A stream read (`XREADGROUP` / `XAUTOCLAIM`) or acknowledgement (`XACK`) failed.
#[error("redis stream error: {0}")]
Stream(#[source] Box<dyn StdError + Send + Sync>),
/// The operation ran through a handle aliasing a connection that was already shut down
/// ([`ConnectedBroker::shutdown`](ruststream::ConnectedBroker::shutdown)).
///
/// The lifecycle ladder makes misuse by the owner of the connected broker a compile error;
/// publishers handed out before the shutdown outlive it, so their operations report this
/// instead of running against a dead pool.
#[error("the redis broker connection is shut down")]
ShutDown,
/// `begin_transaction` was called while a transaction is already open on this publisher handle.
#[error("a transaction is already open on this publisher handle")]
TransactionBusy,
/// `commit` or `abort` was called with no open transaction on this publisher handle.
#[error("no transaction is open on this publisher handle")]
NoTransaction,
/// The supplied subscription descriptor combines fields in a way the broker cannot honour
/// (for example a [`RedisStream`](crate::RedisStream) with no consumer group, or a bare-string
/// subscription with no broker-wide default group).
#[error("invalid subscribe options: {0}")]
InvalidOptions(String),
/// The connected server is older than the subscription's read mode needs.
///
/// [`RedisStream::claiming`](crate::RedisStream::claiming) issues `XREADGROUP ... CLAIM`,
/// which Redis 8.4.0 added. The check runs where the subscription resolves against the
/// connected broker, once and before the first read, so the service fails at startup with
/// both versions named instead of meeting a syntax error on its first message.
///
/// The message is the whole error: it already names the subscription, the mode and the two
/// versions, and a prefix would only repeat them.
#[error("{0}")]
ServerTooOld(String),
/// A window of a `.pipeline()` subscription failed to send what it owed.
///
/// Reported once, on the subscription's delivery stream, and the message names the
/// subscription, how many commands the flush carried and how many failed. On the forms that
/// settle, the entries whose settles failed stay pending and are delivered again.
#[error("{0}")]
Flush(String),
}
impl RedisError {
/// Wraps a `fred` error as a [`RedisError::Stream`].
pub(crate) fn stream(err: fred::error::Error) -> Self {
Self::Stream(Box::new(err))
}
/// Wraps a `fred` error as a [`RedisError::Subscribe`].
pub(crate) fn subscribe(err: fred::error::Error) -> Self {
Self::Subscribe(Box::new(err))
}
/// Wraps a `fred` error as a [`RedisError::Publish`].
pub(crate) fn publish(err: fred::error::Error) -> Self {
Self::Publish(Box::new(err))
}
}