Derive channel and operation messages

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<String>,
+    pub messages: Vec<Path>,
     pub mqtt: Option<OperationMqttBindingsMeta>,
 }
 
@@ -518,10 +519,14 @@ fn extract_mqtt_operation_bindings(
 
 /// Extract operation metadata from `#[asyncapi_operation(...)]` attribute
 fn extract_operation(attr: &Attribute) -> syn::Result<Option<OperationMeta>> {
+    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<Option<OperationMeta>> {
             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<Path, Token![,]> =
+                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<Option<OperationMeta>> {
         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<Attribute> = 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<Attribute> = 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<Attribute> = 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");
+    }
 }

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 {

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!();
     }

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
 //!