ruststream_lapin/
reply.rs

1//! The responder half of the direct reply-to convention, packaged as a publish transform.
2
3use ruststream::Str;
4use ruststream::runtime::{ForReply, Names, Outgoing, PublishContext, PublishTransform};
5
6/// Redirects each reply of a `#[subscriber(.., reply(..))]` handler to the requester's
7/// private reply-to address, echoing its correlation id.
8///
9/// This is the canonical responder wiring for [request/reply over `RabbitMQ` direct
10/// reply-to](crate::LapinRequester): compose it onto the reply publisher at mount time and the
11/// handler stays a pure request-to-reply function. A request without a `reply-to` header falls
12/// through to the mount site's `reply("..")` name.
13///
14/// The transform names the destination per delivery, so it mounts only where nothing has declared
15/// one: the reply type derives `Outgoing` without `#[outgoing(name = "..")]` and the mount site
16/// supplies the fallback name. A reply type that declares its own destination refuses this
17/// transform at the mount site, which is what keeps the generated `AsyncAPI` document and the wire
18/// in step.
19///
20/// It writes none of the per-message AMQP properties, so it is generic over the publisher's
21/// options and mounts over any of this crate's publishers. The reply-to address and the
22/// correlation id it echoes are protocol fields the crate carries as headers, and the publisher
23/// puts them back on the frame.
24///
25/// # Examples
26///
27/// ```
28/// use ruststream_lapin::prelude::*;
29/// use serde::{Deserialize, Serialize};
30///
31/// #[derive(Deserialize)]
32/// struct Ask {
33///     sku: String,
34/// }
35///
36/// // An RPC reply goes wherever the request asked, so the destination belongs to the mount
37/// // site and the type declares none.
38/// #[derive(Serialize, Outgoing)]
39/// struct Stock {
40///     available: bool,
41/// }
42///
43/// #[subscriber("inventory.check", reply("inventory.check.unrouted"))]
44/// async fn check(ask: &Ask) -> Stock {
45///     Stock {
46///         available: !ask.sku.is_empty(),
47///     }
48/// }
49///
50/// fn app() -> RustStream {
51///     let broker = LapinBroker::new("amqp://localhost:5672");
52///     RustStream::new(AppInfo::new("inventory", "0.1.0")).with_broker(broker, |b| {
53///         b.include(check)
54///             .out_reply(Publish::default())
55///             .transform(DirectReplyTo);
56///     })
57/// }
58/// ```
59#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
60pub struct DirectReplyTo;
61
62// --8<-- [start:transform]
63impl<C, Options> PublishTransform<ForReply<C>, Options> for DirectReplyTo {
64    type Destination = Names;
65
66    fn apply(
67        &self,
68        out: &mut Outgoing<'_>,
69        _options: &mut Option<Options>,
70        cx: &PublishContext<'_, C>,
71    ) {
72        // Both fields travel as the buffers the delivery already holds: the reply takes a
73        // reference count on the request's bytes instead of copying them per message. A reply-to
74        // that is not UTF-8 is not an address, and falls through to the mount site's name.
75        if let Some(reply_to) = cx.headers().get_shared("reply-to")
76            && let Ok(name) = Str::try_from(reply_to)
77        {
78            out.set_name(name);
79        }
80        if let Some(correlation_id) = cx.headers().get_shared("correlation-id") {
81            out.headers_mut()
82                .insert(Str::from_static("correlation-id"), correlation_id);
83        }
84    }
85}
86// --8<-- [end:transform]
ruststream_lapin/reply.rs

ruststream_lapin/
reply.rs
