ruststream_rdkafka/testing/
publisher.rs

1//! The in-process publisher.
2
3use std::sync::Arc;
4
5use bytes::Bytes;
6use ruststream::{OutgoingMessage, Publisher};
7
8use super::broker::TestBrokerState;
9use crate::error::KafkaError;
10
11/// Publisher into the in-process router.
12///
13/// Mirrors [`KafkaPublisher`](crate::KafkaPublisher) delivery semantics minus the cluster: the
14/// message name is the topic, and the partition-key header rides along for keyed worker lanes.
15#[derive(Debug, Clone)]
16pub struct KafkaTestPublisher {
17    state: Arc<TestBrokerState>,
18}
19
20impl KafkaTestPublisher {
21    pub(crate) fn new(state: Arc<TestBrokerState>) -> Self {
22        Self { state }
23    }
24}
25
26impl Publisher for KafkaTestPublisher {
27    type Error = KafkaError;
28
29    /// Routes `msg` to subscribers of the topic named by `msg.name()`.
30    ///
31    /// # Errors
32    ///
33    /// Returns [`KafkaError::InvalidOptions`] when the topic name is empty.
34    async fn publish(&self, msg: OutgoingMessage<'_>) -> Result<(), Self::Error> {
35        if msg.name().is_empty() {
36            return Err(KafkaError::InvalidOptions(
37                "topic name must not be empty; the outgoing message name is the destination \
38                 topic"
39                    .to_owned(),
40            ));
41        }
42        self.state.router.publish(
43            msg.name(),
44            &Bytes::copy_from_slice(msg.payload()),
45            msg.headers(),
46            self.state.coordinator().as_ref(),
47        );
48        Ok(())
49    }
50}
ruststream_rdkafka/testing/publisher.rs

ruststream_rdkafka/testing/
publisher.rs
