ruststream_nats/
publisher.rs

1//! Core NATS publishing: the [`NatsPublish`] policy and its live [`NatsPublisher`].
2
3use std::fmt::{Debug, Formatter};
4use std::sync::Arc;
5
6use async_nats::Client;
7use bytes::Bytes;
8use ruststream::{OutgoingMessage, PairError, PublishPolicy, Publisher};
9
10use crate::broker::{ConnectedNatsBroker, NatsConnection};
11use crate::{convert::headers_to_nats, error::NatsError};
12
13use self::sealed::Sealed;
14
15mod sealed {
16    /// Seals [`NatsPublishPolicy`](super::NatsPublishPolicy): pairing a NATS publisher is
17    /// synchronous and infallible for both of this crate's policies, and the synchronous
18    /// [`publisher`](crate::ConnectedNatsBroker::publisher) accessor depends on that.
19    pub trait Sealed {}
20
21    impl Sealed for super::NatsPublish {}
22    impl Sealed for crate::jetstream::JetStreamPublish {}
23}
24
25/// A publish policy that pairs with a connected NATS broker without I/O.
26///
27/// Both NATS policies hold nothing but publish options, so bringing one alive is a constructor
28/// call rather than broker work. That is what lets
29/// [`ConnectedNatsBroker::publisher`](crate::ConnectedNatsBroker::publisher) be synchronous;
30/// [`PublishPolicy::pair`], the framework-side entry point, delegates here.
31pub trait NatsPublishPolicy: PublishPolicy<ConnectedNatsBroker> + Sealed {
32    /// Pairs the policy with the connected broker, producing the live publisher.
33    #[must_use]
34    fn bind(self, connected: &ConnectedNatsBroker) -> Self::Live;
35}
36
37/// The Core NATS publish policy: pure declaration, constructible anywhere.
38///
39/// Core NATS publishing carries no per-publisher options (subject and headers travel with each
40/// message), so the policy is a unit marker. It pairs into [`NatsPublisher`], which also serves
41/// the [`RequestReply`](ruststream::RequestReply) capability, and it is the broker's
42/// [`DefaultPublish`](ruststream::DefaultPublish) policy, so a `publish("subject")` handler
43/// mounted without an explicit publisher replies through it.
44///
45/// # Examples
46///
47/// ```
48/// use ruststream_nats::NatsPublish;
49///
50/// let policy = NatsPublish;
51/// # let _ = policy;
52/// ```
53#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
54#[must_use]
55pub struct NatsPublish;
56
57impl PublishPolicy<ConnectedNatsBroker> for NatsPublish {
58    type Live = NatsPublisher;
59
60    async fn pair(self, connected: &ConnectedNatsBroker) -> Result<Self::Live, PairError> {
61        Ok(self.bind(connected))
62    }
63}
64
65impl NatsPublishPolicy for NatsPublish {
66    fn bind(self, connected: &ConnectedNatsBroker) -> Self::Live {
67        NatsPublisher::new(Arc::clone(connected.connection()))
68    }
69}
70
71/// The live Core NATS publisher. Cheap to clone.
72///
73/// Exists only from a [`ConnectedNatsBroker`], so it always has a connection. It aliases that
74/// connection, though, and may outlive it: after the broker shuts down every publish reports
75/// [`NatsError::Closed`] instead of silently succeeding against a dead connection.
76#[derive(Clone)]
77pub struct NatsPublisher {
78    connection: Arc<NatsConnection>,
79}
80
81impl Debug for NatsPublisher {
82    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
83        f.debug_struct("NatsPublisher").finish_non_exhaustive()
84    }
85}
86
87impl NatsPublisher {
88    pub(crate) const fn new(connection: Arc<NatsConnection>) -> Self {
89        Self { connection }
90    }
91
92    pub(crate) fn client_for(&self, subject: &str) -> Result<Client, NatsError> {
93        self.connection.live_client(subject).cloned()
94    }
95}
96
97impl Publisher for NatsPublisher {
98    type Error = NatsError;
99
100    /// # Cancel safety
101    ///
102    /// Core NATS publishing is fire-and-forget: the message is handed to the connection's writer
103    /// without waiting for the server. Dropping the future may leave the message either sent or
104    /// unsent, with no way to tell which.
105    async fn publish(&self, msg: OutgoingMessage<'_>) -> Result<(), Self::Error> {
106        let client = self.client_for(msg.name())?;
107        let subject = msg.name().to_owned();
108        let payload = Bytes::copy_from_slice(msg.payload());
109        let result = match headers_to_nats(msg.headers()) {
110            Some(headers) => client.publish_with_headers(subject, headers, payload).await,
111            None => client.publish(subject, payload).await,
112        };
113        result.map_err(|err| NatsError::Publish(Box::new(err)))
114    }
115}
ruststream_nats/publisher.rs

ruststream_nats/
publisher.rs
