ruststream_kinesis/
error.rs

1//! The crate-level error type.
2
3use std::error::Error as StdError;
4
5/// Errors returned by the Amazon Kinesis Data Streams broker.
6///
7/// One enum for the whole crate, variants by source, per the `RustStream` broker conventions.
8/// The wrapped sources are boxed `std` errors formatted with their full cause chain, so the
9/// public API does not leak the SDK's layered error types.
10#[derive(Debug, thiserror::Error)]
11#[non_exhaustive]
12pub enum KinesisError {
13    /// Loading the AWS configuration failed.
14    #[error("aws config error: {0}")]
15    Config(String),
16
17    /// A stream admin call (describe, create, shard listing) failed.
18    #[error("kinesis stream error for '{stream}': {source}")]
19    Stream {
20        /// The stream the call was about.
21        stream: String,
22        /// The SDK's failure, with its cause chain.
23        #[source]
24        source: Box<dyn StdError + Send + Sync>,
25    },
26
27    /// Reading a shard failed permanently.
28    #[error("kinesis read error on '{stream}' shard '{shard}': {source}")]
29    Read {
30        /// The stream the shard belongs to.
31        stream: String,
32        /// The shard the read targeted.
33        shard: String,
34        /// The SDK's failure, with its cause chain.
35        #[source]
36        source: Box<dyn StdError + Send + Sync>,
37    },
38
39    /// A delivered record is KPL-aggregated, which this crate does not deaggregate yet;
40    /// failing loudly beats handing the handler an opaque protobuf blob.
41    #[error("kinesis record on shard '{shard}' is KPL-aggregated (unsupported)")]
42    AggregatedRecord {
43        /// The shard the record arrived on.
44        shard: String,
45    },
46
47    /// Writing a record failed.
48    #[error("kinesis publish error to '{stream}': {source}")]
49    Publish {
50        /// The stream the record targeted.
51        stream: String,
52        /// The SDK's failure, with its cause chain.
53        #[source]
54        source: Box<dyn StdError + Send + Sync>,
55    },
56
57    /// The lease store failed.
58    #[error("kinesis lease store error for stream '{stream}' shard '{shard}': {source}")]
59    Lease {
60        /// The stream the shard belongs to.
61        stream: String,
62        /// The shard whose lease was involved.
63        shard: String,
64        /// The store's failure.
65        #[source]
66        source: Box<dyn StdError + Send + Sync>,
67    },
68
69    /// The handle is used before `connect` filled the shared connection, or after `shutdown`.
70    #[error("kinesis broker is not connected")]
71    NotConnected,
72
73    /// The broker already reads this stream.
74    ///
75    /// A broker leases and checkpoints a stream's shards under its one owner id, so a second
76    /// subscription to the stream on the same broker would share those leases with the first and
77    /// lose records. A service that reads one stream twice reads it through a second
78    /// `KinesisBroker` with a lease store of its own.
79    #[error(
80        "kinesis broker already reads stream '{stream}': a second subscription would share its \
81         shard leases and checkpoints; read the stream again through a second `KinesisBroker` \
82         with a lease store of its own"
83    )]
84    AlreadySubscribed {
85        /// The stream the refused subscription named.
86        stream: String,
87    },
88
89    /// A stream descriptor is invalid.
90    #[error("invalid kinesis descriptor: {0}")]
91    Invalid(String),
92}
93
94/// Formats an SDK error with its full cause chain and boxes it.
95pub(crate) fn sdk_err<E, R>(
96    err: &aws_sdk_kinesis::error::SdkError<E, R>,
97) -> Box<dyn StdError + Send + Sync>
98where
99    E: StdError + Send + Sync + 'static,
100    R: std::fmt::Debug + Send + Sync + 'static,
101{
102    Box::from(aws_sdk_kinesis::error::DisplayErrorContext(err).to_string())
103}
ruststream_kinesis/error.rs

ruststream_kinesis/
error.rs
