ruststream_gcp_pubsub/
publisher.rs

1//! [`PubSubPublisher`] and its [`PubSubPublish`] policy.
2
3use google_cloud_pubsub::client::Publisher as GcpPublisher;
4use ruststream::{OutgoingMessage, PairError, PublishPolicy, Publisher};
5
6use crate::broker::{ConnectedPubSubBroker, Core, CoreCell};
7use crate::error::{PubSubError, box_err};
8use crate::message::to_gcp_message;
9
10/// Publishes messages to Pub/Sub topics, one client publisher per topic, created lazily and
11/// shared through the broker core (so `shutdown` can flush buffered batches).
12///
13/// The destination name is the topic id (short or full resource name). A `partition-key`
14/// header becomes the message's ordering key; per-key FIFO is the client's ordered path.
15/// Buildable before `connect` and usable until `shutdown`; afterwards every publish reports
16/// [`PubSubError::NotConnected`] instead of silently succeeding.
17#[derive(Clone)]
18pub struct PubSubPublisher {
19    cell: CoreCell,
20}
21
22impl std::fmt::Debug for PubSubPublisher {
23    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
24        f.debug_struct("PubSubPublisher").finish_non_exhaustive()
25    }
26}
27
28impl PubSubPublisher {
29    pub(crate) fn new(cell: CoreCell) -> Self {
30        Self { cell }
31    }
32
33    fn core(&self) -> Result<&Core, PubSubError> {
34        let core = self.cell.get().ok_or(PubSubError::NotConnected)?;
35        core.ensure_open()?;
36        Ok(core)
37    }
38
39    /// The per-topic client publisher, created on first use and cached on the core.
40    async fn publisher_for(&self, core: &Core, topic: &str) -> GcpPublisher {
41        let name = core.topic_name(topic);
42        let mut publishers = core.publishers.lock().await;
43        if let Some(publisher) = publishers.get(&name) {
44            return publisher.clone();
45        }
46        // Sync and infallible off the connected BasePublisher; the network work happened in
47        // connect.
48        let publisher = core.base_publisher.publisher(name.clone()).build();
49        publishers.insert(name, publisher.clone());
50        publisher
51    }
52}
53
54impl Publisher for PubSubPublisher {
55    type Error = PubSubError;
56
57    async fn publish(&self, msg: OutgoingMessage<'_>) -> Result<(), Self::Error> {
58        let core = self.core()?;
59        let publisher = self.publisher_for(core, msg.name()).await;
60        let (message, ordering_key) = to_gcp_message(&msg);
61        match publisher.publish(message).await {
62            Ok(_message_id) => Ok(()),
63            Err(err) => {
64                // An error on an ordered key pauses the key; resume so the pause cannot wedge
65                // every later publish on this key, and let the caller see this failure.
66                if !ordering_key.is_empty() {
67                    publisher.resume_publish(ordering_key);
68                }
69                Err(PubSubError::Publish {
70                    topic: core.topic_name(msg.name()),
71                    source: box_err(err),
72                })
73            }
74        }
75    }
76}
77
78/// The publish policy for [`PubSubPublisher`]: pure declaration, constructible anywhere,
79/// paired with the connected broker by the runtime after `connect`.
80///
81/// # Examples
82///
83/// ```
84/// use ruststream_gcp_pubsub::PubSubPublish;
85///
86/// let policy = PubSubPublish::default();
87/// # let _ = policy;
88/// ```
89#[derive(Debug, Clone, Copy, Default)]
90#[must_use]
91pub struct PubSubPublish;
92
93impl PublishPolicy<ConnectedPubSubBroker> for PubSubPublish {
94    type Live = PubSubPublisher;
95
96    async fn pair(self, connected: &ConnectedPubSubBroker) -> Result<Self::Live, PairError> {
97        Ok(connected.publisher())
98    }
99}
ruststream_gcp_pubsub/publisher.rs

ruststream_gcp_pubsub/
publisher.rs
