ruststream_rdkafka/testing/
publisher.rs

1//! The in-process publisher.
2
3use std::sync::Arc;
4
5use bytes::Bytes;
6use ruststream::{DefaultPublish, OutgoingMessage, PairError, PublishPolicy, Publisher};
7
8use super::broker::{ConnectedKafkaTestBroker, TestBrokerState};
9use crate::error::KafkaError;
10use crate::publisher::KafkaPublish;
11
12/// Publisher into the in-process router.
13///
14/// Mirrors [`KafkaPublisher`](crate::KafkaPublisher) delivery semantics minus the cluster: the
15/// message name is the topic, and the partition-key header rides along for keyed worker lanes.
16#[derive(Debug, Clone)]
17pub struct KafkaTestPublisher {
18    state: Arc<TestBrokerState>,
19}
20
21impl KafkaTestPublisher {
22    pub(crate) fn new(state: Arc<TestBrokerState>) -> Self {
23        Self { state }
24    }
25}
26
27/// The in-process broker pairs the real [`KafkaPublish`] policy, so an application's include
28/// sites and routers compile unchanged against either broker.
29impl PublishPolicy<ConnectedKafkaTestBroker> for KafkaPublish {
30    type Live = KafkaTestPublisher;
31
32    async fn pair(self, connected: &ConnectedKafkaTestBroker) -> Result<Self::Live, PairError> {
33        Ok(connected.publisher(self))
34    }
35}
36
37impl DefaultPublish for ConnectedKafkaTestBroker {
38    type Policy = KafkaPublish;
39}
40
41impl Publisher for KafkaTestPublisher {
42    type Error = KafkaError;
43
44    /// Routes `msg` to subscribers of the topic named by `msg.name()`.
45    ///
46    /// # Errors
47    ///
48    /// Returns [`KafkaError::InvalidOptions`] when the topic name is empty, and
49    /// [`KafkaError::Closed`] once the transport this handle aliases has been shut down.
50    async fn publish(&self, msg: OutgoingMessage<'_>) -> Result<(), Self::Error> {
51        if msg.name().is_empty() {
52            return Err(KafkaError::InvalidOptions(
53                "topic name must not be empty; the outgoing message name is the destination \
54                 topic"
55                    .to_owned(),
56            ));
57        }
58        self.state.ensure_open(msg.name())?;
59        self.state.router.publish(
60            msg.name(),
61            &Bytes::copy_from_slice(msg.payload()),
62            msg.headers(),
63            self.state.coordinator().as_ref(),
64        );
65        Ok(())
66    }
67}
ruststream_rdkafka/testing/publisher.rs

ruststream_rdkafka/testing/
publisher.rs
