ruststream-rdkafka 0.5.0

Apache Kafka broker implementation for the RustStream messaging framework, backed by rdkafka / librdkafka.
Documentation 
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

//! The crate error type shared by the broker, publishers, and subscribers.

use std::error::Error as StdError;

use thiserror::Error;

/// Errors returned by [`KafkaBroker`](crate::KafkaBroker) and the types it hands out.
///
/// Underlying [`rdkafka`](https://docs.rs/rdkafka) errors are boxed as sources so the client
/// library does not leak into this crate's public API surface.
#[derive(Debug, Error)]
#[non_exhaustive]
pub enum KafkaError {
    /// Creating a client failed or the cluster was unreachable during the connect probe.
    #[error("kafka connection error: {0}")]
    Connect(#[source] Box<dyn StdError + Send + Sync>),

    /// Publishing a message failed or the broker did not confirm its delivery.
    #[error("kafka publish error: {0}")]
    Publish(#[source] Box<dyn StdError + Send + Sync>),

    /// Creating a consumer or subscribing it to its topic failed.
    #[error("kafka subscribe error: {0}")]
    Subscribe(#[source] Box<dyn StdError + Send + Sync>),

    /// Receiving a delivery from an open consumer failed.
    #[error("kafka consume error: {0}")]
    Consume(#[source] Box<dyn StdError + Send + Sync>),

    /// An operation needed the live connection before `Broker::connect` resolved it.
    ///
    /// The runtime connects the broker once at startup; a publisher handed out earlier resolves
    /// the shared connection on first use. Seeing this error means the operation ran before
    /// `connect` completed.
    #[error("kafka broker is not connected; `Broker::connect` must complete first")]
    NotConnected,

    /// The requested combination of options cannot be executed.
    ///
    /// The message names the offending option and the remediation.
    #[error("invalid options: {0}")]
    InvalidOptions(String),
}

impl KafkaError {
    pub(crate) fn connect(err: rdkafka::error::KafkaError) -> Self {
        Self::Connect(Box::new(err))
    }

    pub(crate) fn publish(err: rdkafka::error::KafkaError) -> Self {
        Self::Publish(Box::new(err))
    }

    pub(crate) fn subscribe(err: rdkafka::error::KafkaError) -> Self {
        Self::Subscribe(Box::new(err))
    }

    pub(crate) fn consume(err: rdkafka::error::KafkaError) -> Self {
        Self::Consume(Box::new(err))
    }
}