diff --git a/asyncapi-rust-codegen/src/asyncapi_spec_attrs.rs b/asyncapi-rust-codegen/src/asyncapi_spec_attrs.rs index e14f16c..4012f5e 100644 --- a/asyncapi-rust-codegen/src/asyncapi_spec_attrs.rs +++ b/asyncapi-rust-codegen/src/asyncapi_spec_attrs.rs @@ -90,6 +90,7 @@ pub struct OperationMeta { pub channel: String, #[allow(dead_code)] // Reserved for future use pub description: Option, + pub messages: Vec, pub mqtt: Option, } @@ -518,10 +519,14 @@ fn extract_mqtt_operation_bindings( /// Extract operation metadata from `#[asyncapi_operation(...)]` attribute fn extract_operation(attr: &Attribute) -> syn::Result> { + use syn::Token; + use syn::punctuated::Punctuated; + let mut name = None; let mut action = None; let mut channel = None; let mut description = None; + let mut messages = Vec::new(); let mut mqtt = None; attr.parse_nested_meta(|nested| { @@ -541,6 +546,14 @@ fn extract_operation(attr: &Attribute) -> syn::Result> { let value = nested.value()?; let s: syn::LitStr = value.parse()?; description = Some(s.value()); + } else if nested.path.is_ident("messages") { + // Parse array of type paths: messages = [Type1, Type2, ...] + let _ = nested.value()?; // Parse the equals sign and prepare for value parsing + let content; + syn::bracketed!(content in nested.input); + let types: Punctuated = + content.parse_terminated(|stream| stream.parse(), Token![,])?; + messages = types.into_iter().collect(); } else if nested.path.is_ident("mqtt") { mqtt = extract_mqtt_operation_bindings(&nested)?; } @@ -556,6 +569,7 @@ fn extract_operation(attr: &Attribute) -> syn::Result> { action, channel, description, + messages, mqtt, })) } @@ -943,4 +957,53 @@ mod tests { assert_eq!(op.name, "sendMessage"); assert_eq!(op.description, Some("Send a chat message".to_string())); } + + #[test] + fn test_extract_operation_with_messages() { + let attrs: Vec = vec![parse_quote! { + #[asyncapi_operation(name = "sendMessage", action = "send", channel = "chat", messages = [ChatMessage])] + }]; + + let meta = extract_asyncapi_spec_meta(&attrs).unwrap(); + 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_eq!(meta.operations[0].messages.len(), 1); + let path0 = &meta.operations[0].messages[0]; + assert_eq!(quote!(#path0).to_string(), "ChatMessage"); + } + + #[test] + fn test_extract_operation_with_multiple_messages() { + let attrs: Vec = vec![parse_quote! { + #[asyncapi_operation(name = "sendMessage", action = "send", channel = "chat", messages = [ChatMessage, SystemMessage])] + }]; + + let meta = extract_asyncapi_spec_meta(&attrs).unwrap(); + assert_eq!(meta.operations.len(), 1); + assert_eq!(meta.operations[0].messages.len(), 2); + let path0 = &meta.operations[0].messages[0]; + let path1 = &meta.operations[0].messages[1]; + assert_eq!(quote!(#path0).to_string(), "ChatMessage"); + assert_eq!(quote!(#path1).to_string(), "SystemMessage"); + } + + #[test] + fn test_extract_operation_with_module_path_messages() { + let attrs: Vec = vec![parse_quote! { + #[asyncapi_operation(name = "sendMessage", action = "send", channel = "chat", messages = [super::messages::ChatMessage, crate::SystemMessage])] + }]; + + let meta = extract_asyncapi_spec_meta(&attrs).unwrap(); + assert_eq!(meta.operations.len(), 1); + assert_eq!(meta.operations[0].messages.len(), 2); + let path0 = &meta.operations[0].messages[0]; + let path1 = &meta.operations[0].messages[1]; + assert_eq!( + quote!(#path0).to_string(), + "super :: messages :: ChatMessage" + ); + assert_eq!(quote!(#path1).to_string(), "crate :: SystemMessage"); + } } diff --git a/asyncapi-rust-codegen/src/lib.rs b/asyncapi-rust-codegen/src/lib.rs index 3f0368a..791bf0f 100644 --- a/asyncapi-rust-codegen/src/lib.rs +++ b/asyncapi-rust-codegen/src/lib.rs @@ -75,13 +75,16 @@ //! #[asyncapi_operation( //! name = "sendMessage", //! action = "send", -//! channel = "chat" +//! channel = "chat", +//! messages = [ChatMessage] //! )] //! #[asyncapi_operation( //! name = "receiveMessage", //! action = "receive", -//! channel = "chat" +//! channel = "chat", +//! messages = [ChatMessage, SystemMessage] //! )] +//! #[asyncapi_messages(ChatMessage, SystemMessage)] //! struct ChatApi; //! //! // Generated method: @@ -131,6 +134,12 @@ //! - `name = "..."` - Operation identifier (required) //! - `action = "send"|"receive"` - Operation type (required) //! - `channel = "..."` - Channel reference (required) +//! - `messages = [Type1, Type2, ...]` - Message types available for this operation (optional) +//! +//! When the `messages` parameter is specified on operations, those messages are automatically +//! added to the channel that the operation references. Operation messages reference the channel's +//! messages (e.g., `#/channels/{channel}/messages/{message}`), while channel messages reference +//! the components section (e.g., `#/components/messages/{message}`), following AsyncAPI 3.0 spec. //! //! ## Integration with serde //! @@ -503,6 +512,8 @@ pub fn derive_to_asyncapi_message(input: TokenStream) -> TokenStream { use schemars::schema_for; let schema = schema_for!(Self); + + // Convert schemars RootSchema to JSON let schema_json = serde_json::to_value(&schema) .expect("Failed to serialize schema"); @@ -922,12 +933,54 @@ pub fn derive_asyncapi(input: TokenStream) -> TokenStream { } }; + // Collect messages from all operations that reference this channel + let channel_name_str = name.as_str(); + let operations_for_channel: Vec<_> = spec_meta + .operations + .iter() + .filter(|op| op.channel == channel_name_str) + .collect(); + + let messages_field = if operations_for_channel.is_empty() + || operations_for_channel + .iter() + .all(|op| op.messages.is_empty()) + { + quote! { None } + } else { + let message_calls: Vec<_> = operations_for_channel + .iter() + .flat_map(|op| &op.messages) // Deduplicate + .map(|type_name| { + quote! { + // Call asyncapi_message_names() for this type and add references + for msg_name in #type_name::asyncapi_message_names() { + channel_messages.insert( + msg_name.to_string(), + asyncapi_rust::MessageRef::Reference { + reference: format!("#/components/messages/{}", msg_name), + } + ); + } + } + }) + .collect(); + + quote! { + { + let mut channel_messages = asyncapi_rust::indexmap::IndexMap::new(); + #(#message_calls)* + Some(channel_messages) + } + } + }; + quote! { channels.insert( #name.to_string(), asyncapi_rust::Channel { address: #address, - messages: None, + messages: #messages_field, parameters: #parameters, } ); @@ -952,6 +1005,22 @@ pub fn derive_asyncapi(input: TokenStream) -> TokenStream { let channel_ref = &operation.channel; let action = &operation.action; + // Validate that the operation references a declared channel + if !spec_meta + .channels + .iter() + .any(|channel| channel.name == *channel_ref) + { + return syn::Error::new_spanned( + channel_ref, + format!( + "Operation '{}' references unknown channel '{}'", + name, channel_ref + ), + ) + .to_compile_error(); + } + // Convert action string to OperationAction enum let action_enum = if action == "send" { quote! { asyncapi_rust::OperationAction::Send } @@ -965,6 +1034,44 @@ pub fn derive_asyncapi(input: TokenStream) -> TokenStream { .to_compile_error(); }; + // Generate messages references if any messages are specified + let messages_field = if operation.messages.is_empty() { + quote! { None } + } else { + let message_calls: Vec<_> = operation + .messages + .iter() + .map(|type_name| { + quote! { + for msg_name in #type_name::asyncapi_message_names() { + message_refs + .entry(msg_name.to_string()) + .or_insert_with(|| { + asyncapi_rust::MessageRef::Reference { + reference: format!( + "#/channels/{}/messages/{}", + #channel_ref, + msg_name + ), + } + }); + } + } + }) + .collect(); + + quote! { + { + let mut message_refs = + asyncapi_rust::indexmap::IndexMap::new(); + + #(#message_calls)* + + Some(message_refs.into_values().collect()) + } + } + }; + let bindings = if let Some(mqtt) = &operation.mqtt { let qos = if let Some(s) = mqtt.qos { quote! { Some(#s) } @@ -1030,7 +1137,7 @@ pub fn derive_asyncapi(input: TokenStream) -> TokenStream { channel: asyncapi_rust::ChannelRef { reference: format!("#/channels/{}", #channel_ref), }, - messages: None, + messages: #messages_field, bindings: #bindings } ); @@ -1047,10 +1154,22 @@ pub fn derive_asyncapi(input: TokenStream) -> TokenStream { }; // Generate components with messages and hoisted shared schemas - let components_code = if spec_meta.message_types.is_empty() { + let mut all_message_types = spec_meta.message_types.clone(); + + for type_name in spec_meta + .operations + .iter() + .flat_map(|operation| operation.messages.iter()) + { + if !all_message_types.contains(type_name) { + all_message_types.push(type_name.clone()); + } + } + + let components_code = if all_message_types.is_empty() { quote! { None } } else { - let type_calls = spec_meta.message_types.iter().map(|type_name| { + let type_calls = all_message_types.iter().map(|type_name| { quote! { for msg in #type_name::asyncapi_messages() { if let Some(ref name) = msg.name { diff --git a/asyncapi-rust/examples/message_integration.rs b/asyncapi-rust/examples/message_integration.rs index 5f53e6b..bc4a82f 100644 --- a/asyncapi-rust/examples/message_integration.rs +++ b/asyncapi-rust/examples/message_integration.rs @@ -10,6 +10,12 @@ use asyncapi_rust::{AsyncApi, ToAsyncApiMessage, schemars::JsonSchema}; use serde::{Deserialize, Serialize}; +#[derive(Serialize, Deserialize, JsonSchema)] +pub struct UserJoin { + pub username: String, + pub room: String, +} + /// Chat messages for a chat application #[derive(Serialize, Deserialize, JsonSchema, ToAsyncApiMessage)] #[serde(tag = "type")] @@ -59,6 +65,12 @@ pub enum SystemMessage { /// /// The #[asyncapi_messages(...)] attribute automatically includes /// message definitions from the specified types in the components section. +/// The messages parameter in operations specifies which messages can be used, +/// and these messages are automatically added to the channel that the operation references. +/// +/// Per AsyncAPI 3.0 spec: +/// - Operations reference channel messages: #/channels/{channel}/messages/{message} +/// - Channels reference component messages: #/components/messages/{message} #[allow(clippy::duplicated_attributes)] #[derive(AsyncApi)] #[asyncapi( @@ -73,8 +85,8 @@ pub enum SystemMessage { description = "Production WebSocket server" )] #[asyncapi_channel(name = "chat", address = "/ws/chat")] -#[asyncapi_operation(name = "sendMessage", action = "send", channel = "chat")] -#[asyncapi_operation(name = "receiveMessage", action = "receive", channel = "chat")] +#[asyncapi_operation(name = "sendMessage", action = "send", channel = "chat", messages = [ChatMessage])] +#[asyncapi_operation(name = "receiveMessage", action = "receive", channel = "chat", messages = [ChatMessage, SystemMessage])] #[asyncapi_messages(ChatMessage, SystemMessage)] struct ChatApi; @@ -107,6 +119,14 @@ fn main() { if let Some(address) = &channel.address { println!(" - {}: {}", name, address); } + if let Some(messages) = &channel.messages { + println!(" Channel Messages:"); + for (msg_name, msg_ref) in messages { + if let asyncapi_rust::MessageRef::Reference { reference } = msg_ref { + println!(" - {}: {}", msg_name, reference); + } + } + } } println!(); } @@ -123,6 +143,14 @@ fn main() { " - {}: {} to {}", name, action, operation.channel.reference ); + if let Some(messages) = &operation.messages { + println!(" Messages:"); + for msg in messages { + if let asyncapi_rust::MessageRef::Reference { reference } = msg { + println!(" - {}", reference); + } + } + } } println!(); } diff --git a/asyncapi-rust/src/lib.rs b/asyncapi-rust/src/lib.rs index b47109a..5288dc7 100644 --- a/asyncapi-rust/src/lib.rs +++ b/asyncapi-rust/src/lib.rs @@ -42,8 +42,9 @@ //! #[asyncapi(title = "Chat API", version = "1.0.0")] //! #[asyncapi_server(name = "production", host = "api.example.com", protocol = "wss")] //! #[asyncapi_channel(name = "chat", address = "/ws/chat")] -//! #[asyncapi_operation(name = "sendMessage", action = "send", channel = "chat")] -//! #[asyncapi_operation(name = "receiveMessage", action = "receive", channel = "chat")] +//! #[asyncapi_operation(name = "sendMessage", action = "send", channel = "chat", messages = [ChatMessage])] +//! #[asyncapi_operation(name = "receiveMessage", action = "receive", channel = "chat", messages = [ChatMessage])] +//! #[asyncapi_messages(ChatMessage)] //! struct ChatApi; //! //! fn main() { @@ -76,7 +77,13 @@ //! - `#[asyncapi(...)]` - Basic info (title, version, description) //! - `#[asyncapi_server(...)]` - Server definitions //! - `#[asyncapi_channel(...)]` - Channel definitions -//! - `#[asyncapi_operation(...)]` - Operation definitions +//! - `#[asyncapi_operation(...)]` - Operation definitions (with optional `messages` parameter) +//! - `#[asyncapi_messages(...)]` - Include message types in components +//! +//! When you specify messages in operations, they are automatically added to the channel +//! that the operation references. Operations reference channel messages +//! (`#/channels/{channel}/messages/{message}`), while channels reference components +//! (`#/components/messages/{message}`), following AsyncAPI 3.0 specification. //! //! ## Framework Integration //! diff --git a/asyncapi-rust/tests/integration_test.rs b/asyncapi-rust/tests/integration_test.rs index 3b69d98..c5b541c 100644 --- a/asyncapi-rust/tests/integration_test.rs +++ b/asyncapi-rust/tests/integration_test.rs @@ -441,6 +441,94 @@ fn test_asyncapi_with_messages() { assert_eq!(system_status.summary, Some("System status".to_string())); } +#[test] +fn test_asyncapi_operation_with_messages() { + // Define message types for operations + #[derive(Serialize, Deserialize, JsonSchema, ToAsyncApiMessage)] + #[serde(tag = "type")] + pub enum ChatMessage { + #[serde(rename = "user.join")] + UserJoin { username: String, room: String }, + #[serde(rename = "chat.message")] + ChatMessage { username: String, text: String }, + } + + #[derive(Serialize, Deserialize, JsonSchema, ToAsyncApiMessage)] + #[serde(tag = "type")] + pub enum SystemMessage { + #[serde(rename = "system.status")] + Status { status: String }, + } + + // Define API with operations that specify messages + #[allow(clippy::duplicated_attributes)] + #[derive(AsyncApi)] + #[asyncapi(title = "Chat API", version = "1.0.0")] + #[asyncapi_channel(name = "chat", address = "/ws/chat")] + #[asyncapi_operation(name = "sendMessage", action = "send", channel = "chat", messages = [ChatMessage])] + #[asyncapi_operation(name = "receiveMessage", action = "receive", channel = "chat", messages = [ChatMessage, SystemMessage])] + #[asyncapi_messages(ChatMessage, SystemMessage)] + struct ChatApi; + + let spec = ChatApi::asyncapi_spec(); + + println!("{}", serde_json::to_string_pretty(&spec).unwrap()); + + // Verify operations exist + let operations = spec.operations.expect("Should have operations"); + assert_eq!(operations.len(), 2); + + // Verify sendMessage operation has messages + let send_op = operations + .get("sendMessage") + .expect("Should have sendMessage operation"); + assert!(send_op.messages.is_some()); + let send_messages = send_op.messages.as_ref().unwrap(); + assert_eq!(send_messages.len(), 2); // ChatMessage has 2 variants + + // Verify receiveMessage operation has messages + let receive_op = operations + .get("receiveMessage") + .expect("Should have receiveMessage operation"); + assert!(receive_op.messages.is_some()); + let receive_messages = receive_op.messages.as_ref().unwrap(); + assert_eq!(receive_messages.len(), 3); // ChatMessage (2 variants) + SystemMessage (1 variant) + + // Verify operation message references point to channel messages (not components directly) + match &send_messages[0] { + asyncapi_rust::MessageRef::Reference { reference } => { + println!("{}", reference); + assert!( + reference == "#/channels/chat/messages/UserJoin" + || reference == "#/channels/chat/messages/ChatMessage" + ); + } + _ => panic!("Expected message reference"), + } + + // Verify channels exist and have messages + let channels = spec.channels.expect("Should have channels"); + assert_eq!(channels.len(), 1); + + let chat_channel = channels.get("chat").expect("Should have chat channel"); + assert!(chat_channel.messages.is_some()); + let channel_messages = chat_channel.messages.as_ref().unwrap(); + assert_eq!(channel_messages.len(), 3); // All unique messages from both operations + + // Verify channel messages reference components directly + assert!(channel_messages.contains_key("UserJoin")); + assert!(channel_messages.contains_key("ChatMessage")); + assert!(channel_messages.contains_key("Status")); + + // Verify channel messages reference components (not other channels) + match channel_messages.get("UserJoin").unwrap() { + asyncapi_rust::MessageRef::Reference { reference } => { + assert_eq!(reference, "#/components/messages/UserJoin"); + } + _ => panic!("Expected message reference"), + } +} + // Regression test for issue #4: ToAsyncApiMessage panics on enums with serde_json::Value fields #[test] fn test_enum_with_json_value_fields() {