ruststream 0.6.1

Async messaging framework for Rust: broker-agnostic traits, router, codecs, and a conformance harness for broker authors.
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
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129

//! The publish-policy seam: a policy is pure declaration, pairing against a connected broker is
//! the only way to a live publisher, and pairing is functorial over the typed combinator stack.
#![cfg(all(
    feature = "memory",
    feature = "macros",
    feature = "json",
    feature = "testing"
))]

use std::time::Duration;

use ruststream::memory::{MemoryBroker, MemoryPublish, MemoryPublisher, MemoryRequest};
use ruststream::runtime::{
    AppInfo, Outgoing, PublishContext, PublishTransform, RustStream, TypedPublisher,
};
use ruststream::testing::expect_published;
use ruststream::{Broker, OutgoingMessage, PublishPolicy, Publisher, RequestReply, subscriber};
use serde::{Deserialize, Serialize};

#[derive(Debug, Serialize, Deserialize)]
struct Order {
    id: u32,
}

#[derive(Debug, Serialize, Deserialize)]
struct Receipt {
    id: u32,
}

/// Stamps every outgoing reply, so the test can prove the transform stack survived pairing.
struct Envelope;

impl<C> PublishTransform<C> for Envelope {
    fn apply(&self, out: &mut Outgoing<'_>, _cx: &PublishContext<'_, C>) {
        out.headers_mut().insert("x-envelope", b"1".to_vec());
    }
}

#[tokio::test]
async fn a_bare_policy_pairs_into_a_live_publisher() {
    let connected = MemoryBroker::new()
        .connect()
        .await
        .expect("memory connect is infallible");
    let publisher = MemoryPublish
        .pair(&connected)
        .await
        .expect("memory pairing is infallible");

    publisher
        .publish(OutgoingMessage::new("policy.out", b"paired".as_slice()))
        .await
        .expect("publish through the paired publisher");

    let seen = expect_published(&connected, "policy.out", 1, Duration::from_secs(2)).await;
    assert_eq!(seen.len(), 1);
    assert_eq!(seen[0].payload(), b"paired");
}

#[tokio::test]
async fn a_request_policy_pairs_into_a_requester() {
    let connected = MemoryBroker::new()
        .connect()
        .await
        .expect("memory connect is infallible");
    let requester = MemoryRequest
        .pair(&connected)
        .await
        .expect("memory pairing is infallible");
    // No responder is subscribed; the requester must fail fast on timeout, proving it is live
    // and bound to this broker's bus.
    let unanswered = RequestReply::request(
        &requester,
        OutgoingMessage::new("policy.void", b"ping".as_slice()),
        Duration::from_millis(50),
    )
    .await;
    assert!(unanswered.is_err(), "{unanswered:?}");
}

#[subscriber("policy.requests", publish("policy.replies"))]
async fn respond(order: &Order) -> Receipt {
    Receipt { id: order.id }
}

/// Pairing a typed stack keeps the codec and the transform: the paired form is the same wiring
/// type over the live leaf, accepted by the reply mounts as before.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn a_typed_policy_stack_pairs_functorially() {
    let broker = MemoryBroker::new();
    let publisher = broker.publisher();

    // The stack itself is a policy: pairing it manually against a connected clone yields the
    // same wiring type over the live leaf (the functorial half of the seam)...
    let connected = Broker::connect(broker.clone())
        .await
        .expect("memory connect is infallible");
    let paired = TypedPublisher::new(MemoryPublish)
        .transform(Envelope)
        .pair(&connected)
        .await
        .expect("memory pairing is infallible");
    let _type_check: TypedPublisher<MemoryPublisher, _, _> = paired;

    // ...while the registration takes the unpaired stack and the runtime pairs it at startup.
    let replies = TypedPublisher::new(MemoryPublish).transform(Envelope);
    let app = RustStream::new(AppInfo::new("policy", "0.1.0")).with_broker(broker, |b| {
        b.include(respond).publisher(replies);
    });
    let running = app.start().await.expect("startup failed");

    publisher
        .publish(OutgoingMessage::new(
            "policy.requests",
            serde_json::to_vec(&Order { id: 7 }).unwrap().as_slice(),
        ))
        .await
        .expect("publish request");

    let seen = expect_published(&connected, "policy.replies", 1, Duration::from_secs(2)).await;
    assert_eq!(seen.len(), 1, "the reply must be published");
    assert_eq!(
        seen[0].headers().get("x-envelope"),
        Some(b"1".as_slice()),
        "the transform stack must survive pairing",
    );

    running.shutdown().await.expect("graceful shutdown failed");
}