ruststream_lapin/
reply.rs

1//! The responder half of the direct reply-to convention, packaged as a publish transform.
2
3use ruststream::runtime::{Outgoing, PublishContext, PublishTransform};
4
5/// Redirects each reply of a `#[subscriber(.., publish(..))]` handler to the requester's
6/// private reply-to address, echoing its correlation id.
7///
8/// This is the canonical responder wiring for [request/reply over `RabbitMQ` direct
9/// reply-to](crate::LapinRequester): compose it onto the reply publisher at mount time and the
10/// handler stays a pure request-to-reply function. Requests without a `reply-to` header fall
11/// through to the mount's static destination.
12///
13/// # Examples
14///
15/// ```
16/// use ruststream::runtime::TypedPublisher;
17/// use ruststream_lapin::{DirectReplyTo, LapinBroker};
18///
19/// let broker = LapinBroker::new("amqp://localhost:5672");
20/// let replies = TypedPublisher::new(broker.publisher()).transform(DirectReplyTo);
21/// # let _ = replies;
22/// ```
23#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
24pub struct DirectReplyTo;
25
26// --8<-- [start:transform]
27impl<C> PublishTransform<C> for DirectReplyTo {
28    fn apply(&self, out: &mut Outgoing<'_>, cx: &PublishContext<'_, C>) {
29        if let Some(reply_to) = cx.headers().reply_to() {
30            out.set_name(reply_to.to_owned());
31        }
32        if let Some(correlation_id) = cx.headers().correlation_id() {
33            out.headers_mut()
34                .insert("correlation-id", correlation_id.as_bytes().to_vec());
35        }
36    }
37}
38// --8<-- [end:transform]
ruststream_lapin/reply.rs

ruststream_lapin/
reply.rs
