ruststream-lapin 0.7.0

RabbitMQ / AMQP 0.9.1 broker implementation for the RustStream messaging framework, backed by lapin.
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86

//! The responder half of the direct reply-to convention, packaged as a publish transform.

use ruststream::Str;
use ruststream::runtime::{ForReply, Names, Outgoing, PublishContext, PublishTransform};

/// Redirects each reply of a `#[subscriber(.., reply(..))]` handler to the requester's
/// private reply-to address, echoing its correlation id.
///
/// This is the canonical responder wiring for [request/reply over `RabbitMQ` direct
/// reply-to](crate::LapinRequester): compose it onto the reply publisher at mount time and the
/// handler stays a pure request-to-reply function. A request without a `reply-to` header falls
/// through to the mount site's `publish("..")` name.
///
/// The transform names the destination per delivery, so it mounts only where nothing has declared
/// one: the reply type derives `Outgoing` without `#[outgoing(name = "..")]` and the mount site
/// supplies the fallback name. A reply type that declares its own destination refuses this
/// transform at the mount site, which is what keeps the generated `AsyncAPI` document and the wire
/// in step.
///
/// It writes none of the per-message AMQP properties, so it is generic over the publisher's
/// options and mounts over any of this crate's publishers. The reply-to address and the
/// correlation id it echoes are protocol fields the crate carries as headers, and the publisher
/// puts them back on the frame.
///
/// # Examples
///
/// ```
/// use ruststream_lapin::prelude::*;
/// use serde::{Deserialize, Serialize};
///
/// #[derive(Deserialize)]
/// struct Ask {
///     sku: String,
/// }
///
/// // An RPC reply goes wherever the request asked, so the destination belongs to the mount
/// // site and the type declares none.
/// #[derive(Serialize, Outgoing)]
/// struct Stock {
///     available: bool,
/// }
///
/// #[subscriber("inventory.check", reply("inventory.check.unrouted"))]
/// async fn check(ask: &Ask) -> Stock {
///     Stock {
///         available: !ask.sku.is_empty(),
///     }
/// }
///
/// fn app() -> RustStream {
///     let broker = LapinBroker::new("amqp://localhost:5672");
///     RustStream::new(AppInfo::new("inventory", "0.1.0")).with_broker(broker, |b| {
///         b.include(check)
///             .out_reply(Publish::default())
///             .transform(DirectReplyTo);
///     })
/// }
/// ```
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
pub struct DirectReplyTo;

// --8<-- [start:transform]
impl<C, Options> PublishTransform<ForReply<C>, Options> for DirectReplyTo {
    type Destination = Names;

    fn apply(
        &self,
        out: &mut Outgoing<'_>,
        _options: &mut Option<Options>,
        cx: &PublishContext<'_, C>,
    ) {
        // Both fields travel as the buffers the delivery already holds: the reply takes a
        // reference count on the request's bytes instead of copying them per message. A reply-to
        // that is not UTF-8 is not an address, and falls through to the mount site's name.
        if let Some(reply_to) = cx.headers().get_shared("reply-to")
            && let Ok(name) = Str::try_from(reply_to)
        {
            out.set_name(name);
        }
        if let Some(correlation_id) = cx.headers().get_shared("correlation-id") {
            out.headers_mut()
                .insert(Str::from_static("correlation-id"), correlation_id);
        }
    }
}
// --8<-- [end:transform]