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
//! Serve a service's AsyncAPI document and an interactive viewer over HTTP with axum.
//!
//! ```text
//! cargo run --example asyncapi_http --features macros,memory,asyncapi
//! ```
//!
//! Then open <http://127.0.0.1:8080/> for the viewer, or fetch the raw document:
//!
//! ```text
//! curl http://127.0.0.1:8080/asyncapi.json
//! ```
use axum::Router;
use axum::http::header::CONTENT_TYPE;
use axum::response::{Html, IntoResponse};
use axum::routing::get;
use ruststream::asyncapi::{ViewerOptions, build_spec, render_viewer_html};
use ruststream::memory::MemoryBroker;
use ruststream::runtime::{AppInfo, HandlerResult, RustStream};
use ruststream::schemars::JsonSchema;
use ruststream::{Message, SecurityScheme, ServerSpec, subscriber};
use serde::Deserialize;
// --8<-- [start:payload]
/// An order placed by a customer.
#[derive(Debug, Deserialize, Message, JsonSchema)]
struct Order {
id: u64,
item: String,
}
// --8<-- [end:payload]
#[subscriber("orders")]
async fn handle(order: &Order) -> HandlerResult {
println!("order {} ({})", order.id, order.item);
HandlerResult::Ack
}
// --8<-- [start:server]
fn service() -> RustStream {
// `with_broker_labeled` records the broker under a label that is both its stable identity and
// its AsyncAPI server name, deriving the server entry from the broker's own `DescribeServer`
// spec - here the in-memory broker, which describes itself as an in-process "memory" server
// with no host. A broker without a `DescribeServer` impl is instead declared explicitly with
// `.server(name, spec)` alongside a plain `with_broker`.
RustStream::new(AppInfo::new("orders", "0.1.0"))
// --8<-- [start:security]
// A described external server. Security is the author's statement, not the broker's:
// the same broker is deployed publicly and internally with different authentication,
// so the scheme is attached to the spec at registration and brokers never set it.
.server(
"kafka",
ServerSpec::new("kafka.example.com:9093", "kafka")
.with_security(SecurityScheme::scram_sha512().with_description("SASL over TLS")),
)
// --8<-- [end:security]
.with_broker_labeled("in-process", MemoryBroker::new(), |b| b.include(handle))
}
// --8<-- [end:server]
// --8<-- [start:generate]
/// Builds the AsyncAPI document and the viewer HTML from the service.
fn document() -> Result<(String, String), serde_json::Error> {
let spec = build_spec(&service()).to_json()?;
let viewer = render_viewer_html("/asyncapi.json", &ViewerOptions::default());
Ok((spec, viewer))
}
// --8<-- [end:generate]
#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
let (spec, viewer) = document()?;
let router = Router::new()
.route(
"/",
get(move || {
let viewer = viewer.clone();
async move { Html(viewer) }
}),
)
.route(
"/asyncapi.json",
get(move || {
let spec = spec.clone();
async move { ([(CONTENT_TYPE, "application/json")], spec).into_response() }
}),
);
let listener = tokio::net::TcpListener::bind("127.0.0.1:8080").await?;
println!("AsyncAPI viewer on http://127.0.0.1:8080/");
axum::serve(listener, router).await?;
Ok(())
}