fe2o3-amqp-types

实现了AMQP1.0数据类型，如核心规范中所定义。

功能标志请注意，Performative 将需要同时启用 "transport" 和 "messaging" 功能标志。

"primitive": 启用核心规范第1.6部分中定义的原始类型。
"transport": 启用核心规范第2.4、2.5和2.8部分中定义的大多数类型。
"messaging": 启用核心规范第2.7和第3部分中定义的类型。
"transaction": 启用核心规范第4.5部分中定义的类型。
"security": 启用核心规范第5部分中定义的类型。

default = [
    "primitive",
    "transport",
    "messaging",
    "security",
]

许可证：MIT/Apache-2.0

完整示例代码：

// Cargo.toml
// [dependencies]
// fe2o3-amqp-types = "0.14.0"

use fe2o3_amqp_types::{
    definitions::Fields,
    messaging::{Message, Properties},
    primitives::Binary,
    performative::Performative,
};

fn main() {
    // 创建一个简单的消息
    let properties = Properties::builder()
        .message_id("msg-1".into())
        .content_type("text/plain".into())
        .build();
    
    let message_body = Binary::from("Hello AMQP!".as_bytes());
    
    let message = Message::builder()
        .properties(properties)
        .value(message_body)
        .build();
    
    println!("Created message: {:?}", message);
    
    // 创建一个Performative示例（需要启用transport和messaging功能）
    let performative = Performative::Attach(Fields::default());
    println!("Performative: {:?}", performative);
}

// 启用特定功能的配置示例
// [dependencies]
// fe2o3-amqp-types = { version = "0.14.0", features = ["primitive", "transport", "messaging"] }

完整示例demo：

// Cargo.toml
// [dependencies]
// fe2o3-amqp-types = { version = "0.14.0", features = ["primitive", "transport", "messaging"] }

use fe2o3_amqp_types::{
    definitions::Fields,
    messaging::{Message, Properties},
    primitives::Binary,
    performative::Performative,
};

fn main() {
    // 创建一个简单的消息
    let properties = Properties::builder()
        .message_id("msg-1".into())  // 设置消息ID
        .content_type("text/plain".into())  // 设置内容类型
        .build();  // 构建属性
    
    let message_body = Binary::from("Hello AMQP!".as_bytes());  // 创建二进制消息体
    
    let message = Message::builder()
        .properties(properties)  // 设置消息属性
        .value(message_body)  // 设置消息值
        .build();  // 构建消息
    
    println!("Created message: {:?}", message);  // 打印创建的消息
    
    // 创建一个Performative示例（需要启用transport和messaging功能）
    let performative = Performative::Attach(Fields::default());  // 创建Attach类型的Performative
    println!("Performative: {:?}", performative);  // 打印Performative
}

ionicwang 1楼

Rust AMQP协议实现库fe2o3-amqp-types的使用指南

概述

fe2o3-amqp-types是一个专门为Rust语言设计的AMQP（高级消息队列协议）类型处理库。该库提供了对AMQP协议核心数据结构和消息格式的高效实现，支持AMQP 1.0协议规范，专注于类型安全和性能优化。

主要特性

完整的AMQP 1.0协议类型系统实现
零拷贝序列化和反序列化
强类型保证和编译时检查
与tokio和async-std异步运行时兼容
支持no_std环境（有限功能）

安装方法

在Cargo.toml中添加依赖：

[dependencies]
fe2o3-amqp-types = "0.1"

基本使用方法

1. 基本类型使用示例

use fe2o3_amqp_types::primitives::Value;
use fe2o3_amqp_types::messaging::{Message, Body};

// 创建AMQP消息
let message = Message::builder()
    .body(Body::Value(Value::String("Hello AMQP!".into())))
    .build();

// 序列化消息
let encoded = message.encode().unwrap();

// 反序列化消息
let decoded: Message<Body> = Message::decode(&encoded[..]).unwrap();

2. 复杂类型处理

use fe2o3_amqp_types::primitives::{Value, OrderedMap};
use fe2o3_amqp_types::messaging::{Message, Properties};

// 创建带有属性的消息
let properties = Properties::builder()
    .message_id("12345".into())
    .content_type("text/plain".into())
    .build();

let mut application_properties = OrderedMap::new();
application_properties.insert("priority".into(), Value::Int(1));
application_properties.insert("user_id".into(), Value::String("user123".into()));

let message = Message::builder()
    .properties(properties)
    .application_properties(application_properties)
    .body(Body::Value(Value::String("Important message".into())))
    .build();

3. 自定义消息体处理

use fe2o3_amqp_types::messaging::{Message, Body, Data};
use bytes::Bytes;

// 使用Data主体（二进制数据）
let binary_data = Bytes::from_static(b"raw binary data");
let message = Message::builder()
    .body(Body::Data(Data::from(binary_data)))
    .build();

// 使用序列主体（多个数据段）
let data_segments = vec![
    Data::from(Bytes::from("chunk1")),
    Data::from(Bytes::from("chunk2")),
];
let message = Message::builder()
    .body(Body::Sequence(data_segments))
    .build();

4. 错误处理示例

use fe2o3_amqp_types::{AmqpError, decoding::Error};

