ruststream-rdkafka 0.5.1

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

//! The in-process publisher.

use std::sync::Arc;

use bytes::Bytes;
use ruststream::{OutgoingMessage, Publisher};

use super::broker::TestBrokerState;
use crate::error::KafkaError;

/// Publisher into the in-process router.
///
/// Mirrors [`KafkaPublisher`](crate::KafkaPublisher) delivery semantics minus the cluster: the
/// message name is the topic, and the partition-key header rides along for keyed worker lanes.
#[derive(Debug, Clone)]
pub struct KafkaTestPublisher {
    state: Arc<TestBrokerState>,
}

impl KafkaTestPublisher {
    pub(crate) fn new(state: Arc<TestBrokerState>) -> Self {
        Self { state }
    }
}

impl Publisher for KafkaTestPublisher {
    type Error = KafkaError;

    /// Routes `msg` to subscribers of the topic named by `msg.name()`.
    ///
    /// # Errors
    ///
    /// Returns [`KafkaError::InvalidOptions`] when the topic name is empty.
    async fn publish(&self, msg: OutgoingMessage<'_>) -> Result<(), Self::Error> {
        if msg.name().is_empty() {
            return Err(KafkaError::InvalidOptions(
                "topic name must not be empty; the outgoing message name is the destination \
                 topic"
                    .to_owned(),
            ));
        }
        self.state.router.publish(
            msg.name(),
            &Bytes::copy_from_slice(msg.payload()),
            msg.headers(),
            self.state.coordinator().as_ref(),
        );
        Ok(())
    }
}