ruststream_pulsar/
publisher.rs

1//! [`PulsarPublisher`] and its [`PulsarPublish`] policy.
2
3use std::sync::Arc;
4
5use pulsar::TokioExecutor;
6use ruststream::{OutgoingMessage, PairError, PublishPolicy, Publisher};
7use tokio::sync::Mutex;
8
9use crate::broker::{ConnectedPulsarBroker, Core, CoreCell};
10use crate::error::{PulsarError, box_err};
11use crate::message::to_pulsar_message;
12use crate::topic::PulsarTopic;
13
14pub(crate) type PulsarProducer = pulsar::Producer<TokioExecutor>;
15
16/// Publishes messages to Pulsar topics, one producer per topic, created lazily and shared
17/// through the broker core (so `shutdown` can close them).
18///
19/// A `partition-key` header becomes the message's partition key, which keyed routing and
20/// `KeyShared` subscriptions order by. Awaits the broker's send receipt, so `Ok` means the
21/// broker stored the message. Buildable before `connect` and usable until `shutdown`;
22/// afterwards every publish reports [`PulsarError::NotConnected`].
23#[derive(Clone)]
24pub struct PulsarPublisher {
25    cell: CoreCell,
26}
27
28impl std::fmt::Debug for PulsarPublisher {
29    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
30        f.debug_struct("PulsarPublisher").finish_non_exhaustive()
31    }
32}
33
34impl PulsarPublisher {
35    pub(crate) fn new(cell: CoreCell) -> Self {
36        Self { cell }
37    }
38
39    fn core(&self) -> Result<&Core, PulsarError> {
40        let core = self.cell.get().ok_or(PulsarError::NotConnected)?;
41        core.ensure_open()?;
42        Ok(core)
43    }
44
45    /// The per-topic producer, created on first use and cached on the core.
46    // The map guard intentionally spans the build so two callers cannot race a double
47    // producer for the same topic.
48    #[allow(clippy::significant_drop_tightening)]
49    async fn producer_for(
50        &self,
51        core: &Core,
52        topic: &str,
53    ) -> Result<Arc<Mutex<PulsarProducer>>, PulsarError> {
54        let full = PulsarTopic::parse(topic)?.as_str().to_owned();
55        let mut producers = core.producers.lock().await;
56        if let Some(producer) = producers.get(&full) {
57            return Ok(Arc::clone(producer));
58        }
59        let producer = Box::pin(core.client.producer().with_topic(&full).build())
60            .await
61            .map_err(|e| PulsarError::Publish {
62                topic: topic.to_owned(),
63                source: box_err(e),
64            })?;
65        let producer = Arc::new(Mutex::new(producer));
66        producers.insert(full, Arc::clone(&producer));
67        Ok(producer)
68    }
69}
70
71impl Publisher for PulsarPublisher {
72    type Error = PulsarError;
73
74    async fn publish(&self, msg: OutgoingMessage<'_>) -> Result<(), Self::Error> {
75        let core = self.core()?;
76        let producer = Box::pin(self.producer_for(core, msg.name())).await?;
77        let message = to_pulsar_message(&msg);
78        let receipt = {
79            let mut producer = producer.lock().await;
80            Box::pin(producer.send_non_blocking(message))
81                .await
82                .map_err(|e| PulsarError::Publish {
83                    topic: msg.name().to_owned(),
84                    source: box_err(e),
85                })?
86        };
87        receipt.await.map(|_| ()).map_err(|e| PulsarError::Publish {
88            topic: msg.name().to_owned(),
89            source: box_err(e),
90        })
91    }
92}
93
94/// The publish policy for [`PulsarPublisher`]: pure declaration, constructible anywhere,
95/// paired with the connected broker by the runtime after `connect`.
96///
97/// # Examples
98///
99/// ```
100/// use ruststream_pulsar::PulsarPublish;
101///
102/// let policy = PulsarPublish::default();
103/// # let _ = policy;
104/// ```
105#[derive(Debug, Clone, Copy, Default)]
106#[must_use]
107pub struct PulsarPublish;
108
109impl PublishPolicy<ConnectedPulsarBroker> for PulsarPublish {
110    type Live = PulsarPublisher;
111
112    async fn pair(self, connected: &ConnectedPulsarBroker) -> Result<Self::Live, PairError> {
113        Ok(connected.publisher())
114    }
115}
ruststream_pulsar/publisher.rs

ruststream_pulsar/
publisher.rs