fn process_message(data: &[u8]) -> Result<Message<Body>, Error> {
    match Message::decode(data) {
        Ok(message) => Ok(message),
        Err(e) => {
            eprintln!("Failed to decode message: {}", e);
            Err(e)
        }
    }
}

高级用法

自定义类型序列化

use fe2o3_amqp_types::primitives::Value;
use serde::{Serialize, Deserialize};

#[derive(Serialize, Deserialize)]
struct CustomData {
    id: u32,
    name: String,
    tags: Vec<String>,
}

// 将自定义类型转换为AMQP Value
let custom_data = CustomData {
    id: 42,
    name: "test".to_string(),
    tags: vec!["important".to_string(), "urgent".to_string()],
};

let value = Value::from_serialize(&custom_data).unwrap();

完整示例demo

use fe2o3_amqp_types::primitives::{Value, OrderedMap};
use fe2o3_amqp_types::messaging::{Message, Body, Properties, Data};
use fe2o3_amqp_types::decoding::Error;
use bytes::Bytes;
use serde::{Serialize, Deserialize};

// 自定义数据结构
#[derive(Serialize, Deserialize, Debug)]
struct UserEvent {
    user_id: String,
    event_type: String,
    timestamp: i64,
    metadata: Vec<String>,
}

fn main() -> Result<(), Error> {
    // 示例1：创建基本消息
    println!("=== 基本消息示例 ===");
    let simple_message = Message::builder()
        .body(Body::Value(Value::String("Hello AMQP!".into())))
        .build();
    
    let encoded = simple_message.encode()?;
    let decoded: Message<Body> = Message::decode(&encoded[..])?;
    println!("基本消息解码成功: {:?}", decoded.body());

    // 示例2：创建复杂消息
    println!("\n=== 复杂消息示例 ===");
    let properties = Properties::builder()
        .message_id("msg_001".into())
        .content_type("application/json".into())
        .group_id("user_events".into())
        .build();

    let mut app_properties = OrderedMap::new();
    app_properties.insert("priority".into(), Value::Int(5));
    app_properties.insert("source".into(), Value::String("web_api".into()));

    let user_event = UserEvent {
        user_id: "user_123".to_string(),
        event_type: "login".to_string(),
        timestamp: 1672531200,
        metadata: vec!["browser:chrome".to_string(), "os:windows".to_string()],
    };

    let event_value = Value::from_serialize(&user_event)?;
    
    let complex_message = Message::builder()
        .properties(properties)
        .application_properties(app_properties)
        .body(Body::Value(event_value))
        .build();

    let encoded_complex = complex_message.encode()?;
    let decoded_complex: Message<Body> = Message::decode(&encoded_complex[..])?;
    println!("复杂消息解码成功");

    // 示例3：二进制数据处理
    println!("\n=== 二进制数据示例 ===");
    let image_data = Bytes::from_static(b"fake_image_data_here");
    let binary_message = Message::builder()
        .body(Body::Data(Data::from(image_data)))
        .build();

    let encoded_binary = binary_message.encode()?;
    let decoded_binary: Message<Body> = Message::decode(&encoded_binary[..])?;
    println!("二进制消息解码成功");

    // 示例4：错误处理演示
    println!("\n=== 错误处理示例 ===");
    let invalid_data = b"invalid_amqp_data";
    match Message::decode(invalid_data) {
        Ok(_) => println!("意外成功"),
        Err(e) => println!("预期错误: {}", e),
    }

    Ok(())
}

// 消息处理函数示例
fn process_amqp_message(data: &[u8]) -> Result<Message<Body>, Error> {
    let message = Message::decode(data)?;
    
    // 根据消息类型进行不同处理
    match message.body() {
        Body::Value(value) => {
            println!("收到值类型消息: {:?}", value);
            // 可以尝试反序列化为自定义类型
            if let Ok(user_event) = value.to_deserialize::<UserEvent>() {
                println!("解析为用户事件: {:?}", user_event);
            }
        }
        Body::Data(data) => {
            println!("收到二进制数据，长度: {} bytes", data.len());
        }
        Body::Sequence(seq) => {
            println!("收到序列数据，包含 {} 个片段", seq.len());
        }
        _ => println!("收到其他类型消息"),
    }
    
    Ok(message)
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn test_message_roundtrip() {
        let original = Message::builder()
            .body(Body::Value(Value::String("test".into())))
            .build();
        
        let encoded = original.encode().unwrap();
        let decoded = Message::decode(&encoded[..]).unwrap();
        
        assert!(matches!(decoded.body(), Body::Value(Value::String(s)) if s == "test"));
    }
}

性能优化建议

重用MessageBuilder以减少内存分配
使用Bytes类型处理二进制数据以避免拷贝
在热路径中使用预分配缓冲区
考虑使用no_std版本以减小二进制大小

注意事项

确保正确处理AMQP协议版本兼容性
注意消息大小限制（默认最大帧大小）
合理处理消息确认和错误恢复机制

这个库为Rust开发者提供了处理AMQP协议消息的强大工具，特别适合需要高性能消息处理的分布式系统和消息队列应用场景。

Rust AMQP协议实现库fe2o3-amqp-types的使用，支持高级消息队列协议的高效类型处理