ruststream_nats/
request_reply.rs

1//! [`RequestReply`] capability for the NATS publisher.
2
3// Without the `testing` feature a connection link has one variant; see `broker.rs`.
4#![cfg_attr(
5    not(feature = "testing"),
6    allow(clippy::infallible_destructuring_match)
7)]
8
9use async_nats::Request;
10use bytes::BytesMut;
11use ruststream::{OutgoingMessage, RequestReply};
12use std::time::Duration;
13
14use crate::{
15    broker::Link,
16    convert::nats_parts,
17    error::NatsError,
18    message::{CoreMessage, NatsMessage},
19    publisher::{NatsPublisher, client_for},
20};
21
22impl RequestReply for NatsPublisher {
23    type Reply = NatsMessage;
24
25    async fn request(
26        &self,
27        msg: OutgoingMessage<'_, BytesMut>,
28        timeout: Duration,
29    ) -> Result<Self::Reply, Self::Error> {
30        let connection = match self.link() {
31            Link::Nats(connection) => connection,
32            #[cfg(feature = "testing")]
33            Link::InProcess(bus) => {
34                let (subject, payload, headers) = msg.into_parts();
35                let reply = bus
36                    .request(subject, payload.freeze(), &headers, timeout)
37                    .await?;
38                return Ok(NatsMessage::Core(Box::new(CoreMessage::new(reply))));
39            }
40        };
41        let client = client_for(connection, msg.name())?;
42        let (subject, payload, headers_owned) = nats_parts(msg)?;
43
44        let fut = async {
45            let request = match headers_owned {
46                Some(headers) => Request::new().payload(payload).headers(headers),
47                None => Request::new().payload(payload),
48            };
49            client
50                .send_request(subject, request)
51                .await
52                .map_err(|err| NatsError::Publish(Box::new(err)))
53        };
54
55        let response = tokio::time::timeout(timeout, fut)
56            .await
57            .map_err(|_| NatsError::RequestTimeout)??;
58        Ok(NatsMessage::Core(Box::new(CoreMessage::new(response))))
59    }
60}
ruststream_nats/request_reply.rs

ruststream_nats/
request_reply.rs
