From 7ac32efc7a310f79226e07dab8f820658de4a603 Mon Sep 17 00:00:00 2001 From: Max Christoph Date: Wed, 24 Jun 2026 14:06:53 +0200 Subject: [PATCH 01/12] feat: Prepare Server, Operation and Message bindings. Add mqtt bindings models --- asyncapi-rust-models/src/lib.rs | 139 ++++++++++++++++++++++++++++++++ 1 file changed, 139 insertions(+) diff --git a/asyncapi-rust-models/src/lib.rs b/asyncapi-rust-models/src/lib.rs index a587184..195381d 100644 --- a/asyncapi-rust-models/src/lib.rs +++ b/asyncapi-rust-models/src/lib.rs @@ -184,6 +184,70 @@ pub struct Server { /// A map of variable name to ServerVariable definition for variables used in the pathname #[serde(skip_serializing_if = "Option::is_none")] pub variables: Option>, + + /// Protocol specific bindings. + #[serde(skip_serializing_if = "Option::is_none")] + pub bindings: Option, +} + +/// Protocol specific server bindings +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct ServerBindings { + /// Mqtt protocol specific bindings + #[serde(skip_serializing_if = "Option::is_none")] + pub mqtt: Option, +} + +/// Mqtt last will structure +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct MqttLastWill { + /// The topic where the Last Will and Testament message will be sent. + pub topic: String, + + /// Defines how hard the broker/client will try to ensure that the Last Will and Testament message is received. Its value MUST be either 0, 1 or 2. + pub qos: u8, + + /// Last Will message. + pub message: String, + + /// Whether the broker should retain the Last Will and Testament message or not. + pub retain: bool, +} + +/// Mqtt server binding +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct MqttServerBindings { + /// The client identifier. + #[serde(skip_serializing_if = "Option::is_none", rename = "cliendId")] + pub client_id: Option, + + /// Whether to create a persistent connection or not. When false, + /// the connection will be persistent. This is called clean start in MQTTv5. + #[serde(skip_serializing_if = "Option::is_none", rename = "cleanSession")] + pub clean_session: Option, + + /// Last Will and Testament configuration. topic, qos, message and retain are properties of this object as shown below. + #[serde(skip_serializing_if = "Option::is_none", rename = "lastWill")] + pub last_will: Option, + + /// Interval in seconds of the longest period of time the broker and the client can endure without sending a message. + #[serde(skip_serializing_if = "Option::is_none", rename = "keepAlive")] + pub keep_alive: Option, + + /// Interval in seconds or a Schema Object containing the definition of the interval. The broker maintains a session for a disconnected client until this interval expires. + #[serde( + skip_serializing_if = "Option::is_none", + rename = "sessionExpiryInterval" + )] + pub session_expiry_interval: Option, + + /// Number of bytes or a Schema Object representing the maximum packet size the client is willing to accept. + #[serde(skip_serializing_if = "Option::is_none", rename = "maximumPacketSize")] + pub max_packet_size: Option, + + /// The version of this binding. If omitted, "latest" MUST be assumed. + #[serde(skip_serializing_if = "Option::is_none", rename = "bindingVersion")] + pub binding_version: Option, } /// Server variable definition @@ -446,6 +510,45 @@ pub struct Message { /// JSON Schema defining the structure of the message payload #[serde(skip_serializing_if = "Option::is_none")] pub payload: Option, + + /// Protocol specific bindings. + #[serde(skip_serializing_if = "Option::is_none")] + pub bindings: Option, +} + +/// Protocol specific message bindings +#[derive(Serialize, Deserialize, Clone, Debug)] +pub struct MessageBindings { + /// Mqtt protocol specific message bindings + #[serde(skip_serializing_if = "Option::is_none")] + pub mqtt: Option, +} + +/// Mqtt message bindings +#[derive(Serialize, Deserialize, Clone, Debug)] +pub struct MqttMessageBindings { + /// Either: 0 (zero): Indicates that the payload is unspecified bytes, or 1: Indicates that the payload is UTF-8 encoded character data. + #[serde( + skip_serializing_if = "Option::is_none", + rename = "payloadFormatIndicator" + )] + pub payload_format_indicator: Option, + + /// Correlation Data is used by the sender of the request message to identify which request the response message is for when it is received. + #[serde(skip_serializing_if = "Option::is_none", rename = "correlationData")] + pub correlation_data: Option, + + /// String describing the content type of the message payload. This should not conflict with the contentType field of the associated AsyncAPI Message object. + #[serde(skip_serializing_if = "Option::is_none", rename = "contentType")] + pub content_type: Option, + + /// The topic (channel URI) for a response message. + #[serde(skip_serializing_if = "Option::is_none", rename = "responseTopic")] + pub response_topic: Option, + + /// The version of this binding. If omitted, "latest" MUST be assumed. + #[serde(skip_serializing_if = "Option::is_none", rename = "bindingVersion")] + pub binding_version: Option, } /// Operation (send or receive) @@ -483,6 +586,42 @@ pub struct Operation { /// Optional list of messages that can be used with this operation #[serde(skip_serializing_if = "Option::is_none")] pub messages: Option>, + + /// Protocol specific bindings + #[serde(skip_serializing_if = "Option::is_none")] + pub bindings: Option, +} + +/// Protocol specific operation bindings +#[derive(Serialize, Deserialize, Clone, Debug)] +pub struct OperationBindings { + /// Mqtt specific operation bindings + #[serde(skip_serializing_if = "Option::is_none")] + pub mqtt: Option, +} + +/// Mqtt operation bindings +#[derive(Serialize, Deserialize, Clone, Debug)] +pub struct MqttOperationBindings { + /// Defines the Quality of Service (QoS) levels for the message flow between client and server. + /// Its value MUST be either 0 (At most once delivery), 1 (At least once delivery), or 2 (Exactly once delivery). + #[serde(skip_serializing_if = "Option::is_none")] + pub qos: Option, + + /// Whether the broker should retain the message or not. + #[serde(skip_serializing_if = "Option::is_none")] + pub retain: Option, + + /// Interval in seconds or a Schema Object containing the definition of the lifetime of the message. + #[serde( + skip_serializing_if = "Option::is_none", + rename = "messageExpiryInterval" + )] + pub message_expiry_interval: Option, + + /// The version of this binding. If omitted, "latest" MUST be assumed. + #[serde(skip_serializing_if = "Option::is_none", rename = "bindingVersion")] + pub binding_version: Option, } /// Operation action type From be56c668af0c1ec4219e380523cfdf63e3355b73 Mon Sep 17 00:00:00 2001 From: Max Christoph Date: Wed, 24 Jun 2026 16:53:05 +0200 Subject: [PATCH 02/12] feat: extract server bindings --- .../src/asyncapi_spec_attrs.rs | 151 ++++++++++++++++++ 1 file changed, 151 insertions(+) diff --git a/asyncapi-rust-codegen/src/asyncapi_spec_attrs.rs b/asyncapi-rust-codegen/src/asyncapi_spec_attrs.rs index 978b4ce..2992056 100644 --- a/asyncapi-rust-codegen/src/asyncapi_spec_attrs.rs +++ b/asyncapi-rust-codegen/src/asyncapi_spec_attrs.rs @@ -23,6 +23,26 @@ pub struct ServerMeta { pub pathname: Option, pub description: Option, pub variables: Vec, + pub mqtt: Option +} + +#[derive(Debug, Clone)] +pub struct LastWillMeta { + pub topic: String, + pub qos: u8, + pub retain: bool, + pub message: String, +} + +#[derive(Debug, Clone)] +pub struct MqttServerBindingsMeta { + pub client_id: Option, + pub clean_session: Option, + pub last_will: Option, + pub keep_alive: Option, + pub session_expiry_interval: Option, + pub maximum_packet_size: Option, + pub binding_version: Option } /// Server variable metadata @@ -64,6 +84,15 @@ pub struct OperationMeta { pub channel: String, #[allow(dead_code)] // Reserved for future use pub description: Option, + pub mqtt: Option +} + +#[derive(Debug, Clone)] +pub struct OperationMqttBindingsMeta { + pub qos: Option, + pub retain: Option, + pub message_expiry_interval: Option, + pub binding_version: Option } /// Extract asyncapi spec metadata from `#[asyncapi(...)]` attributes @@ -133,6 +162,7 @@ fn extract_server(attr: &Attribute) -> Option { let mut pathname = None; let mut description = None; let mut variables = Vec::new(); + let mut mqtt = None; let _ = attr.parse_nested_meta(|nested| { if nested.path.is_ident("name") { @@ -160,6 +190,10 @@ fn extract_server(attr: &Attribute) -> Option { if let Some(var) = extract_server_variable(&nested) { variables.push(var); } + } else if nested.path.is_ident("mqtt") { + if let Some(var) = extract_mqtt_server_bindings(&nested) { + mqtt = Some(var); + } } Ok(()) }); @@ -172,6 +206,123 @@ fn extract_server(attr: &Attribute) -> Option { pathname, description, variables, + mqtt + }) +} + +fn extract_mqtt_server_bindings( + nested: &syn::meta::ParseNestedMeta, +) -> Option { + let mut client_id: Option = None; + let mut clean_session: Option = None; + let mut keep_alive = None; + let mut session_expiry_interval = None; + let mut maximum_packet_size = None; + let mut binding_version: Option = None; + + let mut last_will: Option = None; + + let _ = nested.parse_nested_meta(|inner| { + if inner.path.is_ident("client_id") { + let value = inner.value()?; + let s: syn::LitStr = value.parse()?; + client_id = Some(s.value()); + + } else if inner.path.is_ident("clean_session") { + let value = inner.value()?; + let s: syn::LitBool = value.parse()?; + clean_session = Some(s.value()); + + } else if inner.path.is_ident("keep_alive") { + let value = inner.value()?; + let s: syn::LitInt = value.parse()?; + keep_alive = Some(s.base10_parse()?); + + } else if inner.path.is_ident("session_expiry_interval") { + let value = inner.value()?; + let p: Path = value.parse()?; + session_expiry_interval = Some(p); + + } else if inner.path.is_ident("maximum_packet_size") { + let value = inner.value()?; + let p: Path = value.parse()?; + maximum_packet_size = Some(p); + + } else if inner.path.is_ident("binding_version") { + let value = inner.value()?; + let s: syn::LitStr = value.parse()?; + binding_version = Some(s.value()); + + } else if inner.path.is_ident("last_will") { + let mut topic: Option = None; + let mut qos: Option = None; + let mut message: Option = None; + let mut retain: Option = None; + + inner.parse_nested_meta(|lw| { + if lw.path.is_ident("topic") { + let value = lw.value()?; + let s: syn::LitStr = value.parse()?; + topic = Some(s.value()); + + } else if lw.path.is_ident("qos") { + let value = lw.value()?; + let s: syn::LitInt = value.parse()?; + qos = Some(s.base10_parse()?); + + } else if lw.path.is_ident("message") { + let value = lw.value()?; + let s: syn::LitStr = value.parse()?; + message = Some(s.value()); + + } else if lw.path.is_ident("retain") { + let value = lw.value()?; + let s: syn::LitBool = value.parse()?; + retain = Some(s.value()); + } + + Ok(()) + })?; + + last_will = Some(LastWillMeta { + topic: topic.ok_or_else(|| { + syn::Error::new_spanned( + &inner.path, + "missing required field `topic` in last_will", + ) + })?, + qos: qos.ok_or_else(|| { + syn::Error::new_spanned( + &inner.path, + "missing required field `qos` in last_will", + ) + })?, + message: message.ok_or_else(|| { + syn::Error::new_spanned( + &inner.path, + "missing required field `message` in last_will", + ) + })?, + retain: retain.ok_or_else(|| { + syn::Error::new_spanned( + &inner.path, + "missing required field `retain` in last_will", + ) + })?, + }); + } + + Ok(()) + }); + + Some(MqttServerBindingsMeta { + client_id, + clean_session, + binding_version, + last_will, + keep_alive, + maximum_packet_size, + session_expiry_interval }) } From ef5e427228c2f0da95f6fa242cbd05a3f58f1109 Mon Sep 17 00:00:00 2001 From: Max Christoph Date: Thu, 25 Jun 2026 09:14:16 +0200 Subject: [PATCH 03/12] Serverbindings --- asyncapi-rust-codegen/src/asyncapi_attrs.rs | 79 ++++++++++++++++++- .../src/asyncapi_spec_attrs.rs | 74 ++++++++++++----- asyncapi-rust-codegen/src/lib.rs | 75 ++++++++++++++++++ asyncapi-rust/examples/chat_api.rs | 3 + 4 files changed, 209 insertions(+), 22 deletions(-) diff --git a/asyncapi-rust-codegen/src/asyncapi_attrs.rs b/asyncapi-rust-codegen/src/asyncapi_attrs.rs index 635ee0b..395ad01 100644 --- a/asyncapi-rust-codegen/src/asyncapi_attrs.rs +++ b/asyncapi-rust-codegen/src/asyncapi_attrs.rs @@ -1,6 +1,27 @@ //! Utilities for parsing asyncapi attributes -use syn::Attribute; +use syn::{Attribute, Meta, Path}; + +#[derive(Clone, Debug)] +pub enum ResponseTopic { + Reference(Path), + Uri(String), +} + +impl Default for ResponseTopic { + fn default() -> Self { + Self::Uri(String::default()) + } +} + +#[derive(Clone, Debug, Default)] +pub struct MqttMessageBinding { + pub payload_format_indicator: Option, + pub correlation_data: Option, + pub content_type: Option, + pub response_topic: Option, + pub binding_version: Option, +} /// AsyncAPI metadata extracted from attributes #[derive(Debug, Default, Clone)] @@ -13,6 +34,27 @@ pub struct AsyncApiMeta { /// Override the message name used in `components.messages` and `asyncapi_message_names()`. /// When absent the Rust variant/type identifier is used. pub message_name: Option, + pub mqtt: Option, +} + +fn parse_response_topic(meta: Meta) -> Option { + match meta { + Meta::Path(path) => Some(ResponseTopic::Reference(path)), + + Meta::NameValue(nv) => { + match nv.value { + syn::Expr::Lit(expr_lit) => { + match expr_lit.lit { + syn::Lit::Str(s) => Some(ResponseTopic::Uri(s.value())), + _ => None, // invalid literal → ignore + } + } + _ => None, + } + } + + _ => None, + } } /// Extract asyncapi metadata from `#[asyncapi(...)]` attributes @@ -48,6 +90,41 @@ pub fn extract_asyncapi_meta(attrs: &[Attribute]) -> AsyncApiMeta { let value = nested.value()?; let s: syn::LitStr = value.parse()?; meta.message_name = Some(s.value()); + } else if nested.path.is_ident("mqtt") { + let mut binding = MqttMessageBinding::default(); + + let value = nested.value()?; + let mqtt_meta: syn::MetaList = value.parse()?; + + mqtt_meta.parse_nested_meta(|m| { + if m.path.is_ident("payload_format_indicator") { + let v = m.value()?; + let lit: syn::LitInt = v.parse()?; + binding.payload_format_indicator = Some(lit.base10_parse::()?); + } else if m.path.is_ident("correlation_data") { + let v = m.value()?; + let s: syn::Path = v.parse()?; + + binding.correlation_data = Some(s); + } else if m.path.is_ident("content_type") { + let v = m.value()?; + let s: syn::LitStr = v.parse()?; + binding.content_type = Some(s.value()); + } else if m.path.is_ident("response_topic") { + let v = m.value()?; + let meta_inner: syn::Meta = v.parse()?; + + binding.response_topic = parse_response_topic(meta_inner); + } else if m.path.is_ident("binding_version") { + let v = m.value()?; + let s: syn::LitStr = v.parse()?; + binding.binding_version = Some(s.value()); + } + + Ok(()) + })?; + + meta.mqtt = Some(binding); } Ok(()) }); diff --git a/asyncapi-rust-codegen/src/asyncapi_spec_attrs.rs b/asyncapi-rust-codegen/src/asyncapi_spec_attrs.rs index 2992056..bd18351 100644 --- a/asyncapi-rust-codegen/src/asyncapi_spec_attrs.rs +++ b/asyncapi-rust-codegen/src/asyncapi_spec_attrs.rs @@ -1,6 +1,6 @@ //! Utilities for parsing asyncapi spec-level attributes -use syn::{Attribute, Path}; +use syn::{Attribute, LitStr, Path}; /// AsyncAPI spec metadata extracted from attributes #[derive(Debug, Default, Clone)] @@ -23,7 +23,7 @@ pub struct ServerMeta { pub pathname: Option, pub description: Option, pub variables: Vec, - pub mqtt: Option + pub mqtt: Option, } #[derive(Debug, Clone)] @@ -40,9 +40,9 @@ pub struct MqttServerBindingsMeta { pub clean_session: Option, pub last_will: Option, pub keep_alive: Option, - pub session_expiry_interval: Option, + pub session_expiry_interval: Option, pub maximum_packet_size: Option, - pub binding_version: Option + pub binding_version: Option, } /// Server variable metadata @@ -84,15 +84,15 @@ pub struct OperationMeta { pub channel: String, #[allow(dead_code)] // Reserved for future use pub description: Option, - pub mqtt: Option + pub mqtt: Option, } #[derive(Debug, Clone)] pub struct OperationMqttBindingsMeta { pub qos: Option, pub retain: Option, - pub message_expiry_interval: Option, - pub binding_version: Option + pub message_expiry_interval: Option, + pub binding_version: Option, } /// Extract asyncapi spec metadata from `#[asyncapi(...)]` attributes @@ -206,7 +206,7 @@ fn extract_server(attr: &Attribute) -> Option { pathname, description, variables, - mqtt + mqtt, }) } @@ -227,32 +227,26 @@ fn extract_mqtt_server_bindings( let value = inner.value()?; let s: syn::LitStr = value.parse()?; client_id = Some(s.value()); - } else if inner.path.is_ident("clean_session") { let value = inner.value()?; let s: syn::LitBool = value.parse()?; clean_session = Some(s.value()); - } else if inner.path.is_ident("keep_alive") { let value = inner.value()?; let s: syn::LitInt = value.parse()?; keep_alive = Some(s.base10_parse()?); - } else if inner.path.is_ident("session_expiry_interval") { let value = inner.value()?; - let p: Path = value.parse()?; - session_expiry_interval = Some(p); - + let p: syn::LitInt = value.parse()?; + session_expiry_interval = Some(p.base10_parse()?); } else if inner.path.is_ident("maximum_packet_size") { let value = inner.value()?; let p: Path = value.parse()?; maximum_packet_size = Some(p); - } else if inner.path.is_ident("binding_version") { let value = inner.value()?; let s: syn::LitStr = value.parse()?; binding_version = Some(s.value()); - } else if inner.path.is_ident("last_will") { let mut topic: Option = None; let mut qos: Option = None; @@ -264,17 +258,14 @@ fn extract_mqtt_server_bindings( let value = lw.value()?; let s: syn::LitStr = value.parse()?; topic = Some(s.value()); - } else if lw.path.is_ident("qos") { let value = lw.value()?; let s: syn::LitInt = value.parse()?; qos = Some(s.base10_parse()?); - } else if lw.path.is_ident("message") { let value = lw.value()?; let s: syn::LitStr = value.parse()?; message = Some(s.value()); - } else if lw.path.is_ident("retain") { let value = lw.value()?; let s: syn::LitBool = value.parse()?; @@ -312,7 +303,7 @@ fn extract_mqtt_server_bindings( }); } - Ok(()) + Ok(()) }); Some(MqttServerBindingsMeta { @@ -322,7 +313,7 @@ fn extract_mqtt_server_bindings( last_will, keep_alive, maximum_packet_size, - session_expiry_interval + session_expiry_interval, }) } @@ -468,12 +459,50 @@ fn extract_channel_parameter(nested: &syn::meta::ParseNestedMeta) -> Option Option { + let mut qos = None; + let mut retain = None; + let mut message_expiry_interval = None; + let mut binding_version = None; + + let _ = nested.parse_nested_meta(|inner| { + if inner.path.is_ident("qos") { + let value = inner.value()?; + let s: syn::LitInt = value.parse()?; + qos = Some(s.base10_parse()?); + } else if inner.path.is_ident("retain") { + let value = inner.value()?; + let s: syn::LitBool = value.parse()?; + retain = Some(s.value()); + } else if inner.path.is_ident("message_expiry_interval") { + let value = inner.value()?; + let p: syn::LitInt = value.parse()?; + message_expiry_interval = Some(p.base10_parse()?); + } else if inner.path.is_ident("binding_version") { + let value = inner.value()?; + let s: LitStr = value.parse()?; + binding_version = Some(s.value()); + } + Ok(()) + }); + + Some(OperationMqttBindingsMeta { + qos, + retain, + message_expiry_interval, + binding_version, + }) +} + /// Extract operation metadata from `#[asyncapi_operation(...)]` attribute fn extract_operation(attr: &Attribute) -> Option { let mut name = None; let mut action = None; let mut channel = None; let mut description = None; + let mut mqtt = None; let _ = attr.parse_nested_meta(|nested| { if nested.path.is_ident("name") { @@ -492,6 +521,8 @@ fn extract_operation(attr: &Attribute) -> Option { let value = nested.value()?; let s: syn::LitStr = value.parse()?; description = Some(s.value()); + } else if nested.path.is_ident("mqtt") { + mqtt = extract_mqtt_operation_bindings(&nested); } Ok(()) }); @@ -502,6 +533,7 @@ fn extract_operation(attr: &Attribute) -> Option { action: action?, channel: channel?, description, + mqtt, }) } diff --git a/asyncapi-rust-codegen/src/lib.rs b/asyncapi-rust-codegen/src/lib.rs index 4bd97b2..321e4d2 100644 --- a/asyncapi-rust-codegen/src/lib.rs +++ b/asyncapi-rust-codegen/src/lib.rs @@ -634,6 +634,56 @@ pub fn derive_asyncapi(input: TokenStream) -> TokenStream { } }; + let bindings = if let Some(mqtt) = &server.mqtt { + let client_id = if let Some(s) = mqtt.client_id { + quote! { Some(#s.to_string()) } + } else { + quote! { None } + }; + let clean_session = if let Some(s) = mqtt.clean_session { + quote! { Some(#s.to_string()) } + } else { + quote! { None } + }; + let last_will = if let Some(lw) = mqtt.last_will { + let topic = lw.topic; + let qos = lw.qos; + let retain = lw.retain; + let message = lw.message; + quote! { + Some( + asyncapi_rust::LastWill { + topic: quote! { #topic.to_string() }, + qos: quote! { #qos }, + retain: quote! { #retain }, + message: quote! { #message.to_string() } + } + ) + } + } else { + quote! { None } + }; + let binding_version = mqtt.binding_version.as_deref(); + let session_expiry_interval = mqtt.session_expiry_interval.as_ref().map(|v| { + quote! { + asyncapi_rust::Schema::Any(serde_json::json!(#v)) + } + }); + + quote! { + Some(asyncapi_rust::ServerBindings { + mqtt: Some(asyncapi_rust::MqttServerBindings { + client_id: #client_id, + client_session: #client_session, + session_expiry_interval: #session_expiry_interval, + binding_version: #binding_version.map(|s| s.to_string()), + }) + }) + } + } else { + quote! { None } + }; + quote! { servers.insert( #name.to_string(), @@ -769,6 +819,30 @@ pub fn derive_asyncapi(input: TokenStream) -> TokenStream { .to_compile_error(); }; + let bindings = if let Some(mqtt) = &operation.mqtt { + let qos = mqtt.qos; + let retain = mqtt.retain; + let binding_version = mqtt.binding_version.as_deref(); + let message_expiry = mqtt.message_expiry_interval.as_ref().map(|v| { + quote! { + asyncapi_rust::Schema::Any(serde_json::json!(#v)) + } + }); + + quote! { + Some(asyncapi_rust::OperationBindings { + mqtt: Some(asyncapi_rust::MqttOperationBindings { + qos: #qos, + retain: #retain, + message_expiry_interval: #message_expiry, + binding_version: #binding_version.map(|s| s.to_string()), + }) + }) + } + } else { + quote! { None } + }; + quote! { operations.insert( #name.to_string(), @@ -778,6 +852,7 @@ pub fn derive_asyncapi(input: TokenStream) -> TokenStream { reference: format!("#/channels/{}", #channel_ref), }, messages: None, + bindings: #bindings } ); } diff --git a/asyncapi-rust/examples/chat_api.rs b/asyncapi-rust/examples/chat_api.rs index d26210c..ce88dbc 100644 --- a/asyncapi-rust/examples/chat_api.rs +++ b/asyncapi-rust/examples/chat_api.rs @@ -132,6 +132,7 @@ fn build_asyncapi_spec(messages: Vec) -> AsyncApiSpec { pathname: None, description: Some("Production WebSocket server".to_string()), variables: None, + bindings: None, }, ); @@ -168,6 +169,7 @@ fn build_asyncapi_spec(messages: Vec) -> AsyncApiSpec { }) .collect(), ), + bindings: None, }, ); @@ -190,6 +192,7 @@ fn build_asyncapi_spec(messages: Vec) -> AsyncApiSpec { }) .collect(), ), + bindings: None, }, ); From a11a82074a246b4f8c1f508eddf0e51524e041b8 Mon Sep 17 00:00:00 2001 From: Max Christoph Date: Thu, 25 Jun 2026 10:50:25 +0200 Subject: [PATCH 04/12] feat: add mqtt message binding --- asyncapi-rust-codegen/src/asyncapi_attrs.rs | 7 +- .../src/asyncapi_spec_attrs.rs | 1 + asyncapi-rust-codegen/src/lib.rs | 84 ++++++++++++++++--- 3 files changed, 79 insertions(+), 13 deletions(-) diff --git a/asyncapi-rust-codegen/src/asyncapi_attrs.rs b/asyncapi-rust-codegen/src/asyncapi_attrs.rs index 395ad01..7d25278 100644 --- a/asyncapi-rust-codegen/src/asyncapi_attrs.rs +++ b/asyncapi-rust-codegen/src/asyncapi_attrs.rs @@ -4,6 +4,7 @@ use syn::{Attribute, Meta, Path}; #[derive(Clone, Debug)] pub enum ResponseTopic { + #[allow(unused)] Reference(Path), Uri(String), } @@ -15,7 +16,7 @@ impl Default for ResponseTopic { } #[derive(Clone, Debug, Default)] -pub struct MqttMessageBinding { +pub struct MqttMessageBindingsMeta { pub payload_format_indicator: Option, pub correlation_data: Option, pub content_type: Option, @@ -34,7 +35,7 @@ pub struct AsyncApiMeta { /// Override the message name used in `components.messages` and `asyncapi_message_names()`. /// When absent the Rust variant/type identifier is used. pub message_name: Option, - pub mqtt: Option, + pub mqtt: Option, } fn parse_response_topic(meta: Meta) -> Option { @@ -91,7 +92,7 @@ pub fn extract_asyncapi_meta(attrs: &[Attribute]) -> AsyncApiMeta { let s: syn::LitStr = value.parse()?; meta.message_name = Some(s.value()); } else if nested.path.is_ident("mqtt") { - let mut binding = MqttMessageBinding::default(); + let mut binding = MqttMessageBindingsMeta::default(); let value = nested.value()?; let mqtt_meta: syn::MetaList = value.parse()?; diff --git a/asyncapi-rust-codegen/src/asyncapi_spec_attrs.rs b/asyncapi-rust-codegen/src/asyncapi_spec_attrs.rs index bd18351..a5d8e84 100644 --- a/asyncapi-rust-codegen/src/asyncapi_spec_attrs.rs +++ b/asyncapi-rust-codegen/src/asyncapi_spec_attrs.rs @@ -41,6 +41,7 @@ pub struct MqttServerBindingsMeta { pub last_will: Option, pub keep_alive: Option, pub session_expiry_interval: Option, + #[allow(unused)] pub maximum_packet_size: Option, pub binding_version: Option, } diff --git a/asyncapi-rust-codegen/src/lib.rs b/asyncapi-rust-codegen/src/lib.rs index 321e4d2..d706155 100644 --- a/asyncapi-rust-codegen/src/lib.rs +++ b/asyncapi-rust-codegen/src/lib.rs @@ -222,6 +222,7 @@ pub fn derive_to_asyncapi_message(input: TokenStream) -> TokenStream { title: Option, content_type: Option, triggers_binary: bool, + mqtt: Option, } // Parse enum variants or struct @@ -256,6 +257,7 @@ pub fn derive_to_asyncapi_message(input: TokenStream) -> TokenStream { title: asyncapi_meta.title, content_type: asyncapi_meta.content_type, triggers_binary: asyncapi_meta.triggers_binary, + mqtt: asyncapi_meta.mqtt, }); } @@ -278,6 +280,7 @@ pub fn derive_to_asyncapi_message(input: TokenStream) -> TokenStream { title: asyncapi_meta.title, content_type: asyncapi_meta.content_type, triggers_binary: asyncapi_meta.triggers_binary, + mqtt: asyncapi_meta.mqtt, }] } Data::Union(_) => { @@ -326,6 +329,50 @@ pub fn derive_to_asyncapi_message(input: TokenStream) -> TokenStream { } }); + let message_mqtt_bindings = messages.iter().map(|m| { + if let Some(ref mqtt) = m.mqtt { + let payload_format_indicator = if let Some(s) = mqtt.payload_format_indicator { + quote! { Some(#s) } + } else { + quote! { None } + }; + let content_type = if let Some(s) = &mqtt.content_type { + quote! { Some(#s.to_string()) } + } else { + quote! { None } + }; + let binding_version = if let Some(s) = &mqtt.binding_version { + quote! { Some(#s.to_string()) } + } else { + quote! { None } + }; + let response_topic = match &mqtt.response_topic { + Some(crate::asyncapi_attrs::ResponseTopic::Uri(t)) => { + quote! { + Some(crate::asyncapi_attrs::ResponseTopic::Uri(#t.to_string())) + } + } + _ => quote! { None }, + }; + + quote! { + Some( + asyncapi_rust::MessageBindings { + mqtt: Some(asyncapi_rust::MqttMessageBindings { + payload_format_indicator: #payload_format_indicator, + content_type: #content_type, + binding_version: #binding_version, + response_topic: #response_topic, + correlation_data: None + }) + } + ) + } + } else { + quote! { None } + } + }); + let tag_info = if let Some(tag) = tag_field { quote! { Some(#tag) @@ -466,6 +513,7 @@ pub fn derive_to_asyncapi_message(input: TokenStream) -> TokenStream { let summaries: &[Option] = &[#(#message_summaries),*]; let descriptions: &[Option] = &[#(#message_descriptions),*]; let content_types: &[Option] = &[#(#message_content_types),*]; + let bindings: &[Option] = &[#(#message_mqtt_bindings),*]; let mut messages = Vec::with_capacity(names.len()); for i in 0..names.len() { @@ -490,6 +538,7 @@ pub fn derive_to_asyncapi_message(input: TokenStream) -> TokenStream { description: descriptions[i].clone(), content_type: content_types[i].clone(), payload, + bindings: bindings[i].clone() }); } messages @@ -635,7 +684,7 @@ pub fn derive_asyncapi(input: TokenStream) -> TokenStream { }; let bindings = if let Some(mqtt) = &server.mqtt { - let client_id = if let Some(s) = mqtt.client_id { + let client_id = if let Some(s) = &mqtt.client_id { quote! { Some(#s.to_string()) } } else { quote! { None } @@ -645,11 +694,11 @@ pub fn derive_asyncapi(input: TokenStream) -> TokenStream { } else { quote! { None } }; - let last_will = if let Some(lw) = mqtt.last_will { - let topic = lw.topic; + let last_will = if let Some(lw) = &mqtt.last_will { + let topic = &lw.topic; let qos = lw.qos; let retain = lw.retain; - let message = lw.message; + let message = &lw.message; quote! { Some( asyncapi_rust::LastWill { @@ -663,20 +712,34 @@ pub fn derive_asyncapi(input: TokenStream) -> TokenStream { } else { quote! { None } }; - let binding_version = mqtt.binding_version.as_deref(); - let session_expiry_interval = mqtt.session_expiry_interval.as_ref().map(|v| { + let binding_version = if let Some(s) = &mqtt.binding_version { + quote! { Some(#s.to_string()) } + } else { + quote! { None } + }; + let session_expiry_interval = if let Some(s) = mqtt.session_expiry_interval { quote! { - asyncapi_rust::Schema::Any(serde_json::json!(#v)) + Some(asyncapi_rust::Schema::Any(serde_json::json!(#s))) } - }); + } else { + quote! { None } + }; + + let keep_alive = if let Some(k) = mqtt.keep_alive { + quote! { Some(#k) } + } else { + quote! { None } + }; quote! { Some(asyncapi_rust::ServerBindings { mqtt: Some(asyncapi_rust::MqttServerBindings { client_id: #client_id, - client_session: #client_session, + client_session: #clean_session, session_expiry_interval: #session_expiry_interval, - binding_version: #binding_version.map(|s| s.to_string()), + binding_version: #binding_version, + last_will: #last_will, + keep_alive: #keep_alive }) }) } @@ -693,6 +756,7 @@ pub fn derive_asyncapi(input: TokenStream) -> TokenStream { pathname: #pathname, description: #desc, variables: #variables, + bindings: #bindings } ); } From 6ae89788ba69d1e395d7054ff4f596a75896637c Mon Sep 17 00:00:00 2001 From: Max Christoph Date: Thu, 25 Jun 2026 11:52:15 +0200 Subject: [PATCH 05/12] feat: Tests for mqtt server, message bindings --- asyncapi-rust-codegen/src/asyncapi_attrs.rs | 52 ++++---- .../src/asyncapi_spec_attrs.rs | 113 +++++++++++------- 2 files changed, 94 insertions(+), 71 deletions(-) diff --git a/asyncapi-rust-codegen/src/asyncapi_attrs.rs b/asyncapi-rust-codegen/src/asyncapi_attrs.rs index 7d25278..49d9c2a 100644 --- a/asyncapi-rust-codegen/src/asyncapi_attrs.rs +++ b/asyncapi-rust-codegen/src/asyncapi_attrs.rs @@ -1,8 +1,8 @@ //! Utilities for parsing asyncapi attributes -use syn::{Attribute, Meta, Path}; +use syn::{Attribute, Path}; -#[derive(Clone, Debug)] +#[derive(Clone, Debug, PartialEq)] pub enum ResponseTopic { #[allow(unused)] Reference(Path), @@ -15,7 +15,7 @@ impl Default for ResponseTopic { } } -#[derive(Clone, Debug, Default)] +#[derive(Clone, Debug, Default, PartialEq)] pub struct MqttMessageBindingsMeta { pub payload_format_indicator: Option, pub correlation_data: Option, @@ -38,22 +38,13 @@ pub struct AsyncApiMeta { pub mqtt: Option, } -fn parse_response_topic(meta: Meta) -> Option { - match meta { - Meta::Path(path) => Some(ResponseTopic::Reference(path)), - - Meta::NameValue(nv) => { - match nv.value { - syn::Expr::Lit(expr_lit) => { - match expr_lit.lit { - syn::Lit::Str(s) => Some(ResponseTopic::Uri(s.value())), - _ => None, // invalid literal → ignore - } - } - _ => None, - } - } - +fn parse_response_topic(expr: syn::Expr) -> Option { + match expr { + syn::Expr::Lit(expr_lit) => match expr_lit.lit { + syn::Lit::Str(s) => Some(ResponseTopic::Uri(s.value())), + _ => None, + }, + syn::Expr::Path(expr_path) => Some(ResponseTopic::Reference(expr_path.path)), _ => None, } } @@ -94,10 +85,7 @@ pub fn extract_asyncapi_meta(attrs: &[Attribute]) -> AsyncApiMeta { } else if nested.path.is_ident("mqtt") { let mut binding = MqttMessageBindingsMeta::default(); - let value = nested.value()?; - let mqtt_meta: syn::MetaList = value.parse()?; - - mqtt_meta.parse_nested_meta(|m| { + nested.parse_nested_meta(|m| { if m.path.is_ident("payload_format_indicator") { let v = m.value()?; let lit: syn::LitInt = v.parse()?; @@ -105,7 +93,6 @@ pub fn extract_asyncapi_meta(attrs: &[Attribute]) -> AsyncApiMeta { } else if m.path.is_ident("correlation_data") { let v = m.value()?; let s: syn::Path = v.parse()?; - binding.correlation_data = Some(s); } else if m.path.is_ident("content_type") { let v = m.value()?; @@ -113,9 +100,8 @@ pub fn extract_asyncapi_meta(attrs: &[Attribute]) -> AsyncApiMeta { binding.content_type = Some(s.value()); } else if m.path.is_ident("response_topic") { let v = m.value()?; - let meta_inner: syn::Meta = v.parse()?; - - binding.response_topic = parse_response_topic(meta_inner); + let expr: syn::Expr = v.parse()?; + binding.response_topic = parse_response_topic(expr); } else if m.path.is_ident("binding_version") { let v = m.value()?; let s: syn::LitStr = v.parse()?; @@ -153,7 +139,7 @@ mod tests { #[test] fn test_extract_multiple() { let attrs: Vec = vec![parse_quote! { - #[asyncapi(summary = "Send message", description = "Sends a chat message to a room")] + #[asyncapi(summary = "Send message", description = "Sends a chat message to a room", mqtt(response_topic = "/a/b/d"))] }]; let meta = extract_asyncapi_meta(&attrs); @@ -162,6 +148,16 @@ mod tests { meta.description, Some("Sends a chat message to a room".to_string()) ); + assert_eq!( + meta.mqtt, + Some(MqttMessageBindingsMeta { + response_topic: Some(ResponseTopic::Uri("/a/b/d".to_string())), + payload_format_indicator: None, + content_type: None, + correlation_data: None, + binding_version: None + }) + ) } #[test] diff --git a/asyncapi-rust-codegen/src/asyncapi_spec_attrs.rs b/asyncapi-rust-codegen/src/asyncapi_spec_attrs.rs index a5d8e84..9df62de 100644 --- a/asyncapi-rust-codegen/src/asyncapi_spec_attrs.rs +++ b/asyncapi-rust-codegen/src/asyncapi_spec_attrs.rs @@ -249,58 +249,49 @@ fn extract_mqtt_server_bindings( let s: syn::LitStr = value.parse()?; binding_version = Some(s.value()); } else if inner.path.is_ident("last_will") { + let content; + syn::parenthesized!(content in inner.input); + let mut topic: Option = None; let mut qos: Option = None; let mut message: Option = None; let mut retain: Option = None; - inner.parse_nested_meta(|lw| { - if lw.path.is_ident("topic") { - let value = lw.value()?; - let s: syn::LitStr = value.parse()?; - topic = Some(s.value()); - } else if lw.path.is_ident("qos") { - let value = lw.value()?; - let s: syn::LitInt = value.parse()?; - qos = Some(s.base10_parse()?); - } else if lw.path.is_ident("message") { - let value = lw.value()?; - let s: syn::LitStr = value.parse()?; - message = Some(s.value()); - } else if lw.path.is_ident("retain") { - let value = lw.value()?; - let s: syn::LitBool = value.parse()?; - retain = Some(s.value()); + while !content.is_empty() { + let key: syn::Ident = content.parse()?; + content.parse::()?; + + match key.to_string().as_str() { + "topic" => { + let v: syn::LitStr = content.parse()?; + topic = Some(v.value()); + } + "qos" => { + let v: syn::LitInt = content.parse()?; + qos = Some(v.base10_parse()?); + } + "message" => { + let v: syn::LitStr = content.parse()?; + message = Some(v.value()); + } + "retain" => { + let v: syn::LitBool = content.parse()?; + retain = Some(v.value()); + } + _ => {} } - Ok(()) - })?; + let _ = content.parse::(); + } last_will = Some(LastWillMeta { - topic: topic.ok_or_else(|| { - syn::Error::new_spanned( - &inner.path, - "missing required field `topic` in last_will", - ) - })?, - qos: qos.ok_or_else(|| { - syn::Error::new_spanned( - &inner.path, - "missing required field `qos` in last_will", - ) - })?, - message: message.ok_or_else(|| { - syn::Error::new_spanned( - &inner.path, - "missing required field `message` in last_will", - ) - })?, - retain: retain.ok_or_else(|| { - syn::Error::new_spanned( - &inner.path, - "missing required field `retain` in last_will", - ) - })?, + topic: topic + .ok_or_else(|| syn::Error::new_spanned(&inner.path, "missing topic"))?, + qos: qos.ok_or_else(|| syn::Error::new_spanned(&inner.path, "missing qos"))?, + message: message + .ok_or_else(|| syn::Error::new_spanned(&inner.path, "missing message"))?, + retain: retain + .ok_or_else(|| syn::Error::new_spanned(&inner.path, "missing retain"))?, }); } @@ -757,6 +748,42 @@ mod tests { assert_eq!(var1.examples, vec!["12".to_string(), "13".to_string()]); } + #[test] + fn test_extract_server_with_mqtt_bindings() { + let attrs: Vec = vec![parse_quote! { + #[asyncapi_server( + name = "staging", + host = "staging.example.com", + protocol = "wss", + pathname = "/api/{version}/ws/{userId}", + mqtt( + client_id = "abc", + last_will( + topic: "a", + qos: 0, + message: "B", + retain: true + ) + ) + )] + }]; + + let meta = extract_asyncapi_spec_meta(&attrs); + let server = &meta.servers[0]; + + let mqtt = &server.mqtt; + + assert!(mqtt.is_some()); + let mqtt = mqtt.clone().unwrap(); + assert_eq!(mqtt.client_id, Some("abc".to_string())); + assert!(mqtt.last_will.is_some()); + let lw = mqtt.last_will.unwrap(); + assert_eq!(lw.topic, "a".to_string()); + assert_eq!(lw.qos, 0); + assert_eq!(lw.message, "B".to_string()); + assert!(lw.retain) + } + #[test] fn test_extract_channel_with_parameters() { let attrs: Vec = vec![parse_quote! { From 7d296caa65beec472fa59e21d88d1857c1630c80 Mon Sep 17 00:00:00 2001 From: Max Christoph Date: Thu, 25 Jun 2026 12:20:33 +0200 Subject: [PATCH 06/12] feat: Add test for operation binding parsing --- .../src/asyncapi_spec_attrs.rs | 20 +++++++++++++++++++ 1 file changed, 20 insertions(+) diff --git a/asyncapi-rust-codegen/src/asyncapi_spec_attrs.rs b/asyncapi-rust-codegen/src/asyncapi_spec_attrs.rs index 9df62de..6e4749e 100644 --- a/asyncapi-rust-codegen/src/asyncapi_spec_attrs.rs +++ b/asyncapi-rust-codegen/src/asyncapi_spec_attrs.rs @@ -634,6 +634,26 @@ mod tests { assert_eq!(meta.operations[0].channel, "chat"); } + #[test] + fn test_extract_operation_with_mqtt_binding() { + let attrs: Vec = vec![parse_quote! { + #[asyncapi_operation(name = "sendMessage", action = "send", channel = "chat", mqtt(qos = 1, retain = true))] + }]; + + let meta = extract_asyncapi_spec_meta(&attrs); + assert_eq!(meta.operations.len(), 1); + assert_eq!(meta.operations[0].name, "sendMessage"); + assert_eq!(meta.operations[0].action, "send"); + assert_eq!(meta.operations[0].channel, "chat"); + + assert!(meta.operations[0].mqtt.is_some()); + let mqtt = meta.operations[0].mqtt.clone().unwrap(); + assert_eq!(mqtt.qos, Some(1)); + assert!(mqtt.retain.unwrap()); + assert!(mqtt.binding_version.is_none()); + assert!(mqtt.message_expiry_interval.is_none()); + } + #[test] fn test_extract_multiple_components() { let attrs: Vec = vec![ From d65c7bd6d2e1fab856b86387040aad1955ece51c Mon Sep 17 00:00:00 2001 From: Max Christoph Date: Thu, 25 Jun 2026 13:17:09 +0200 Subject: [PATCH 07/12] fix: separate test for mqtt operation bindings --- asyncapi-rust-codegen/src/asyncapi_attrs.rs | 11 ++++++++++- 1 file changed, 10 insertions(+), 1 deletion(-) diff --git a/asyncapi-rust-codegen/src/asyncapi_attrs.rs b/asyncapi-rust-codegen/src/asyncapi_attrs.rs index 49d9c2a..69dadce 100644 --- a/asyncapi-rust-codegen/src/asyncapi_attrs.rs +++ b/asyncapi-rust-codegen/src/asyncapi_attrs.rs @@ -139,7 +139,7 @@ mod tests { #[test] fn test_extract_multiple() { let attrs: Vec = vec![parse_quote! { - #[asyncapi(summary = "Send message", description = "Sends a chat message to a room", mqtt(response_topic = "/a/b/d"))] + #[asyncapi(summary = "Send message", description = "Sends a chat message to a room")] }]; let meta = extract_asyncapi_meta(&attrs); @@ -148,6 +148,15 @@ mod tests { meta.description, Some("Sends a chat message to a room".to_string()) ); + } + + #[test] + fn test_extract_multiple_with_mqtt_bindings() { + let attrs: Vec = vec![parse_quote! { + #[asyncapi(mqtt(response_topic = "/a/b/d"))] + }]; + + let meta = extract_asyncapi_meta(&attrs); assert_eq!( meta.mqtt, Some(MqttMessageBindingsMeta { From 735d44b94869bc8258accadc962bb63d1fbadc5e Mon Sep 17 00:00:00 2001 From: Max Christoph Date: Mon, 13 Jul 2026 07:35:58 +0200 Subject: [PATCH 08/12] fix reviewed issues --- asyncapi-rust-codegen/src/asyncapi_attrs.rs | 1 - .../src/asyncapi_spec_attrs.rs | 87 +++++++------ asyncapi-rust-codegen/src/lib.rs | 111 +++++++++++++--- asyncapi-rust-models/src/lib.rs | 47 +++++-- asyncapi-rust/examples/mqtt_bindings.rs | 120 ++++++++++++++++++ 5 files changed, 294 insertions(+), 72 deletions(-) create mode 100644 asyncapi-rust/examples/mqtt_bindings.rs diff --git a/asyncapi-rust-codegen/src/asyncapi_attrs.rs b/asyncapi-rust-codegen/src/asyncapi_attrs.rs index 69dadce..6e515c2 100644 --- a/asyncapi-rust-codegen/src/asyncapi_attrs.rs +++ b/asyncapi-rust-codegen/src/asyncapi_attrs.rs @@ -4,7 +4,6 @@ use syn::{Attribute, Path}; #[derive(Clone, Debug, PartialEq)] pub enum ResponseTopic { - #[allow(unused)] Reference(Path), Uri(String), } diff --git a/asyncapi-rust-codegen/src/asyncapi_spec_attrs.rs b/asyncapi-rust-codegen/src/asyncapi_spec_attrs.rs index 6e4749e..3a8b5f2 100644 --- a/asyncapi-rust-codegen/src/asyncapi_spec_attrs.rs +++ b/asyncapi-rust-codegen/src/asyncapi_spec_attrs.rs @@ -88,14 +88,37 @@ pub struct OperationMeta { pub mqtt: Option, } +#[derive(Debug, Clone)] +pub enum MqttMessageExpiryIntervalMeta { + Interval(u32), + Reference(Path), +} + #[derive(Debug, Clone)] pub struct OperationMqttBindingsMeta { pub qos: Option, pub retain: Option, - pub message_expiry_interval: Option, + pub message_expiry_interval: Option, pub binding_version: Option, } +fn parse_message_expiry_interval( + expr: syn::Expr, +) -> Result, syn::Error> { + match expr { + syn::Expr::Lit(expr_lit) => match expr_lit.lit { + syn::Lit::Int(s) => Ok(Some(MqttMessageExpiryIntervalMeta::Interval( + s.base10_parse()?, + ))), + _ => Ok(None), + }, + syn::Expr::Path(expr_path) => Ok(Some(MqttMessageExpiryIntervalMeta::Reference( + expr_path.path, + ))), + _ => Ok(None), + } +} + /// Extract asyncapi spec metadata from `#[asyncapi(...)]` attributes pub fn extract_asyncapi_spec_meta(attrs: &[Attribute]) -> AsyncApiSpecMeta { let mut meta = AsyncApiSpecMeta::default(); @@ -249,40 +272,32 @@ fn extract_mqtt_server_bindings( let s: syn::LitStr = value.parse()?; binding_version = Some(s.value()); } else if inner.path.is_ident("last_will") { - let content; - syn::parenthesized!(content in inner.input); - let mut topic: Option = None; let mut qos: Option = None; let mut message: Option = None; let mut retain: Option = None; - while !content.is_empty() { - let key: syn::Ident = content.parse()?; - content.parse::()?; - - match key.to_string().as_str() { - "topic" => { - let v: syn::LitStr = content.parse()?; - topic = Some(v.value()); - } - "qos" => { - let v: syn::LitInt = content.parse()?; - qos = Some(v.base10_parse()?); - } - "message" => { - let v: syn::LitStr = content.parse()?; - message = Some(v.value()); - } - "retain" => { - let v: syn::LitBool = content.parse()?; - retain = Some(v.value()); - } - _ => {} + inner.parse_nested_meta(|meta| { + if meta.path.is_ident("topic") { + let value = meta.value()?; + let s: syn::LitStr = value.parse()?; + topic = Some(s.value()); + } else if meta.path.is_ident("qos") { + let value = meta.value()?; + let v: syn::LitInt = value.parse()?; + qos = Some(v.base10_parse()?); + } else if meta.path.is_ident("message") { + let value = meta.value()?; + let s: syn::LitStr = value.parse()?; + message = Some(s.value()); + } else if meta.path.is_ident("retain") { + let value = meta.value()?; + let b: syn::LitBool = value.parse()?; + retain = Some(b.value()); } - let _ = content.parse::(); - } + Ok(()) + })?; last_will = Some(LastWillMeta { topic: topic @@ -470,8 +485,8 @@ fn extract_mqtt_operation_bindings( retain = Some(s.value()); } else if inner.path.is_ident("message_expiry_interval") { let value = inner.value()?; - let p: syn::LitInt = value.parse()?; - message_expiry_interval = Some(p.base10_parse()?); + let expr: syn::Expr = value.parse()?; + message_expiry_interval = parse_message_expiry_interval(expr)?; } else if inner.path.is_ident("binding_version") { let value = inner.value()?; let s: LitStr = value.parse()?; @@ -778,12 +793,6 @@ mod tests { pathname = "/api/{version}/ws/{userId}", mqtt( client_id = "abc", - last_will( - topic: "a", - qos: 0, - message: "B", - retain: true - ) ) )] }]; @@ -796,12 +805,6 @@ mod tests { assert!(mqtt.is_some()); let mqtt = mqtt.clone().unwrap(); assert_eq!(mqtt.client_id, Some("abc".to_string())); - assert!(mqtt.last_will.is_some()); - let lw = mqtt.last_will.unwrap(); - assert_eq!(lw.topic, "a".to_string()); - assert_eq!(lw.qos, 0); - assert_eq!(lw.message, "B".to_string()); - assert!(lw.retain) } #[test] diff --git a/asyncapi-rust-codegen/src/lib.rs b/asyncapi-rust-codegen/src/lib.rs index d706155..5db9cd1 100644 --- a/asyncapi-rust-codegen/src/lib.rs +++ b/asyncapi-rust-codegen/src/lib.rs @@ -349,12 +349,43 @@ pub fn derive_to_asyncapi_message(input: TokenStream) -> TokenStream { let response_topic = match &mqtt.response_topic { Some(crate::asyncapi_attrs::ResponseTopic::Uri(t)) => { quote! { - Some(crate::asyncapi_attrs::ResponseTopic::Uri(#t.to_string())) + Some(asyncapi_rust::MqttReponseTopic::Uri( + #t.to_string() + )) + } + } + Some(crate::asyncapi_attrs::ResponseTopic::Reference(r)) => { + quote! { + Some(asyncapi_rust::MqttReponseTopic::Schema({ + let schema = schemars::schema_for!(#r); + + let schema_json = serde_json::to_value(&schema) + .expect("Failed to serialize schema"); + + serde_json::from_value(schema_json) + .expect("Failed to deserialize schema") + })) } } _ => quote! { None }, }; + let correlation_data = if let Some(c) = &mqtt.correlation_data { + quote! { + Some({ + let schema = schemars::schema_for!(#c); + + let schema_json = serde_json::to_value(&schema) + .expect("Failed to serialize schema"); + + serde_json::from_value::(schema_json) + .expect("Failed to deserialize schema") + }) + } + } else { + quote! { None } + }; + quote! { Some( asyncapi_rust::MessageBindings { @@ -363,7 +394,7 @@ pub fn derive_to_asyncapi_message(input: TokenStream) -> TokenStream { content_type: #content_type, binding_version: #binding_version, response_topic: #response_topic, - correlation_data: None + correlation_data: #correlation_data }) } ) @@ -690,7 +721,7 @@ pub fn derive_asyncapi(input: TokenStream) -> TokenStream { quote! { None } }; let clean_session = if let Some(s) = mqtt.clean_session { - quote! { Some(#s.to_string()) } + quote! { Some(#s) } } else { quote! { None } }; @@ -701,11 +732,11 @@ pub fn derive_asyncapi(input: TokenStream) -> TokenStream { let message = &lw.message; quote! { Some( - asyncapi_rust::LastWill { - topic: quote! { #topic.to_string() }, - qos: quote! { #qos }, - retain: quote! { #retain }, - message: quote! { #message.to_string() } + asyncapi_rust::MqttLastWill { + topic: #topic.to_string(), + qos: #qos, + retain: #retain, + message: #message.to_string() } ) } @@ -731,15 +762,22 @@ pub fn derive_asyncapi(input: TokenStream) -> TokenStream { quote! { None } }; + let max_packet_size = if let Some(s) = &mqtt.maximum_packet_size { + quote! { Some(#s) } + } else { + quote! { None } + }; + quote! { Some(asyncapi_rust::ServerBindings { mqtt: Some(asyncapi_rust::MqttServerBindings { client_id: #client_id, - client_session: #clean_session, + clean_session: #clean_session, session_expiry_interval: #session_expiry_interval, binding_version: #binding_version, last_will: #last_will, - keep_alive: #keep_alive + keep_alive: #keep_alive, + max_packet_size: #max_packet_size }) }) } @@ -884,14 +922,51 @@ pub fn derive_asyncapi(input: TokenStream) -> TokenStream { }; let bindings = if let Some(mqtt) = &operation.mqtt { - let qos = mqtt.qos; - let retain = mqtt.retain; - let binding_version = mqtt.binding_version.as_deref(); - let message_expiry = mqtt.message_expiry_interval.as_ref().map(|v| { - quote! { - asyncapi_rust::Schema::Any(serde_json::json!(#v)) + let qos = if let Some(s) = mqtt.qos { + quote! { Some(#s) } + } else { + quote! { None } + }; + + let retain = if let Some(b) = mqtt.retain { + quote! { Some(#b) } + } else { + quote! { None } + }; + + let binding_version = if let Some(s) = &mqtt.binding_version { + quote! { Some(#s.to_string()) } + } else { + quote! { None } + }; + + let message_expiry = match &mqtt.message_expiry_interval { + Some(crate::asyncapi_spec_attrs::MqttMessageExpiryIntervalMeta::Interval( + t, + )) => { + quote! { + Some(asyncapi_rust::MqttMessageExpiryInterval::Interval( + #t + )) + } } - }); + Some(crate::asyncapi_spec_attrs::MqttMessageExpiryIntervalMeta::Reference( + r, + )) => { + quote! { + Some(asyncapi_rust::MqttMessageExpiryInterval::Schema({ + let schema = schemars::schema_for!(#r); + + let schema_json = serde_json::to_value(&schema) + .expect("Failed to serialize schema"); + + serde_json::from_value(schema_json) + .expect("Failed to deserialize schema") + })) + } + } + _ => quote! { None }, + }; quote! { Some(asyncapi_rust::OperationBindings { @@ -899,7 +974,7 @@ pub fn derive_asyncapi(input: TokenStream) -> TokenStream { qos: #qos, retain: #retain, message_expiry_interval: #message_expiry, - binding_version: #binding_version.map(|s| s.to_string()), + binding_version: #binding_version, }) }) } diff --git a/asyncapi-rust-models/src/lib.rs b/asyncapi-rust-models/src/lib.rs index 195381d..5a3d211 100644 --- a/asyncapi-rust-models/src/lib.rs +++ b/asyncapi-rust-models/src/lib.rs @@ -151,9 +151,10 @@ pub struct Info { /// pathname: Some("/api/ws/{userId}".to_string()), /// description: Some("Production WebSocket server".to_string()), /// variables: Some(variables), +/// ..Default::default() /// }; /// ``` -#[derive(Debug, Clone, Serialize, Deserialize)] +#[derive(Debug, Clone, Serialize, Deserialize, Default)] pub struct Server { /// Server URL or host /// @@ -186,7 +187,7 @@ pub struct Server { pub variables: Option>, /// Protocol specific bindings. - #[serde(skip_serializing_if = "Option::is_none")] + #[serde(skip_serializing_if = "Option::is_none", default)] pub bindings: Option, } @@ -218,7 +219,7 @@ pub struct MqttLastWill { #[derive(Debug, Clone, Serialize, Deserialize)] pub struct MqttServerBindings { /// The client identifier. - #[serde(skip_serializing_if = "Option::is_none", rename = "cliendId")] + #[serde(skip_serializing_if = "Option::is_none", rename = "clientId")] pub client_id: Option, /// Whether to create a persistent connection or not. When false, @@ -414,6 +415,7 @@ pub struct Parameter { /// description: None, /// content_type: Some("application/json".to_string()), /// payload: None, +/// ..Default::default() /// })); /// ``` #[derive(Debug, Clone, Serialize, Deserialize)] @@ -467,9 +469,10 @@ pub enum MessageRef { /// all_of: None, /// additional: IndexMap::new(), /// }))), +/// ..Default::default() /// }; /// ``` -#[derive(Debug, Clone, Serialize, Deserialize)] +#[derive(Debug, Clone, Serialize, Deserialize, Default)] pub struct Message { /// Message name /// @@ -512,7 +515,7 @@ pub struct Message { pub payload: Option, /// Protocol specific bindings. - #[serde(skip_serializing_if = "Option::is_none")] + #[serde(skip_serializing_if = "Option::is_none", default)] pub bindings: Option, } @@ -524,6 +527,16 @@ pub struct MessageBindings { pub mqtt: Option, } +/// Mqtt response topic +#[derive(Serialize, Deserialize, Clone, Debug)] +#[serde(untagged)] +pub enum MqttReponseTopic { + /// Topic Uri + Uri(String), + /// Schema for the the response topic + Schema(Schema), +} + /// Mqtt message bindings #[derive(Serialize, Deserialize, Clone, Debug)] pub struct MqttMessageBindings { @@ -544,7 +557,7 @@ pub struct MqttMessageBindings { /// The topic (channel URI) for a response message. #[serde(skip_serializing_if = "Option::is_none", rename = "responseTopic")] - pub response_topic: Option, + pub response_topic: Option, /// The version of this binding. If omitted, "latest" MUST be assumed. #[serde(skip_serializing_if = "Option::is_none", rename = "bindingVersion")] @@ -567,9 +580,10 @@ pub struct MqttMessageBindings { /// reference: "#/channels/chat".to_string(), /// }, /// messages: None, +/// ..Default::default() /// }; /// ``` -#[derive(Debug, Clone, Serialize, Deserialize)] +#[derive(Debug, Clone, Serialize, Deserialize, Default)] pub struct Operation { /// Operation action (send or receive) /// @@ -588,7 +602,7 @@ pub struct Operation { pub messages: Option>, /// Protocol specific bindings - #[serde(skip_serializing_if = "Option::is_none")] + #[serde(skip_serializing_if = "Option::is_none", default)] pub bindings: Option, } @@ -600,6 +614,16 @@ pub struct OperationBindings { pub mqtt: Option, } +/// Interval in seconds or a Schema Object containing the definition of the lifetime of the message. +#[derive(Serialize, Deserialize, Clone, Debug)] +#[serde(untagged)] +pub enum MqttMessageExpiryInterval { + /// Interval in seconds + Interval(u32), + /// Schema Object containing the definition of the lifetime of the message + Schema(Schema), +} + /// Mqtt operation bindings #[derive(Serialize, Deserialize, Clone, Debug)] pub struct MqttOperationBindings { @@ -617,7 +641,7 @@ pub struct MqttOperationBindings { skip_serializing_if = "Option::is_none", rename = "messageExpiryInterval" )] - pub message_expiry_interval: Option, + pub message_expiry_interval: Option, /// The version of this binding. If omitted, "latest" MUST be assumed. #[serde(skip_serializing_if = "Option::is_none", rename = "bindingVersion")] @@ -625,17 +649,18 @@ pub struct MqttOperationBindings { } /// Operation action type -#[derive(Debug, Clone, Serialize, Deserialize)] +#[derive(Debug, Clone, Serialize, Deserialize, Default)] #[serde(rename_all = "lowercase")] pub enum OperationAction { /// Send message + #[default] Send, /// Receive message Receive, } /// Reference to a channel -#[derive(Debug, Clone, Serialize, Deserialize)] +#[derive(Debug, Clone, Serialize, Deserialize, Default)] pub struct ChannelRef { /// $ref path #[serde(rename = "$ref")] diff --git a/asyncapi-rust/examples/mqtt_bindings.rs b/asyncapi-rust/examples/mqtt_bindings.rs new file mode 100644 index 0000000..5b94879 --- /dev/null +++ b/asyncapi-rust/examples/mqtt_bindings.rs @@ -0,0 +1,120 @@ +use asyncapi_rust::{AsyncApi, ToAsyncApiMessage}; +use schemars::JsonSchema; +use serde::{Deserialize, Serialize}; + +#[derive(Serialize, Deserialize, JsonSchema)] +pub struct ExampleResponseTopic { + #[schemars(regex(pattern = "response/client/([a-z1-9]+)"))] + pub response: String, +} + +#[derive(Serialize, Deserialize, JsonSchema)] +pub struct A { + pub correlation_data: u8, +} + +#[derive(Serialize, Deserialize, JsonSchema, ToAsyncApiMessage)] +#[asyncapi( + mqtt( + payload_format_indicator = 1, + content_type = "application/json", + response_topic = ExampleResponseTopic, + binding_version = "1.0", + correlation_data = A + ) +)] +pub struct MqttMessage; + +#[derive(AsyncApi)] +#[asyncapi( + title = "Mqtt Api", + version = "1.0.0", + description = "Mqtt bindings example" +)] +#[asyncapi_server( + name = "production", + host = "api.example.com", + protocol = "mqtt", + description = "Production mqtt broker", + mqtt( + client_id = "client1", + clean_session = false, + last_will( + topic = "devices/", + qos = 0, + message = "LastWillMessage", + retain = false + ), + keep_alive = 60, + session_expiry_interval = 3600, + binding_version = "1.0" + ) +)] +#[asyncapi_channel(name = "test", address = "/test")] +#[asyncapi_operation( + name = "sendMqttMessage", + action = "send", + channel = "test", + mqtt( + qos = 2, + retain = true, + message_expiry_interval = 300, + binding_version = "1.0" + ) +)] +#[asyncapi_messages(MqttMessage)] +pub struct MqttApi; + +fn main() { + let spec = MqttApi::asyncapi_spec(); + + // Display Servers + if let Some(servers) = &spec.servers { + println!("🖥️ Servers ({}):", servers.len()); + for (name, server) in servers { + println!(" • {}", name); + println!(" Host: {}", server.host); + println!(" Protocol: {}", server.protocol); + if let Some(desc) = &server.description { + println!(" Description: {}", desc); + } + println!(" bindings: {:?}", &server.bindings) + } + println!(); + } + + // Display messages (automatically populated from ChatMessage and SystemMessage) + if let Some(components) = &spec.components { + if let Some(messages) = &components.messages { + println!("Messages (automatically included from message types):"); + for (name, message) in messages { + println!(" - {}", name); + if let Some(mqtt) = &message.bindings.as_ref().unwrap().mqtt { + println!(" Mqtt Bindings: {:?}", mqtt); + } + } + println!(); + } + } + + // Display Operations + if let Some(operations) = &spec.operations { + println!("⚡ Operations ({}):", operations.len()); + for (name, operation) in operations { + let action = match operation.action { + asyncapi_rust::OperationAction::Send => "send", + asyncapi_rust::OperationAction::Receive => "receive", + }; + println!(" • {} ({})", name, action); + println!(" Channel: {}", operation.channel.reference); + println!(" Mqtt: {:?}", &operation.bindings.as_ref().unwrap().mqtt); + } + println!(); + } + + // Serialize to JSON + let json = serde_json::to_string_pretty(&spec).expect("Failed to serialize spec"); + + println!("📄 Complete JSON Specification:\n"); + println!("{}", json); +} From 2230ea094a026c0049b923cee15ba9bf0263ae24 Mon Sep 17 00:00:00 2001 From: Max Christoph Date: Mon, 13 Jul 2026 12:15:16 +0200 Subject: [PATCH 09/12] fixed messageExpiryInterval, sessionExpiryInterval and maximumPacketSize parsing --- asyncapi-rust-codegen/src/asyncapi_attrs.rs | 14 ++-- .../src/asyncapi_spec_attrs.rs | 45 ++++++------ asyncapi-rust-codegen/src/lib.rs | 69 ++++++++++++++----- asyncapi-rust-models/src/lib.rs | 26 +++---- asyncapi-rust/examples/mqtt_bindings.rs | 22 ++++++ 5 files changed, 112 insertions(+), 64 deletions(-) diff --git a/asyncapi-rust-codegen/src/asyncapi_attrs.rs b/asyncapi-rust-codegen/src/asyncapi_attrs.rs index 6e515c2..2222d80 100644 --- a/asyncapi-rust-codegen/src/asyncapi_attrs.rs +++ b/asyncapi-rust-codegen/src/asyncapi_attrs.rs @@ -3,12 +3,12 @@ use syn::{Attribute, Path}; #[derive(Clone, Debug, PartialEq)] -pub enum ResponseTopic { +pub enum ResponseTopicMeta { Reference(Path), Uri(String), } -impl Default for ResponseTopic { +impl Default for ResponseTopicMeta { fn default() -> Self { Self::Uri(String::default()) } @@ -19,7 +19,7 @@ pub struct MqttMessageBindingsMeta { pub payload_format_indicator: Option, pub correlation_data: Option, pub content_type: Option, - pub response_topic: Option, + pub response_topic: Option, pub binding_version: Option, } @@ -37,13 +37,13 @@ pub struct AsyncApiMeta { pub mqtt: Option, } -fn parse_response_topic(expr: syn::Expr) -> Option { +fn parse_response_topic(expr: syn::Expr) -> Option { match expr { syn::Expr::Lit(expr_lit) => match expr_lit.lit { - syn::Lit::Str(s) => Some(ResponseTopic::Uri(s.value())), + syn::Lit::Str(s) => Some(ResponseTopicMeta::Uri(s.value())), _ => None, }, - syn::Expr::Path(expr_path) => Some(ResponseTopic::Reference(expr_path.path)), + syn::Expr::Path(expr_path) => Some(ResponseTopicMeta::Reference(expr_path.path)), _ => None, } } @@ -159,7 +159,7 @@ mod tests { assert_eq!( meta.mqtt, Some(MqttMessageBindingsMeta { - response_topic: Some(ResponseTopic::Uri("/a/b/d".to_string())), + response_topic: Some(ResponseTopicMeta::Uri("/a/b/d".to_string())), payload_format_indicator: None, content_type: None, correlation_data: None, diff --git a/asyncapi-rust-codegen/src/asyncapi_spec_attrs.rs b/asyncapi-rust-codegen/src/asyncapi_spec_attrs.rs index 3a8b5f2..74a11d3 100644 --- a/asyncapi-rust-codegen/src/asyncapi_spec_attrs.rs +++ b/asyncapi-rust-codegen/src/asyncapi_spec_attrs.rs @@ -34,15 +34,20 @@ pub struct LastWillMeta { pub message: String, } +#[derive(Debug, Clone)] +pub enum MqttBindingNumValueMeta { + Value(u32), + Reference(Path), +} + #[derive(Debug, Clone)] pub struct MqttServerBindingsMeta { pub client_id: Option, pub clean_session: Option, pub last_will: Option, pub keep_alive: Option, - pub session_expiry_interval: Option, - #[allow(unused)] - pub maximum_packet_size: Option, + pub session_expiry_interval: Option, + pub maximum_packet_size: Option, pub binding_version: Option, } @@ -88,33 +93,23 @@ pub struct OperationMeta { pub mqtt: Option, } -#[derive(Debug, Clone)] -pub enum MqttMessageExpiryIntervalMeta { - Interval(u32), - Reference(Path), -} - #[derive(Debug, Clone)] pub struct OperationMqttBindingsMeta { pub qos: Option, pub retain: Option, - pub message_expiry_interval: Option, + pub message_expiry_interval: Option, pub binding_version: Option, } -fn parse_message_expiry_interval( +fn parse_mqtt_binding_value( expr: syn::Expr, -) -> Result, syn::Error> { +) -> Result, syn::Error> { match expr { syn::Expr::Lit(expr_lit) => match expr_lit.lit { - syn::Lit::Int(s) => Ok(Some(MqttMessageExpiryIntervalMeta::Interval( - s.base10_parse()?, - ))), + syn::Lit::Int(s) => Ok(Some(MqttBindingNumValueMeta::Value(s.base10_parse()?))), _ => Ok(None), }, - syn::Expr::Path(expr_path) => Ok(Some(MqttMessageExpiryIntervalMeta::Reference( - expr_path.path, - ))), + syn::Expr::Path(expr_path) => Ok(Some(MqttBindingNumValueMeta::Reference(expr_path.path))), _ => Ok(None), } } @@ -260,13 +255,13 @@ fn extract_mqtt_server_bindings( let s: syn::LitInt = value.parse()?; keep_alive = Some(s.base10_parse()?); } else if inner.path.is_ident("session_expiry_interval") { - let value = inner.value()?; - let p: syn::LitInt = value.parse()?; - session_expiry_interval = Some(p.base10_parse()?); + let v = inner.value()?; + let expr: syn::Expr = v.parse()?; + session_expiry_interval = parse_mqtt_binding_value(expr)?; } else if inner.path.is_ident("maximum_packet_size") { - let value = inner.value()?; - let p: Path = value.parse()?; - maximum_packet_size = Some(p); + let v = inner.value()?; + let expr: syn::Expr = v.parse()?; + maximum_packet_size = parse_mqtt_binding_value(expr)?; } else if inner.path.is_ident("binding_version") { let value = inner.value()?; let s: syn::LitStr = value.parse()?; @@ -486,7 +481,7 @@ fn extract_mqtt_operation_bindings( } else if inner.path.is_ident("message_expiry_interval") { let value = inner.value()?; let expr: syn::Expr = value.parse()?; - message_expiry_interval = parse_message_expiry_interval(expr)?; + message_expiry_interval = parse_mqtt_binding_value(expr)?; } else if inner.path.is_ident("binding_version") { let value = inner.value()?; let s: LitStr = value.parse()?; diff --git a/asyncapi-rust-codegen/src/lib.rs b/asyncapi-rust-codegen/src/lib.rs index 5db9cd1..e0063e1 100644 --- a/asyncapi-rust-codegen/src/lib.rs +++ b/asyncapi-rust-codegen/src/lib.rs @@ -347,14 +347,14 @@ pub fn derive_to_asyncapi_message(input: TokenStream) -> TokenStream { quote! { None } }; let response_topic = match &mqtt.response_topic { - Some(crate::asyncapi_attrs::ResponseTopic::Uri(t)) => { + Some(crate::asyncapi_attrs::ResponseTopicMeta::Uri(t)) => { quote! { Some(asyncapi_rust::MqttReponseTopic::Uri( #t.to_string() )) } } - Some(crate::asyncapi_attrs::ResponseTopic::Reference(r)) => { + Some(crate::asyncapi_attrs::ResponseTopicMeta::Reference(r)) => { quote! { Some(asyncapi_rust::MqttReponseTopic::Schema({ let schema = schemars::schema_for!(#r); @@ -748,12 +748,29 @@ pub fn derive_asyncapi(input: TokenStream) -> TokenStream { } else { quote! { None } }; - let session_expiry_interval = if let Some(s) = mqtt.session_expiry_interval { - quote! { - Some(asyncapi_rust::Schema::Any(serde_json::json!(#s))) + + let session_expiry_interval = match &mqtt.session_expiry_interval { + Some(crate::asyncapi_spec_attrs::MqttBindingNumValueMeta::Value(t)) => { + quote! { + Some(asyncapi_rust::MqttBindingNumValue::Value( + #t + )) + } } - } else { - quote! { None } + Some(crate::asyncapi_spec_attrs::MqttBindingNumValueMeta::Reference(r)) => { + quote! { + Some(asyncapi_rust::MqttBindingNumValue::Schema({ + let schema = schemars::schema_for!(#r); + + let schema_json = serde_json::to_value(&schema) + .expect("Failed to serialize schema"); + + serde_json::from_value(schema_json) + .expect("Failed to deserialize schema") + })) + } + } + _ => quote! { None }, }; let keep_alive = if let Some(k) = mqtt.keep_alive { @@ -762,10 +779,28 @@ pub fn derive_asyncapi(input: TokenStream) -> TokenStream { quote! { None } }; - let max_packet_size = if let Some(s) = &mqtt.maximum_packet_size { - quote! { Some(#s) } - } else { - quote! { None } + let max_packet_size = match &mqtt.maximum_packet_size { + Some(crate::asyncapi_spec_attrs::MqttBindingNumValueMeta::Value(t)) => { + quote! { + Some(asyncapi_rust::MqttBindingNumValue::Value( + #t + )) + } + } + Some(crate::asyncapi_spec_attrs::MqttBindingNumValueMeta::Reference(r)) => { + quote! { + Some(asyncapi_rust::MqttBindingNumValue::Schema({ + let schema = schemars::schema_for!(#r); + + let schema_json = serde_json::to_value(&schema) + .expect("Failed to serialize schema"); + + serde_json::from_value(schema_json) + .expect("Failed to deserialize schema") + })) + } + } + _ => quote! { None }, }; quote! { @@ -941,20 +976,16 @@ pub fn derive_asyncapi(input: TokenStream) -> TokenStream { }; let message_expiry = match &mqtt.message_expiry_interval { - Some(crate::asyncapi_spec_attrs::MqttMessageExpiryIntervalMeta::Interval( - t, - )) => { + Some(crate::asyncapi_spec_attrs::MqttBindingNumValueMeta::Value(t)) => { quote! { - Some(asyncapi_rust::MqttMessageExpiryInterval::Interval( + Some(asyncapi_rust::MqttBindingNumValue::Value( #t )) } } - Some(crate::asyncapi_spec_attrs::MqttMessageExpiryIntervalMeta::Reference( - r, - )) => { + Some(crate::asyncapi_spec_attrs::MqttBindingNumValueMeta::Reference(r)) => { quote! { - Some(asyncapi_rust::MqttMessageExpiryInterval::Schema({ + Some(asyncapi_rust::MqttBindingNumValue::Schema({ let schema = schemars::schema_for!(#r); let schema_json = serde_json::to_value(&schema) diff --git a/asyncapi-rust-models/src/lib.rs b/asyncapi-rust-models/src/lib.rs index 5a3d211..df57c76 100644 --- a/asyncapi-rust-models/src/lib.rs +++ b/asyncapi-rust-models/src/lib.rs @@ -215,6 +215,16 @@ pub struct MqttLastWill { pub retain: bool, } +/// Represents mqtt binding properties which can be either a number or a json schema like sessionExpiryInterval or maximumPacketSize +#[derive(Debug, Clone, Serialize, Deserialize)] +#[serde(untagged)] +pub enum MqttBindingNumValue { + /// The value variant + Value(u32), + /// The schema variant + Schema(Schema), +} + /// Mqtt server binding #[derive(Debug, Clone, Serialize, Deserialize)] pub struct MqttServerBindings { @@ -240,11 +250,11 @@ pub struct MqttServerBindings { skip_serializing_if = "Option::is_none", rename = "sessionExpiryInterval" )] - pub session_expiry_interval: Option, + pub session_expiry_interval: Option, /// Number of bytes or a Schema Object representing the maximum packet size the client is willing to accept. #[serde(skip_serializing_if = "Option::is_none", rename = "maximumPacketSize")] - pub max_packet_size: Option, + pub max_packet_size: Option, /// The version of this binding. If omitted, "latest" MUST be assumed. #[serde(skip_serializing_if = "Option::is_none", rename = "bindingVersion")] @@ -614,16 +624,6 @@ pub struct OperationBindings { pub mqtt: Option, } -/// Interval in seconds or a Schema Object containing the definition of the lifetime of the message. -#[derive(Serialize, Deserialize, Clone, Debug)] -#[serde(untagged)] -pub enum MqttMessageExpiryInterval { - /// Interval in seconds - Interval(u32), - /// Schema Object containing the definition of the lifetime of the message - Schema(Schema), -} - /// Mqtt operation bindings #[derive(Serialize, Deserialize, Clone, Debug)] pub struct MqttOperationBindings { @@ -641,7 +641,7 @@ pub struct MqttOperationBindings { skip_serializing_if = "Option::is_none", rename = "messageExpiryInterval" )] - pub message_expiry_interval: Option, + pub message_expiry_interval: Option, /// The version of this binding. If omitted, "latest" MUST be assumed. #[serde(skip_serializing_if = "Option::is_none", rename = "bindingVersion")] diff --git a/asyncapi-rust/examples/mqtt_bindings.rs b/asyncapi-rust/examples/mqtt_bindings.rs index 5b94879..0418dcb 100644 --- a/asyncapi-rust/examples/mqtt_bindings.rs +++ b/asyncapi-rust/examples/mqtt_bindings.rs @@ -13,6 +13,18 @@ pub struct A { pub correlation_data: u8, } +#[derive(Serialize, Deserialize, JsonSchema)] +pub struct MinMax { + #[schemars(range(min = 30, max = 1200))] + value: u32, +} + +#[derive(Serialize, Deserialize, JsonSchema)] +pub struct Min { + #[schemars(range(min = 256))] + value: u32, +} + #[derive(Serialize, Deserialize, JsonSchema, ToAsyncApiMessage)] #[asyncapi( mqtt( @@ -47,9 +59,19 @@ pub struct MqttMessage; ), keep_alive = 60, session_expiry_interval = 3600, + maximum_packet_size = 1200 binding_version = "1.0" ) )] +#[asyncapi_server( + name = "dev", + host = "dev.api.example.com", + protocol = "mqtt", + mqtt( + session_expiry_interval = MinMax, + maximum_packet_size = Min + ) +)] #[asyncapi_channel(name = "test", address = "/test")] #[asyncapi_operation( name = "sendMqttMessage", From 259c52177c10405bcecdf537a2c9c8890d79ce7e Mon Sep 17 00:00:00 2001 From: Max Christoph Date: Wed, 15 Jul 2026 08:41:53 +0200 Subject: [PATCH 10/12] =?UTF-8?q?Renamed=20MqttReponseTopic=20=E2=86=92=20?= =?UTF-8?q?MqttResponseTopic?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- asyncapi-rust-codegen/src/lib.rs | 4 ++-- asyncapi-rust-models/src/lib.rs | 4 ++-- 2 files changed, 4 insertions(+), 4 deletions(-) diff --git a/asyncapi-rust-codegen/src/lib.rs b/asyncapi-rust-codegen/src/lib.rs index e0063e1..2f3c854 100644 --- a/asyncapi-rust-codegen/src/lib.rs +++ b/asyncapi-rust-codegen/src/lib.rs @@ -349,14 +349,14 @@ pub fn derive_to_asyncapi_message(input: TokenStream) -> TokenStream { let response_topic = match &mqtt.response_topic { Some(crate::asyncapi_attrs::ResponseTopicMeta::Uri(t)) => { quote! { - Some(asyncapi_rust::MqttReponseTopic::Uri( + Some(asyncapi_rust::MqttResponseTopic::Uri( #t.to_string() )) } } Some(crate::asyncapi_attrs::ResponseTopicMeta::Reference(r)) => { quote! { - Some(asyncapi_rust::MqttReponseTopic::Schema({ + Some(asyncapi_rust::MqttResponseTopic::Schema({ let schema = schemars::schema_for!(#r); let schema_json = serde_json::to_value(&schema) diff --git a/asyncapi-rust-models/src/lib.rs b/asyncapi-rust-models/src/lib.rs index df57c76..6886733 100644 --- a/asyncapi-rust-models/src/lib.rs +++ b/asyncapi-rust-models/src/lib.rs @@ -540,7 +540,7 @@ pub struct MessageBindings { /// Mqtt response topic #[derive(Serialize, Deserialize, Clone, Debug)] #[serde(untagged)] -pub enum MqttReponseTopic { +pub enum MqttResponseTopic { /// Topic Uri Uri(String), /// Schema for the the response topic @@ -567,7 +567,7 @@ pub struct MqttMessageBindings { /// The topic (channel URI) for a response message. #[serde(skip_serializing_if = "Option::is_none", rename = "responseTopic")] - pub response_topic: Option, + pub response_topic: Option, /// The version of this binding. If omitted, "latest" MUST be assumed. #[serde(skip_serializing_if = "Option::is_none", rename = "bindingVersion")] From 89329a506629b5542d71aaa055d87a6abf58cf9f Mon Sep 17 00:00:00 2001 From: Max Christoph Date: Thu, 23 Jul 2026 18:55:24 -0400 Subject: [PATCH 11/12] fix: address Copilot review feedback - Remove redundant & in example println! args (clippy -D warnings) - Add missing comma after maximum_packet_size in server mqtt(...) attr - Avoid unwrap() panics in example; print bindings safely with and_then - Fix stale comment referencing ChatMessage/SystemMessage - Fix duplicated word in MqttResponseTopic::Schema doc comment --- asyncapi-rust-models/src/lib.rs | 2 +- asyncapi-rust/examples/mqtt_bindings.rs | 13 ++++++++----- 2 files changed, 9 insertions(+), 6 deletions(-) diff --git a/asyncapi-rust-models/src/lib.rs b/asyncapi-rust-models/src/lib.rs index 6886733..d56a295 100644 --- a/asyncapi-rust-models/src/lib.rs +++ b/asyncapi-rust-models/src/lib.rs @@ -543,7 +543,7 @@ pub struct MessageBindings { pub enum MqttResponseTopic { /// Topic Uri Uri(String), - /// Schema for the the response topic + /// Schema for the response topic Schema(Schema), } diff --git a/asyncapi-rust/examples/mqtt_bindings.rs b/asyncapi-rust/examples/mqtt_bindings.rs index 0418dcb..8310da4 100644 --- a/asyncapi-rust/examples/mqtt_bindings.rs +++ b/asyncapi-rust/examples/mqtt_bindings.rs @@ -59,7 +59,7 @@ pub struct MqttMessage; ), keep_alive = 60, session_expiry_interval = 3600, - maximum_packet_size = 1200 + maximum_packet_size = 1200, binding_version = "1.0" ) )] @@ -100,18 +100,18 @@ fn main() { if let Some(desc) = &server.description { println!(" Description: {}", desc); } - println!(" bindings: {:?}", &server.bindings) + println!(" bindings: {:?}", server.bindings) } println!(); } - // Display messages (automatically populated from ChatMessage and SystemMessage) + // Display messages (automatically populated from MqttMessage) if let Some(components) = &spec.components { if let Some(messages) = &components.messages { println!("Messages (automatically included from message types):"); for (name, message) in messages { println!(" - {}", name); - if let Some(mqtt) = &message.bindings.as_ref().unwrap().mqtt { + if let Some(mqtt) = message.bindings.as_ref().and_then(|b| b.mqtt.as_ref()) { println!(" Mqtt Bindings: {:?}", mqtt); } } @@ -129,7 +129,10 @@ fn main() { }; println!(" • {} ({})", name, action); println!(" Channel: {}", operation.channel.reference); - println!(" Mqtt: {:?}", &operation.bindings.as_ref().unwrap().mqtt); + println!( + " Mqtt: {:?}", + operation.bindings.as_ref().and_then(|b| b.mqtt.as_ref()) + ); } println!(); } From ec4743bd09cfb3143bd657ca1c48d4cd8ecd1afb Mon Sep 17 00:00:00 2001 From: Max Christoph Date: Thu, 23 Jul 2026 19:48:26 -0400 Subject: [PATCH 12/12] test: assert emitted mqtt bindings; fix example responseTopic schema - Add tests/mqtt_bindings_integration.rs deriving mqtt(...) on a server, operation, and message and asserting spec.{servers,operations, components.messages}[..].bindings.mqtt (closes the assertion gap noted in review; the example only prints and unit tests only cover parsing) - Model example ExampleResponseTopic as a transparent String newtype so the MQTT responseTopic emits a string schema, not an object; anchor the pattern and fix the [a-z1-9] -> [a-z0-9] class --- asyncapi-rust/examples/mqtt_bindings.rs | 10 +- .../tests/mqtt_bindings_integration.rs | 108 ++++++++++++++++++ 2 files changed, 114 insertions(+), 4 deletions(-) create mode 100644 asyncapi-rust/tests/mqtt_bindings_integration.rs diff --git a/asyncapi-rust/examples/mqtt_bindings.rs b/asyncapi-rust/examples/mqtt_bindings.rs index 8310da4..1148344 100644 --- a/asyncapi-rust/examples/mqtt_bindings.rs +++ b/asyncapi-rust/examples/mqtt_bindings.rs @@ -2,11 +2,13 @@ use asyncapi_rust::{AsyncApi, ToAsyncApiMessage}; use schemars::JsonSchema; use serde::{Deserialize, Serialize}; +// MQTT responseTopic is a topic string; model it as a string schema (a +// transparent newtype) rather than an object so the emitted schema is faithful. #[derive(Serialize, Deserialize, JsonSchema)] -pub struct ExampleResponseTopic { - #[schemars(regex(pattern = "response/client/([a-z1-9]+)"))] - pub response: String, -} +#[schemars(transparent)] +pub struct ExampleResponseTopic( + #[schemars(regex(pattern = "^response/client/([a-z0-9]+)$"))] pub String, +); #[derive(Serialize, Deserialize, JsonSchema)] pub struct A { diff --git a/asyncapi-rust/tests/mqtt_bindings_integration.rs b/asyncapi-rust/tests/mqtt_bindings_integration.rs new file mode 100644 index 0000000..3de0f6c --- /dev/null +++ b/asyncapi-rust/tests/mqtt_bindings_integration.rs @@ -0,0 +1,108 @@ +//! Integration tests that derive a type with `mqtt(...)` bindings on a server, +//! operation, and message and assert the bindings appear in the generated +//! `AsyncApiSpec`. Unlike the `extract_*` unit tests (which only cover parsing) +//! and the `mqtt_bindings` example (which only prints), these assert the emitted +//! output, guarding against silent codegen regressions. + +use asyncapi_rust::{AsyncApi, ToAsyncApiMessage, schemars::JsonSchema}; +use serde::{Deserialize, Serialize}; + +#[derive(Serialize, Deserialize, JsonSchema, ToAsyncApiMessage)] +#[asyncapi(mqtt( + payload_format_indicator = 1, + content_type = "application/json", + binding_version = "1.0" +))] +pub struct SensorReading { + pub value: f64, +} + +#[derive(AsyncApi)] +#[asyncapi(title = "MQTT Binding Test API", version = "1.0.0")] +#[asyncapi_server( + name = "production", + host = "mqtt.example.com", + protocol = "mqtt", + mqtt( + client_id = "test-client", + clean_session = false, + keep_alive = 60, + session_expiry_interval = 3600, + maximum_packet_size = 1200, + binding_version = "1.0" + ) +)] +#[asyncapi_channel(name = "sensors", address = "/sensors")] +#[asyncapi_operation( + name = "publishReading", + action = "send", + channel = "sensors", + mqtt( + qos = 2, + retain = true, + message_expiry_interval = 300, + binding_version = "1.0" + ) +)] +#[asyncapi_messages(SensorReading)] +pub struct MqttBindingApi; + +#[test] +fn server_mqtt_bindings_emitted() { + let spec = MqttBindingApi::asyncapi_spec(); + let servers = spec.servers.expect("servers present"); + let server = servers + .get("production") + .expect("production server present"); + let mqtt = server + .bindings + .as_ref() + .and_then(|b| b.mqtt.as_ref()) + .expect("server mqtt bindings present"); + + assert_eq!(mqtt.client_id.as_deref(), Some("test-client")); + assert_eq!(mqtt.clean_session, Some(false)); + assert_eq!(mqtt.keep_alive, Some(60)); + assert_eq!(mqtt.binding_version.as_deref(), Some("1.0")); + // Numeric-or-schema fields: assert they were carried through. + assert!(mqtt.session_expiry_interval.is_some()); + assert!(mqtt.max_packet_size.is_some()); +} + +#[test] +fn operation_mqtt_bindings_emitted() { + let spec = MqttBindingApi::asyncapi_spec(); + let operations = spec.operations.expect("operations present"); + let op = operations + .get("publishReading") + .expect("publishReading operation present"); + let mqtt = op + .bindings + .as_ref() + .and_then(|b| b.mqtt.as_ref()) + .expect("operation mqtt bindings present"); + + assert_eq!(mqtt.qos, Some(2)); + assert_eq!(mqtt.retain, Some(true)); + assert_eq!(mqtt.binding_version.as_deref(), Some("1.0")); + assert!(mqtt.message_expiry_interval.is_some()); +} + +#[test] +fn message_mqtt_bindings_emitted() { + let spec = MqttBindingApi::asyncapi_spec(); + let components = spec.components.expect("components present"); + let messages = components.messages.expect("messages present"); + let message = messages + .get("SensorReading") + .expect("SensorReading message present"); + let mqtt = message + .bindings + .as_ref() + .and_then(|b| b.mqtt.as_ref()) + .expect("message mqtt bindings present"); + + assert_eq!(mqtt.payload_format_indicator, Some(1)); + assert_eq!(mqtt.content_type.as_deref(), Some("application/json")); + assert_eq!(mqtt.binding_version.as_deref(), Some("1.0")); +}