flare_core/common/message/
parser.rs

1//! 消息解析模块
2//!
3//! 负责将原始字节数据解析为 Frame 消息
4//! 使用压缩器、序列化器和加密器模块的标准接口，支持自动检测和扩展
5
6use crate::common::compression::{CompressionAlgorithm, CompressionUtil};
7use crate::common::encryption::{EncryptionAlgorithm, EncryptionUtil};
8use crate::common::error::Result;
9use crate::common::protocol::{Frame, SerializationFormat};
10use crate::common::serializer::SerializationUtil;
11use lazy_static::lazy_static;
12
13/// 小包压缩阈值。
14///
15/// 移动 IM 的心跳、ACK、已读等小帧通常低于该值；压缩这些帧会额外消耗 CPU，
16/// gzip 头部还可能让包体变大。超过该阈值才应用协商出的压缩算法。
17pub const MIN_COMPRESSION_PAYLOAD_BYTES: usize = 512;
18
19// 协商前的消息解析器（全局共享）
20// 所有连接在协商完成前都使用相同的配置：JSON、不压缩、不加密
21// 使用场景：CONNECT、CONNECT_ACK、NEGOTIATION_READY 消息的解析和序列化
22// 使用 lazy_static 实现全局单例，避免每次消息处理都创建新的 parser
23lazy_static! {
24    pub static ref PRE_NEGOTIATION_PARSER: MessageParser = MessageParser::new(
25        SerializationFormat::Json,
26        CompressionAlgorithm::None,
27        EncryptionAlgorithm::None,
28    );
29}
30
31/// 消息解析器
32#[derive(Debug, Clone)]
33pub struct MessageParser {
34    default_format: SerializationFormat,
35    default_compression: CompressionAlgorithm,
36    default_encryption: EncryptionAlgorithm,
37    /// 自定义序列化格式名称（可选）
38    ///
39    /// 当使用自定义序列化格式时，通过此字段指定格式名称
40    /// 如果设置了此字段，序列化/反序列化时会优先使用名称查找序列化器
41    /// 否则使用 `default_format` 对应的内置序列化器
42    custom_format_name: Option<String>,
43}
44
45impl MessageParser {
46    /// 创建新的消息解析器
47    pub fn new(
48        format: SerializationFormat,
49        compression: CompressionAlgorithm,
50        encryption: EncryptionAlgorithm,
51    ) -> Self {
52        Self {
53            default_format: format,
54            default_compression: compression,
55            default_encryption: encryption,
56            custom_format_name: None,
57        }
58    }
59
60    /// 创建使用自定义序列化格式的解析器
61    ///
62    /// # 参数
63    /// - `format_name`: 自定义序列化格式名称（必须在注册表中注册）
64    /// - `compression`: 压缩算法
65    /// - `encryption`: 加密算法
66    ///
67    /// # 示例
68    /// ```rust
69    /// use flare_core::common::message::MessageParser;
70    /// use flare_core::common::compression::CompressionAlgorithm;
71    /// use flare_core::common::encryption::EncryptionAlgorithm;
72    ///
73    /// // 创建使用自定义格式的解析器
74    /// let parser = MessageParser::with_custom_format(
75    ///     "messagepack",
76    ///     CompressionAlgorithm::None,
77    ///     EncryptionAlgorithm::None,
78    /// );
79    /// ```
80    pub fn with_custom_format(
81        format_name: &str,
82        compression: CompressionAlgorithm,
83        encryption: EncryptionAlgorithm,
84    ) -> Self {
85        Self {
86            default_format: SerializationFormat::Json, // 占位符，实际使用 custom_format_name
87            default_compression: compression,
88            default_encryption: encryption,
89            custom_format_name: Some(format_name.to_string()),
90        }
91    }
92
93    /// 创建使用指定格式和压缩的解析器
94    pub fn new_with_format_compression(
95        format: SerializationFormat,
96        compression: CompressionAlgorithm,
97    ) -> Self {
98        Self::new(format, compression, EncryptionAlgorithm::None)
99    }
100
101    /// 创建使用 Protobuf 格式的解析器
102    pub fn protobuf() -> Self {
103        Self::new(
104            SerializationFormat::Protobuf,
105            CompressionAlgorithm::None,
106            EncryptionAlgorithm::None,
107        )
108    }
109
110    /// 创建使用 JSON 格式的解析器
111    pub fn json() -> Self {
112        Self::new(
113            SerializationFormat::Json,
114            CompressionAlgorithm::None,
115            EncryptionAlgorithm::None,
116        )
117    }
118
119    /// 获取默认序列化格式
120    pub fn default_format(&self) -> SerializationFormat {
121        self.default_format
122    }
123
124    /// 获取默认压缩算法
125    pub fn default_compression(&self) -> CompressionAlgorithm {
126        self.default_compression.clone()
127    }
128
129    /// 获取默认加密算法
130    pub fn default_encryption(&self) -> EncryptionAlgorithm {
131        self.default_encryption.clone()
132    }
133
134    /// 解析消息（自动检测格式、压缩和加密）
135    ///
136    /// 处理流程：解密 -> 解压缩 -> 反序列化
137    /// 默认使用容错模式（解密失败时尝试作为未加密数据处理）
138    pub fn parse(&self, data: &[u8]) -> Result<Frame> {
139        self.parse_with_fallback(data, true)
140    }
141
142    /// 解析消息（支持容错标记）
143    ///
144    /// # 参数
145    /// - `data`: 要解析的原始数据
146    /// - `allow_fallback`: 如果为 true，启用容错模式：
147    ///   - 解密失败时尝试作为未加密数据处理
148    ///   - 解压缩失败时尝试作为未压缩数据处理
149    ///   - 反序列化失败时尝试所有序列化格式
150    ///   - 如果为 false，严格模式：任何步骤失败都直接返回错误
151    ///
152    /// # 处理流程
153    /// 解密（根据 allow_fallback 决定是否容错） -> 解压缩（根据 allow_fallback 决定是否容错） -> 反序列化（根据 allow_fallback 决定是否容错）
154    pub fn parse_with_fallback(&self, data: &[u8], allow_fallback: bool) -> Result<Frame> {
155        // 1. 解密数据（根据 allow_fallback 决定是否容错）
156        let decrypted = self.decrypt_data_with_fallback(data, allow_fallback)?;
157
158        // 2. 解压缩数据（根据 allow_fallback 决定是否容错）
159        let decompressed = self.decompress_data_with_fallback(&decrypted, allow_fallback)?;
160
161        // 3. 反序列化（根据 allow_fallback 决定是否容错）
162        self.parse_decompressed_with_fallback(&decompressed, allow_fallback)
163    }
164
165    /// 根据指定格式解析消息
166    pub fn parse_with_format(&self, data: &[u8], format: SerializationFormat) -> Result<Frame> {
167        // 1. 解密数据
168        let decrypted = self.decrypt_data(data)?;
169
170        // 2. 解压缩数据
171        let decompressed = self.decompress_data(&decrypted)?;
172
173        // 3. 使用指定的序列化器（优先使用自定义格式名称）
174        let serializer = if let Some(custom_name) = &self.custom_format_name {
175            // 如果指定了自定义格式名称，优先使用名称查找
176            SerializationUtil::get_serializer_by_name(custom_name)
177        } else {
178            // 否则使用格式枚举查找
179            SerializationUtil::get_serializer(format)
180        }
181        .ok_or_else(|| {
182            let format_info = if let Some(name) = &self.custom_format_name {
183                format!("custom format '{}'", name)
184            } else {
185                format!("{:?}", format)
186            };
187            crate::common::error::FlareError::deserialization_error(format!(
188                "Serializer not found: {}",
189                format_info
190            ))
191        })?;
192
193        serializer.deserialize(&decompressed)
194    }
195
196    /// 序列化消息（使用默认格式、压缩和加密）
197    pub fn serialize(&self, frame: &Frame) -> Result<Vec<u8>> {
198        self.serialize_with_format(
199            frame,
200            self.default_format,
201            self.default_compression.clone(),
202            self.default_encryption.clone(),
203        )
204    }
205
206    /// 序列化消息（指定格式、压缩和加密）
207    ///
208    /// 处理流程：序列化 -> 压缩 -> 加密
209    pub fn serialize_with_format(
210        &self,
211        frame: &Frame,
212        format: SerializationFormat,
213        compression: CompressionAlgorithm,
214        encryption: EncryptionAlgorithm,
215    ) -> Result<Vec<u8>> {
216        // 1. 使用指定的序列化器序列化（优先使用自定义格式名称）
217        let serializer = if let Some(custom_name) = &self.custom_format_name {
218            // 如果指定了自定义格式名称，优先使用名称查找
219            SerializationUtil::get_serializer_by_name(custom_name)
220        } else {
221            // 否则使用格式枚举查找
222            SerializationUtil::get_serializer(format)
223        }
224        .ok_or_else(|| {
225            let format_info = if let Some(name) = &self.custom_format_name {
226                format!("custom format '{}'", name)
227            } else {
228                format!("{:?}", format)
229            };
230            crate::common::error::FlareError::encoding_error(format!(
231                "Serializer not found: {}",
232                format_info
233            ))
234        })?;
235
236        let data = serializer.serialize(frame)?;
237
238        // 2. 应用压缩。低于阈值的小包保持未压缩，避免压缩头和 CPU 开销反噬。
239        let compressed = if Self::should_compress_payload(data.len(), &compression) {
240            CompressionUtil::compress(&data, compression)?
241        } else {
242            data
243        };
244
245        // 3. 应用加密
246        self.encrypt_data(&compressed, encryption)
247    }
248
249    /// 序列化消息（指定格式和压缩，使用默认加密）
250    pub fn serialize_with_format_compression(
251        &self,
252        frame: &Frame,
253        format: SerializationFormat,
254        compression: CompressionAlgorithm,
255    ) -> Result<Vec<u8>> {
256        self.serialize_with_format(frame, format, compression, self.default_encryption.clone())
257    }
258
259    /// 从 Frame 的 metadata 中读取压缩算法
260    pub fn get_compression_from_frame(frame: &Frame) -> CompressionAlgorithm {
261        frame
262            .metadata
263            .get("compression")
264            .and_then(|bytes| std::str::from_utf8(bytes).ok())
265            .and_then(CompressionAlgorithm::from_str)
266            .unwrap_or(CompressionAlgorithm::None)
267    }
268
269    /// 判断序列化后的 payload 是否值得压缩。
270    pub fn should_compress_payload(payload_len: usize, compression: &CompressionAlgorithm) -> bool {
271        *compression != CompressionAlgorithm::None && payload_len > MIN_COMPRESSION_PAYLOAD_BYTES
272    }
273
274    /// 从 Frame 的 metadata 中读取序列化格式
275    pub fn get_format_from_frame(frame: &Frame) -> Option<SerializationFormat> {
276        frame
277            .metadata
278            .get("format")
279            .and_then(|bytes| std::str::from_utf8(bytes).ok())
280            .and_then(|s| {
281                if s.eq_ignore_ascii_case("protobuf") {
282                    Some(SerializationFormat::Protobuf)
283                } else if s.eq_ignore_ascii_case("json") {
284                    Some(SerializationFormat::Json)
285                } else {
286                    None
287                }
288            })
289    }
290
291    /// 从 Frame 的 metadata 中读取加密算法
292    pub fn get_encryption_from_frame(frame: &Frame) -> EncryptionAlgorithm {
293        frame
294            .metadata
295            .get("encryption")
296            .and_then(|bytes| std::str::from_utf8(bytes).ok())
297            .and_then(EncryptionAlgorithm::from_str)
298            .unwrap_or(EncryptionAlgorithm::None)
299    }
300
301    // ============================================================================
302    // 内部辅助方法
303    // ============================================================================
304
305    /// 解密数据（内部辅助方法）
306    ///
307    /// # 参数
308    /// - `data`: 要解密的数据
309    /// - `allow_fallback`: 如果为 true，解密失败时尝试作为未加密数据处理（容错）
310    ///   如果为 false，解密失败直接返回错误（严格模式）
311    fn decrypt_data_with_fallback(&self, data: &[u8], allow_fallback: bool) -> Result<Vec<u8>> {
312        // 如果加密算法是 None，直接返回数据
313        if self.default_encryption == EncryptionAlgorithm::None {
314            return Ok(data.to_vec());
315        }
316
317        // 从全局注册表中查找加密器
318        let encryptor_name = self.default_encryption.as_str();
319        let encryptor = EncryptionUtil::find(&encryptor_name).ok_or_else(|| {
320            // 提供更详细的错误信息，包括已注册的加密器列表
321            let registered = EncryptionUtil::list_registered();
322            let error_msg = format!(
323                "Encryptor '{}' not found. Registered: {:?}",
324                encryptor_name, registered
325            );
326            tracing::error!("{}", error_msg);
327            crate::common::error::FlareError::deserialization_error(error_msg)
328        })?;
329
330        match encryptor.decrypt(data) {
331            Ok(decrypted) => Ok(decrypted),
332            Err(e) => {
333                if allow_fallback {
334                    tracing::trace!(
335                        "解密失败，尝试作为未加密数据处理: encryption={:?}, data_len={}",
336                        self.default_encryption,
337                        data.len()
338                    );
339                    Ok(data.to_vec())
340                } else {
341                    Err(crate::common::error::FlareError::deserialization_error(
342                        format!(
343                            "解密失败: encryption={:?}, error={}, data_len={}",
344                            self.default_encryption,
345                            e,
346                            data.len()
347                        ),
348                    ))
349                }
350            }
351        }
352    }
353
354    /// 解密数据（内部辅助方法，默认容错模式）
355    ///
356    /// 如果解密失败，尝试将数据作为未加密数据返回（容错处理）
357    /// 这样可以兼容客户端在收到 CONNECT_ACK 之前发送的未加密消息
358    fn decrypt_data(&self, data: &[u8]) -> Result<Vec<u8>> {
359        self.decrypt_data_with_fallback(data, true)
360    }
361
362    /// 加密数据（内部辅助方法）
363    fn encrypt_data(&self, data: &[u8], encryption: EncryptionAlgorithm) -> Result<Vec<u8>> {
364        // 如果加密算法是 None，直接返回数据
365        if encryption == EncryptionAlgorithm::None {
366            return Ok(data.to_vec());
367        }
368
369        // 从全局注册表中查找加密器
370        let encryptor_name = encryption.as_str();
371        let encryptor = EncryptionUtil::find(&encryptor_name).ok_or_else(|| {
372            // 提供更详细的错误信息，包括已注册的加密器列表
373            let registered = EncryptionUtil::list_registered();
374            let error_msg = format!(
375                "Encryptor '{}' not found. Registered: {:?}",
376                encryptor_name, registered
377            );
378            tracing::error!("{}", error_msg);
379            crate::common::error::FlareError::encoding_error(error_msg)
380        })?;
381
382        // 加密数据
383        encryptor.encrypt(data)
384    }
385
386    /// 解压缩数据（内部辅助方法，默认容错模式）
387    fn decompress_data(&self, data: &[u8]) -> Result<Vec<u8>> {
388        self.decompress_data_with_fallback(data, true)
389    }
390
391    /// 解压缩数据（支持容错标记）
392    ///
393    /// # 参数
394    /// - `data`: 要解压缩的数据
395    /// - `allow_fallback`: 如果为 true，解压缩失败时尝试作为未压缩数据处理（容错模式）
396    ///   如果为 false，解压缩失败直接返回错误（严格模式）
397    fn decompress_data_with_fallback(&self, data: &[u8], allow_fallback: bool) -> Result<Vec<u8>> {
398        // 如果压缩算法是 None，直接返回数据
399        if self.default_compression == CompressionAlgorithm::None {
400            return Ok(data.to_vec());
401        }
402
403        // 如果配置了压缩，先尝试自动检测并解压缩
404        // 这样可以处理即使配置了压缩，但数据可能未压缩的情况（容错）
405        match CompressionUtil::auto_decompress(data) {
406            Ok((decompressed, detected_algorithm)) => {
407                // 如果检测到压缩算法，说明数据确实是压缩的，返回解压缩后的数据
408                if detected_algorithm != CompressionAlgorithm::None {
409                    Ok(decompressed)
410                } else {
411                    // 如果自动检测没有检测到压缩，但配置了压缩
412                    if allow_fallback {
413                        tracing::trace!(
414                            "自动检测未发现压缩，按阈值策略作为未压缩数据处理: compression={:?}, data_len={}",
415                            self.default_compression,
416                            data.len()
417                        );
418                        Ok(data.to_vec())
419                    } else {
420                        // 严格模式：配置了压缩但数据未压缩，返回错误
421                        Err(crate::common::error::FlareError::deserialization_error(
422                            format!(
423                                "解压缩失败（严格模式）: 配置了压缩 {:?} 但数据未压缩",
424                                self.default_compression
425                            ),
426                        ))
427                    }
428                }
429            }
430            Err(e) => {
431                if allow_fallback {
432                    tracing::trace!(
433                        "解压缩失败，尝试作为未压缩数据处理: compression={:?}, data_len={}",
434                        self.default_compression,
435                        data.len()
436                    );
437                    Ok(data.to_vec())
438                } else {
439                    // 严格模式：解压缩失败直接返回错误
440                    Err(crate::common::error::FlareError::deserialization_error(
441                        format!(
442                            "解压缩失败（严格模式）: compression={:?}, error={}",
443                            self.default_compression, e
444                        ),
445                    ))
446                }
447            }
448        }
449    }
450
451    /// 解析已解压缩的数据（内部辅助方法，默认容错模式）
452    #[allow(dead_code)]
453    fn parse_decompressed(&self, decompressed: &[u8]) -> Result<Frame> {
454        self.parse_decompressed_with_fallback(decompressed, true)
455    }
456
457    /// 解析已解压缩的数据（支持容错标记）
458    ///
459    /// # 参数
460    /// - `decompressed`: 已解压缩的数据
461    /// - `allow_fallback`: 如果为 true，反序列化失败时尝试所有序列化格式（容错模式）
462    ///   如果为 false，只尝试默认格式，失败直接返回错误（严格模式）
463    fn parse_decompressed_with_fallback(
464        &self,
465        decompressed: &[u8],
466        allow_fallback: bool,
467    ) -> Result<Frame> {
468        // 尝试自动检测序列化格式
469        let detected_serializers = SerializationUtil::auto_detect(decompressed);
470
471        // 尝试每个检测到的序列化器
472        for serializer in detected_serializers {
473            if let Ok(frame) = serializer.deserialize(decompressed) {
474                return Ok(frame);
475            }
476        }
477
478        if allow_fallback {
479            // 容错模式：如果自动检测失败，尝试所有已注册的序列化器
480            self.try_all_serializers(decompressed)
481        } else {
482            // 严格模式：只尝试默认格式（优先使用自定义格式名称）
483            let serializer = if let Some(custom_name) = &self.custom_format_name {
484                // 如果指定了自定义格式名称，优先使用名称查找
485                SerializationUtil::get_serializer_by_name(custom_name)
486            } else {
487                // 否则使用格式枚举查找
488                SerializationUtil::get_serializer(self.default_format)
489            }
490            .ok_or_else(|| {
491                let format_info = if let Some(name) = &self.custom_format_name {
492                    format!("custom format '{}'", name)
493                } else {
494                    format!("format {:?}", self.default_format)
495                };
496                crate::common::error::FlareError::deserialization_error(format!(
497                    "Serializer not found for {}",
498                    format_info
499                ))
500            })?;
501            serializer.deserialize(decompressed).map_err(|e| {
502                let format_info = if let Some(name) = &self.custom_format_name {
503                    format!("custom format '{}'", name)
504                } else {
505                    format!("format {:?}", self.default_format)
506                };
507                crate::common::error::FlareError::deserialization_error(format!(
508                    "反序列化失败（严格模式）: {}, error={}",
509                    format_info, e
510                ))
511            })
512        }
513    }
514
515    /// 尝试所有已注册的序列化器（内部辅助方法）
516    fn try_all_serializers(&self, data: &[u8]) -> Result<Frame> {
517        [SerializationFormat::Protobuf, SerializationFormat::Json]
518            .iter()
519            .find_map(|&format| {
520                SerializationUtil::get_serializer(format)
521                    .and_then(|serializer| serializer.deserialize(data).ok())
522            })
523            .ok_or_else(|| {
524                crate::common::error::FlareError::deserialization_error(
525                    "Failed to parse message: no compatible serializer found".to_string(),
526                )
527            })
528    }
529}
530
531#[cfg(test)]
532mod tests {
533    use super::*;
534    use crate::common::protocol::{FrameBuilder, ping};
535
536    #[test]
537    fn test_parse_protobuf() {
538        let parser = MessageParser::protobuf();
539        let frame = FrameBuilder::new()
540            .with_command(crate::common::protocol::Command {
541                r#type: Some(
542                    crate::common::protocol::flare::core::commands::command::Type::System(ping()),
543                ),
544            })
545            .build();
546
547        let data = parser.serialize(&frame).unwrap();
548        let parsed = parser.parse(&data).unwrap();
549        assert_eq!(parsed.message_id, frame.message_id);
550    }
551
552    #[test]
553    fn test_parse_json() {
554        let parser = &PRE_NEGOTIATION_PARSER;
555        let frame = FrameBuilder::new()
556            .with_command(crate::common::protocol::Command {
557                r#type: Some(
558                    crate::common::protocol::flare::core::commands::command::Type::System(ping()),
559                ),
560            })
561            .build();
562
563        let data = parser.serialize(&frame).unwrap();
564        let parsed = parser.parse(&data).unwrap();
565        assert_eq!(parsed.message_id, frame.message_id);
566    }
567
568    #[test]
569    #[cfg(feature = "compression-gzip")]
570    fn small_payload_skips_compression_and_strict_parse_rejects_it() {
571        let parser = MessageParser::new(
572            SerializationFormat::Protobuf,
573            CompressionAlgorithm::Gzip,
574            EncryptionAlgorithm::None,
575        );
576        let frame = FrameBuilder::new()
577            .with_command(crate::common::protocol::Command {
578                r#type: Some(
579                    crate::common::protocol::flare::core::commands::command::Type::System(ping()),
580                ),
581            })
582            .build();
583
584        let data = parser.serialize(&frame).unwrap();
585        let (_, detected) = CompressionUtil::auto_decompress(&data).unwrap();
586        assert_eq!(detected, CompressionAlgorithm::None);
587
588        let parsed = parser.parse_with_fallback(&data, true).unwrap();
589        assert_eq!(parsed.message_id, frame.message_id);
590
591        let error = parser.parse_with_fallback(&data, false).unwrap_err();
592        assert!(error.to_string().contains("严格模式"));
593    }
594
595    #[test]
596    #[cfg(feature = "compression-gzip")]
597    fn large_payload_uses_negotiated_compression() {
598        let parser = MessageParser::new(
599            SerializationFormat::Protobuf,
600            CompressionAlgorithm::Gzip,
601            EncryptionAlgorithm::None,
602        );
603        let frame = FrameBuilder::new()
604            .with_metadata(
605                "padding".to_string(),
606                vec![b'x'; MIN_COMPRESSION_PAYLOAD_BYTES * 2],
607            )
608            .build();
609
610        let data = parser.serialize(&frame).unwrap();
611        let (_, detected) = CompressionUtil::auto_decompress(&data).unwrap();
612        assert_eq!(detected, CompressionAlgorithm::Gzip);
613
614        let parsed = parser.parse_with_fallback(&data, false).unwrap();
615        assert_eq!(parsed.message_id, frame.message_id);
616    }
617
618    #[test]
619    fn test_get_encryption_from_frame() {
620        use crate::common::protocol::FrameBuilder;
621        use std::collections::HashMap;
622
623        // 测试无加密 metadata
624        let frame = FrameBuilder::new().build();
625        assert_eq!(
626            MessageParser::get_encryption_from_frame(&frame),
627            EncryptionAlgorithm::None
628        );
629
630        // 测试有加密 metadata
631        let mut metadata = HashMap::new();
632        metadata.insert("encryption".to_string(), b"aes256gcm".to_vec());
633        let frame = FrameBuilder::new()
634            .with_metadata("encryption".to_string(), b"aes256gcm".to_vec())
635            .build();
636        assert_eq!(
637            MessageParser::get_encryption_from_frame(&frame),
638            EncryptionAlgorithm::Aes256Gcm
639        );
640
641        // 测试无效加密 metadata（应返回 Custom 算法）
642        let frame = FrameBuilder::new()
643            .with_metadata("encryption".to_string(), b"invalid".to_vec())
644            .build();
645        assert_eq!(
646            MessageParser::get_encryption_from_frame(&frame),
647            EncryptionAlgorithm::Custom("invalid".to_string())
648        );
649    }
650}
flare_core/common/message/parser.rs

flare_core/common/message/
parser.rs
