use std::collections::BTreeMap;
use std::num::NonZeroU32;
use schemars::{JsonSchema, generate::SchemaSettings};
use serde::Serialize;
use serde_json::Value;
use tracing::warn;
mod bindings;
mod viewer;
pub use bindings::{Binding, BindingError, Bindings, PublishBindings, SubscriptionBindings};
pub use viewer::{ViewerOptions, render_viewer_html};
use crate::describe::{AppId, Contact, ExternalDocs, License, Tag};
use crate::runtime::{App, OutgoingKind};
const ASYNCAPI_VERSION: &str = "3.1.0";
#[derive(Debug, Clone, Serialize)]
#[non_exhaustive]
pub struct Spec {
pub asyncapi: String,
#[serde(skip_serializing_if = "Option::is_none")]
pub id: Option<AppId>,
pub info: Info,
#[serde(skip_serializing_if = "BTreeMap::is_empty")]
pub servers: BTreeMap<String, Server>,
#[serde(rename = "defaultContentType", skip_serializing_if = "Option::is_none")]
pub default_content_type: Option<String>,
#[serde(skip_serializing_if = "BTreeMap::is_empty")]
pub channels: BTreeMap<String, Channel>,
#[serde(skip_serializing_if = "BTreeMap::is_empty")]
pub operations: BTreeMap<String, Operation>,
pub components: Components,
}
impl Spec {
pub fn to_json(&self) -> Result<String, serde_json::Error> {
serde_json::to_string_pretty(self)
}
#[must_use]
pub fn messages_without_schema(&self) -> Vec<&str> {
self.components
.messages
.values()
.filter(|message| {
message.payload.is_none() && !message.schemaless && message.name != "bytes"
})
.map(|message| message.name.as_str())
.collect()
}
pub fn to_yaml(&self) -> Result<String, serde_norway::Error> {
serde_norway::to_string(self)
}
}
#[derive(Debug, Clone, Serialize)]
#[non_exhaustive]
pub struct Server {
#[serde(skip_serializing_if = "Option::is_none")]
pub host: Option<String>,
pub protocol: String,
#[serde(rename = "protocolVersion", skip_serializing_if = "Option::is_none")]
pub protocol_version: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
pub description: Option<String>,
#[serde(skip_serializing_if = "Vec::is_empty")]
pub security: Vec<Reference>,
#[serde(skip_serializing_if = "Bindings::is_empty")]
pub bindings: Bindings,
}
#[derive(Debug, Clone, Serialize)]
#[non_exhaustive]
pub struct Info {
pub title: String,
pub version: String,
#[serde(skip_serializing_if = "Option::is_none")]
pub description: Option<String>,
#[serde(rename = "termsOfService", skip_serializing_if = "Option::is_none")]
pub terms_of_service: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
pub contact: Option<Contact>,
#[serde(skip_serializing_if = "Option::is_none")]
pub license: Option<License>,
#[serde(skip_serializing_if = "Vec::is_empty")]
pub tags: Vec<Tag>,
#[serde(rename = "externalDocs", skip_serializing_if = "Option::is_none")]
pub external_docs: Option<ExternalDocs>,
}
#[derive(Debug, Clone, Serialize)]
#[non_exhaustive]
pub struct Channel {
pub address: Option<String>,
pub messages: BTreeMap<String, Reference>,
#[serde(skip_serializing_if = "BTreeMap::is_empty")]
pub parameters: BTreeMap<String, Parameter>,
#[serde(skip_serializing_if = "Vec::is_empty")]
pub servers: Vec<Reference>,
#[serde(skip_serializing_if = "Bindings::is_empty")]
pub bindings: Bindings,
}
#[derive(Debug, Clone, Default, Serialize)]
#[non_exhaustive]
pub struct Parameter {
#[serde(skip_serializing_if = "Option::is_none")]
pub description: Option<String>,
}
#[derive(Debug, Clone, Serialize)]
#[non_exhaustive]
pub struct Operation {
pub action: String,
pub channel: Reference,
pub messages: Vec<Reference>,
#[serde(skip_serializing_if = "Option::is_none")]
pub description: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
pub reply: Option<OperationReply>,
#[serde(skip_serializing_if = "Vec::is_empty")]
pub tags: Vec<Tag>,
#[serde(rename = "x-ruststream-retry", skip_serializing_if = "Option::is_none")]
pub retry: Option<RetryExtension>,
#[serde(skip_serializing_if = "Bindings::is_empty")]
pub bindings: Bindings,
}
#[derive(Debug, Clone, Serialize)]
#[non_exhaustive]
pub struct OperationReply {
#[serde(skip_serializing_if = "Option::is_none")]
pub address: Option<ReplyAddress>,
pub channel: Reference,
pub messages: Vec<Reference>,
}
#[derive(Debug, Clone, Serialize)]
#[non_exhaustive]
pub struct ReplyAddress {
pub location: String,
}
impl ReplyAddress {
#[must_use]
pub fn new(location: impl Into<String>) -> Self {
Self {
location: location.into(),
}
}
}
#[derive(Debug, Clone, Serialize)]
#[non_exhaustive]
pub struct RetryExtension {
#[serde(rename = "maxAttempts", skip_serializing_if = "Option::is_none")]
pub max_attempts: Option<u32>,
#[serde(rename = "deadLetter", skip_serializing_if = "Option::is_none")]
pub dead_letter: Option<String>,
#[serde(rename = "retryDestination", skip_serializing_if = "Option::is_none")]
pub retry_destination: Option<String>,
}
#[derive(Debug, Clone, Serialize)]
#[non_exhaustive]
pub struct Components {
#[serde(skip_serializing_if = "BTreeMap::is_empty")]
pub messages: BTreeMap<String, MessageObject>,
#[serde(rename = "securitySchemes", skip_serializing_if = "BTreeMap::is_empty")]
pub security_schemes: BTreeMap<String, Value>,
}
#[derive(Debug, Clone, Serialize)]
#[non_exhaustive]
pub struct MessageObject {
pub name: String,
#[serde(skip_serializing_if = "Option::is_none")]
pub title: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
pub description: Option<String>,
#[serde(rename = "contentType", skip_serializing_if = "Option::is_none")]
pub content_type: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
pub payload: Option<Value>,
#[serde(skip_serializing_if = "Option::is_none")]
pub headers: Option<Value>,
#[serde(skip)]
pub schemaless: bool,
#[serde(skip_serializing_if = "Bindings::is_empty")]
pub bindings: Bindings,
}
#[derive(Debug, Clone, Serialize)]
pub struct Reference {
#[serde(rename = "$ref")]
pub reference: String,
}
impl Reference {
fn new(target: impl Into<String>) -> Self {
Self {
reference: target.into(),
}
}
}
#[must_use]
pub fn build_spec<A: App>(app: &A) -> Spec {
let contact = app.info().contact.clone();
let info = Info {
title: app.info().title.clone(),
version: app.info().version.clone(),
description: app.info().description.clone(),
terms_of_service: app.info().terms_of_service.clone(),
contact: (!contact.is_empty()).then_some(contact),
license: app.info().license.clone(),
tags: app.info().tags.clone(),
external_docs: app.info().external_docs.clone(),
};
let (servers, security_schemes) = build_servers(app);
let mut channels = BTreeMap::new();
let mut operations = BTreeMap::new();
let mut messages = BTreeMap::new();
for handler in app.handlers() {
let server = resolve_server(handler.server.as_deref(), &servers);
let on_servers = server
.map(|name| vec![Reference::new(format!("#/servers/{name}"))])
.unwrap_or_default();
let tags = server
.and_then(|name| servers.get(name))
.map(|server| vec![Tag::new(server.protocol.clone())])
.unwrap_or_default();
let reply = handler
.outgoing
.iter()
.find(|outgoing| outgoing.kind == OutgoingKind::Answer)
.map(|outgoing| {
let refs = add_outgoing(outgoing, &on_servers, &mut channels, &mut messages);
OperationReply {
address: outgoing.reply_address_location.map(ReplyAddress::new),
channel: refs.channel_ref(),
messages: vec![refs.message_ref()],
}
});
add_receive(
handler,
reply,
&on_servers,
&tags,
&mut channels,
&mut operations,
&mut messages,
);
for outgoing in &handler.outgoing {
if outgoing.kind == OutgoingKind::Answer {
continue;
}
let refs = add_outgoing(outgoing, &on_servers, &mut channels, &mut messages);
operations.insert(
send_operation_id(&operations, handler.name.as_ref(), &refs.channel),
Operation {
action: "send".to_owned(),
channel: refs.channel_ref(),
messages: vec![refs.message_ref()],
description: None,
reply: None,
tags: tags.clone(),
retry: None,
bindings: outgoing.bindings.operation.clone(),
},
);
}
}
let default_content_type = agreed_content_type(&messages);
Spec {
asyncapi: ASYNCAPI_VERSION.to_owned(),
id: app.info().id.clone(),
info,
servers,
default_content_type,
channels,
operations,
components: Components {
messages,
security_schemes,
},
}
}
fn build_servers<A: App>(app: &A) -> (BTreeMap<String, Server>, BTreeMap<String, Value>) {
let mut security_schemes = BTreeMap::new();
let servers = app
.servers()
.iter()
.map(|(name, spec)| {
let security = spec
.security
.iter()
.enumerate()
.map(|(index, scheme)| {
let key = if index == 0 {
name.clone()
} else {
format!("{name}-{index}")
};
security_schemes.insert(key.clone(), security_scheme_object(scheme));
Reference::new(format!("#/components/securitySchemes/{key}"))
})
.collect();
(
name.clone(),
Server {
host: spec.host.clone(),
protocol: spec.protocol.clone(),
protocol_version: spec.protocol_version.clone(),
description: spec.description.clone(),
security,
bindings: spec.bindings.clone(),
},
)
})
.collect();
(servers, security_schemes)
}
fn resolve_server<'a>(
label: Option<&'a str>,
servers: &'a BTreeMap<String, Server>,
) -> Option<&'a str> {
match label {
Some(label) if servers.contains_key(label) => Some(label),
_ if servers.len() == 1 => servers.keys().next().map(String::as_str),
_ => None,
}
}
fn agreed_content_type(messages: &BTreeMap<String, MessageObject>) -> Option<String> {
let mut agreed: Option<&str> = None;
for message in messages.values() {
let content_type = message.content_type.as_deref()?;
match agreed {
None => agreed = Some(content_type),
Some(existing) if existing == content_type => {}
Some(_) => return None,
}
}
agreed.map(str::to_owned)
}
fn security_scheme_object(scheme: &crate::SecurityScheme) -> Value {
use crate::capability::SecuritySchemeKind as Kind;
let parse = |raw: &str| serde_json::from_str::<Value>(raw).unwrap_or(Value::Null);
let mut object = match &scheme.kind {
Kind::UserPassword => serde_json::json!({ "type": "userPassword" }),
Kind::ApiKey { location } => {
serde_json::json!({ "type": "apiKey", "in": location.as_api() })
}
Kind::X509 => serde_json::json!({ "type": "X509" }),
Kind::Plain => serde_json::json!({ "type": "plain" }),
Kind::ScramSha256 => serde_json::json!({ "type": "scramSha256" }),
Kind::ScramSha512 => serde_json::json!({ "type": "scramSha512" }),
Kind::Gssapi => serde_json::json!({ "type": "gssapi" }),
Kind::Http { scheme } => serde_json::json!({ "type": "http", "scheme": scheme }),
Kind::HttpApiKey { name, location } => serde_json::json!({
"type": "httpApiKey",
"name": name,
"in": location.as_api(),
}),
Kind::OpenIdConnect { url } => serde_json::json!({
"type": "openIdConnect",
"openIdConnectUrl": url,
}),
Kind::Oauth2 { flows } => serde_json::json!({ "type": "oauth2", "flows": parse(flows) }),
Kind::Custom { object } => parse(object),
};
if let (Some(description), Some(fields)) = (&scheme.description, object.as_object_mut()) {
fields.insert("description".to_owned(), Value::String(description.clone()));
}
object
}
pub(crate) fn schema_json<T: JsonSchema + ?Sized>() -> Option<String> {
let schema = SchemaSettings::draft07()
.into_generator()
.into_root_schema_for::<T>();
serde_json::to_string(&schema).ok()
}
fn message_name(type_name: &str) -> String {
type_name
.rsplit("::")
.next()
.unwrap_or(type_name)
.to_owned()
}
fn add_receive(
handler: &crate::runtime::HandlerMetadata,
reply: Option<OperationReply>,
on_servers: &[Reference],
tags: &[Tag],
channels: &mut BTreeMap<String, Channel>,
operations: &mut BTreeMap<String, Operation>,
messages: &mut BTreeMap<String, MessageObject>,
) {
let name = handler.name.as_ref();
let payload = handler
.payload_schema
.as_deref()
.and_then(|json| serde_json::from_str::<Value>(json).ok());
if payload.is_none() && !handler.deserialized && handler.input_type != "bytes" {
warn!(
target: "ruststream::asyncapi",
subscription = %name,
message_type = handler.input_type,
"message type has no JSON Schema; its AsyncAPI message carries no payload schema - \
derive schemars::JsonSchema on it",
);
}
let headers = handler
.headers_schema
.as_deref()
.and_then(|json| serde_json::from_str::<Value>(json).ok());
let schema_str = |key: &str| {
payload
.as_ref()
.and_then(|schema| schema.get(key))
.and_then(Value::as_str)
.map(str::to_owned)
};
let message_name = handler
.message_name
.as_ref()
.map(ToString::to_string)
.or_else(|| schema_str("title"))
.unwrap_or_else(|| message_name(handler.input_type));
let channel = channels.entry(name.to_owned()).or_insert_with(|| Channel {
address: Some(name.to_owned()),
messages: BTreeMap::new(),
parameters: BTreeMap::new(),
servers: on_servers.to_vec(),
bindings: Bindings::new(),
});
channel.address = Some(name.to_owned());
channel.bindings.fill_from(&handler.bindings.channel);
channel.messages.insert(
message_name.clone(),
Reference::new(format!("#/components/messages/{message_name}")),
);
operations.insert(
receive_operation_id(operations, name),
Operation {
action: "receive".to_owned(),
channel: Reference::new(format!("#/channels/{name}")),
messages: vec![Reference::new(format!(
"#/channels/{name}/messages/{message_name}"
))],
description: handler.description.as_ref().map(ToString::to_string),
reply,
tags: tags.to_vec(),
retry: retry_extension(handler),
bindings: handler.bindings.operation.clone(),
},
);
let message_description = handler
.message_description
.as_ref()
.map(ToString::to_string)
.or_else(|| schema_str("description"))
.or_else(|| handler.description.as_ref().map(ToString::to_string));
merge_message(
messages,
MessageContribution {
name: message_name,
title: schema_str("title"),
description: message_description,
content_type: handler.content_type.map(str::to_owned),
payload,
headers,
schemaless: handler.deserialized,
bindings: handler.bindings.message.clone(),
},
name,
);
}
fn retry_extension(handler: &crate::runtime::HandlerMetadata) -> Option<RetryExtension> {
let retry_destination = handler
.outgoing
.iter()
.find(|outgoing| outgoing.kind == OutgoingKind::RetryCopy)
.map(|outgoing| outgoing.channel.as_ref().to_owned());
let declaration = &handler.retry;
if declaration.declares_nothing() && retry_destination.is_none() {
return None;
}
Some(RetryExtension {
max_attempts: declaration.max_attempts().map(NonZeroU32::get),
dead_letter: declaration.dead_letter().map(str::to_owned),
retry_destination,
})
}
struct MessageContribution {
name: String,
title: Option<String>,
description: Option<String>,
content_type: Option<String>,
payload: Option<Value>,
headers: Option<Value>,
schemaless: bool,
bindings: Bindings,
}
fn merge_message(
messages: &mut BTreeMap<String, MessageObject>,
contribution: MessageContribution,
context: &str,
) {
let MessageContribution {
name,
title,
description,
content_type,
payload,
headers,
schemaless,
bindings,
} = contribution;
let entry = messages
.entry(name.clone())
.or_insert_with(|| MessageObject {
name,
title: None,
description: None,
content_type: None,
payload: None,
headers: None,
schemaless: false,
bindings: Bindings::new(),
});
entry.bindings.fill_from(&bindings);
if entry.title.is_none() && title.as_deref().is_some_and(|title| title != entry.name) {
entry.title = title;
}
if entry.description.is_none() {
entry.description = description;
}
if entry.content_type.is_none() {
entry.content_type = content_type;
}
entry.schemaless |= schemaless;
if entry.payload.is_none() {
entry.payload = payload;
}
match (&entry.headers, headers) {
(None, Some(headers)) => entry.headers = Some(headers),
(Some(existing), Some(headers)) if *existing != headers => {
warn!(
target: "ruststream::asyncapi",
message = %entry.name,
context = %context,
"conflicting headers schemas for one message component; keeping the first",
);
}
_ => {}
}
}
struct OutgoingRefs {
channel: String,
message: String,
}
impl OutgoingRefs {
fn channel_ref(&self) -> Reference {
Reference::new(format!("#/channels/{}", self.channel))
}
fn message_ref(&self) -> Reference {
Reference::new(format!(
"#/channels/{}/messages/{}",
self.channel, self.message
))
}
}
fn add_outgoing(
outgoing: &crate::runtime::OutgoingMessageMetadata,
on_servers: &[Reference],
channels: &mut BTreeMap<String, Channel>,
messages: &mut BTreeMap<String, MessageObject>,
) -> OutgoingRefs {
let payload = outgoing
.payload_schema
.as_deref()
.and_then(|json| serde_json::from_str::<Value>(json).ok());
if payload.is_none() && !outgoing.serialized && outgoing.message_type != "bytes" {
warn!(
target: "ruststream::asyncapi",
channel = %outgoing.channel,
message_type = outgoing.message_type,
"outgoing message type has no JSON Schema; its AsyncAPI message carries no payload \
schema - derive schemars::JsonSchema on it",
);
}
let headers = outgoing
.headers_schema
.as_deref()
.and_then(|json| serde_json::from_str::<Value>(json).ok());
let title = payload
.as_ref()
.and_then(|schema| schema.get("title"))
.and_then(Value::as_str)
.map(str::to_owned);
let name = outgoing
.message_name
.map(str::to_owned)
.or_else(|| title.clone())
.unwrap_or_else(|| message_name(outgoing.message_type));
let description = outgoing.message_description.map(str::to_owned).or_else(|| {
payload
.as_ref()
.and_then(|schema| schema.get("description"))
.and_then(Value::as_str)
.map(str::to_owned)
});
let channel = outgoing.channel.as_ref();
let entry = channels
.entry(channel.to_owned())
.or_insert_with(|| Channel {
address: outgoing
.reply_address_location
.is_none()
.then(|| channel.to_owned()),
messages: BTreeMap::new(),
parameters: outgoing
.parameters
.iter()
.map(|segment| ((*segment).to_owned(), Parameter::default()))
.collect(),
servers: on_servers.to_vec(),
bindings: Bindings::new(),
});
entry.bindings.fill_from(&outgoing.bindings.channel);
entry.messages.insert(
name.clone(),
Reference::new(format!("#/components/messages/{name}")),
);
merge_message(
messages,
MessageContribution {
name: name.clone(),
title,
description,
content_type: outgoing.content_type.map(str::to_owned),
payload,
headers,
schemaless: outgoing.serialized,
bindings: outgoing.bindings.message.clone(),
},
channel,
);
OutgoingRefs {
channel: channel.to_owned(),
message: name,
}
}
fn receive_operation_id(operations: &BTreeMap<String, Operation>, name: &str) -> String {
let base = format!("receive_{}", sanitize_id(name));
if !operations.contains_key(&base) {
return base;
}
let mut suffix = 2;
loop {
let candidate = format!("{base}_{suffix}");
if !operations.contains_key(&candidate) {
return candidate;
}
suffix += 1;
}
}
fn send_operation_id(
operations: &BTreeMap<String, Operation>,
name: &str,
channel: &str,
) -> String {
let base = format!("send_{}_{}", sanitize_id(name), sanitize_id(channel));
if !operations.contains_key(&base) {
return base;
}
let mut suffix = 2;
loop {
let candidate = format!("{base}_{suffix}");
if !operations.contains_key(&candidate) {
return candidate;
}
suffix += 1;
}
}
fn sanitize_id(name: &str) -> String {
name.chars()
.map(|c| if c.is_alphanumeric() { c } else { '_' })
.collect()
}
#[cfg(test)]
mod tests {
use crate::{ApiKeyLocation, HttpApiKeyLocation, SecurityScheme};
use super::security_scheme_object;
#[test]
fn every_scheme_kind_renders_its_document_object() {
let cases = [
(
SecurityScheme::user_password(),
serde_json::json!({ "type": "userPassword" }),
),
(
SecurityScheme::api_key(ApiKeyLocation::Password),
serde_json::json!({ "type": "apiKey", "in": "password" }),
),
(
SecurityScheme::x509(),
serde_json::json!({ "type": "X509" }),
),
(
SecurityScheme::plain(),
serde_json::json!({ "type": "plain" }),
),
(
SecurityScheme::scram_sha256(),
serde_json::json!({ "type": "scramSha256" }),
),
(
SecurityScheme::scram_sha512(),
serde_json::json!({ "type": "scramSha512" }),
),
(
SecurityScheme::gssapi(),
serde_json::json!({ "type": "gssapi" }),
),
(
SecurityScheme::http("bearer"),
serde_json::json!({ "type": "http", "scheme": "bearer" }),
),
(
SecurityScheme::http_api_key("X-Api-Key", HttpApiKeyLocation::Header),
serde_json::json!({ "type": "httpApiKey", "name": "X-Api-Key", "in": "header" }),
),
(
SecurityScheme::open_id_connect("https://idp.example.com/.well-known"),
serde_json::json!({
"type": "openIdConnect",
"openIdConnectUrl": "https://idp.example.com/.well-known",
}),
),
(
SecurityScheme::oauth2(serde_json::json!({ "clientCredentials": {} })),
serde_json::json!({ "type": "oauth2", "flows": { "clientCredentials": {} } }),
),
(
SecurityScheme::custom(serde_json::json!({ "type": "symmetricEncryption" })),
serde_json::json!({ "type": "symmetricEncryption" }),
),
];
for (scheme, expected) in cases {
assert_eq!(security_scheme_object(&scheme), expected);
}
}
}