ruststream 0.7.5

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
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149

//! Shared helpers for the integration tests.
//!
//! What lives here is what a suite needs but is never the subject of its assertions: the
//! stand-in message and the connected form the publish-log assertions read from. A suite that
//! tests one of these shapes itself declares its own instead.
//!
//! Each test binary compiles its own copy of this module and uses what it needs, hence the
//! `dead_code` allowances.
//!
//! Registrations are documented by default, so every message type here derives `JsonSchema`:
//! a suite mounting one of them is not asking to be the file that opts documentation out.

use std::time::Duration;

use serde::{Deserialize, Serialize};

/// The message the suites drive when the payload is not what they are asserting on: one field,
/// so a decode either happened or did not.
///
/// The `Outgoing` derive rides along wherever the macros feature is on, which lets a suite
/// publish it through the builder. It declares no name: the subject differs per suite, so the
/// call site keeps naming it with `to(..)`.
#[cfg_attr(feature = "macros", derive(ruststream::Outgoing))]
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, schemars::JsonSchema)]
#[allow(dead_code)]
pub(crate) struct Order {
    pub(crate) id: u32,
}

/// The reply half of a request/reply suite: what a handler answers with when the answer's shape
/// is not what the suite is asserting on.
///
/// The `Outgoing` derive is what makes it a reply at all. It declares no name, because the reply
/// subject differs per suite, so the mount site keeps naming it.
#[cfg_attr(feature = "macros", derive(ruststream::Outgoing))]
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, schemars::JsonSchema)]
#[allow(dead_code)]
pub(crate) struct Receipt {
    pub(crate) id: u32,
}

/// The stand-in message of the suites that forward rather than reply: same role as [`Order`],
/// distinct so a test driving both sides can tell input from output.
#[cfg_attr(feature = "macros", derive(ruststream::Outgoing))]
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, schemars::JsonSchema)]
#[allow(dead_code)]
pub(crate) struct Event {
    pub(crate) id: u64,
}

/// An [`Event`] encoded for the wire.
#[allow(dead_code)]
pub(crate) fn payload(id: u64) -> Vec<u8> {
    serde_json::to_vec(&Event { id }).expect("an event serializes")
}

/// Bytes published as themselves: the named wire a suite injects an arbitrary payload through.
///
/// Publishing is typed, so bytes that are not a model of their own still travel as a declared
/// type; `#[derive(Serialized)]` is what says "these bytes are already the payload", and no
/// codec runs on them. Suites use it for the two payloads that are deliberately not a model:
/// what a raw (`&[u8]` or `Deserialized`) subscriber is meant to receive, and what a decode
/// policy is meant to reject. It declares no name - the subject differs per suite, so the call
/// site keeps naming it with `to(..)`.
#[cfg_attr(
    feature = "macros",
    derive(ruststream::Outgoing, ruststream::Serialized)
)]
#[derive(Debug, Clone, PartialEq, Eq)]
#[allow(dead_code)]
pub(crate) struct Wire(pub(crate) Vec<u8>);

#[allow(dead_code)]
impl Wire {
    /// The wire form of `bytes`, for the call sites that hold a slice or a literal.
    pub(crate) fn of(bytes: impl AsRef<[u8]>) -> Self {
        Self(bytes.as_ref().to_vec())
    }
}

/// The request/response pair of the middleware suites, whose subject is what runs around a
/// handler rather than what the handler carries.
///
/// The request half rides the `Outgoing` derive like [`Order`]: a suite drives these middleware
/// through it, so the request is published, and it declares no name because the subject differs
/// per suite.
#[cfg_attr(feature = "macros", derive(ruststream::Outgoing))]
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, schemars::JsonSchema)]
#[allow(dead_code)]
pub(crate) struct Req {
    pub(crate) n: u32,
}

/// The answer to a [`Req`]. See its docs.
#[cfg_attr(feature = "macros", derive(ruststream::Outgoing))]
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, schemars::JsonSchema)]
#[allow(dead_code)]
pub(crate) struct Resp {
    pub(crate) n: u32,
}

/// The connected form of `broker`, for the assertions that read a broker's publish log.
///
/// A broker's `TestableBroker` surface appears on the connected form: the log belongs to the
/// connection, not to the configuration the app was built from.
#[cfg(feature = "memory")]
#[allow(dead_code)]
pub(crate) async fn connected<Log: ruststream::memory::LogMode>(
    broker: &ruststream::memory::MemoryBroker<Log>,
) -> ruststream::memory::ConnectedMemoryBroker<Log> {
    ruststream::Broker::connect(broker.clone())
        .await
        .expect("memory connect is infallible")
}

/// A fresh memory broker with the two handles a suite drives it through: one to publish input,
/// one to read the publish log. The broker itself is returned to be moved into the app.
///
/// Reading a publish log back means the broker has to keep one, so this is the retaining form;
/// the window is wider than anything these suites publish.
#[cfg(feature = "memory")]
#[allow(dead_code)]
pub(crate) async fn observed_memory() -> (
    ruststream::memory::MemoryBroker<ruststream::memory::Retaining>,
    ruststream::memory::MemoryPublisher,
    ruststream::memory::ConnectedMemoryBroker<ruststream::memory::Retaining>,
) {
    let broker = ruststream::memory::MemoryBroker::retaining(
        ruststream::memory::Retention::Messages(ruststream::nonzero!(64)),
    );
    let ingress = broker.publisher();
    let observer = connected(&broker).await;
    (broker, ingress, observer)
}

/// Asserts that exactly one [`Event`] carrying `id` was published to `name`.
#[cfg(all(feature = "memory", feature = "testing"))]
#[allow(dead_code)]
pub(crate) async fn expect_id<Log: ruststream::memory::LogMode>(
    observer: &ruststream::memory::ConnectedMemoryBroker<Log>,
    name: &str,
    id: u64,
) {
    let seen =
        ruststream::testing::expect_published(observer, name, 1, Duration::from_secs(2)).await;
    assert_eq!(seen.len(), 1, "expected one publish on {name}");
    let event: Event = serde_json::from_slice(seen[0].payload()).expect("decodes");
    assert_eq!(event.id, id);
}