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
//! Unit-test a service with the in-process [`TestApp`](ruststream::testing::TestApp) harness.
//!
//! The harness drives a built application with no network `connect` and no server: it injects input
//! through the broker bus, records what each handler received and how it settled, and exposes
//! per-broker assertions. `MemoryBroker` is a real broker here, not a test double - the harness
//! drives it (or any broker) through the same dispatch path the production runtime uses.
//!
//! ```text
//! cargo run --example testing --features testing,macros,memory,json
//! ```
use ruststream::memory::MemoryPublisher;
use ruststream::memory::prelude::*;
use ruststream::testing::TestApp;
use serde::{Deserialize, Serialize};
// --8<-- [start:app]
#[derive(Outgoing, Serialize, Deserialize, PartialEq, Debug)]
struct Order {
id: u64,
}
#[derive(Outgoing, Serialize, Deserialize, PartialEq, Debug)]
struct Receipt {
order_id: u64,
}
/// The application state: a publisher the handler uses to emit receipts.
struct AppState {
receipts: MemoryPublisher,
}
/// Validates an order and publishes a receipt for it.
#[subscriber("orders")]
async fn handle_order(
order: &Order,
ctx: &mut ruststream::runtime::Context<'_, (), AppState>,
) -> HandlerOutcome {
let receipt = Receipt { order_id: order.id };
if ctx
.state()
.receipts
.message(&receipt)
.to("receipts")
.publish()
.await
.is_err()
{
return HandlerOutcome::retry();
}
HandlerOutcome::ack()
}
/// Builds the service: one broker, the order handler, and a receipts publisher in state.
fn service() -> RustStream<ruststream::runtime::Identity, AppState> {
let broker = MemoryBroker::new();
let receipts = broker.publisher();
RustStream::new(AppInfo::new("orders", "0.1.0"))
.on_startup(
move |()| async move { Ok::<_, std::convert::Infallible>(AppState { receipts }) },
)
.with_broker(broker, |b| {
b.include(handle_order);
})
}
// --8<-- [end:app]
// --8<-- [start:test]
#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
let tb = TestApp::start(service()).await?;
// Inject an order; the publish drives the handler to a standstill before returning.
tb.broker::<MemoryBroker>()
.message(&Order { id: 42 })
.to("orders")
.publish()
.await?;
// The handler ran once, decoded the order, and acked.
tb.broker::<MemoryBroker>()
.subscriber("orders")
.assert_called_once()
.with(&Order { id: 42 })
.settled(HandlerOutcome::ack());
// It published the matching receipt downstream.
tb.broker::<MemoryBroker>()
.published::<Receipt>("receipts")
.assert_called_once()
.with(&Receipt { order_id: 42 });
println!("ok: handler validated, acked, and published a receipt");
Ok(())
}
// --8<-- [end:test]