raft_rust/net/
codec.rs

1//! 长度前缀帧：`u32 LE length + bincode(WireMsg)`。
2
3// 从任意 Read/Write 读写帧（TCP 流）
4use std::io::{Read, Write};
5
6// WireMsg 需跨进程序列化
7use serde::{Deserialize, Serialize};
8// 客户端请求用 UUID 关联请求与响应
9use uuid::Uuid;
10
11// 编解码错误映射为库错误
12use crate::error::{Error, Result};
13// Raft 信封与客户端请求/响应类型
14use crate::raft::{Envelope, Request, Response};
15
16// 全模块统一的 bincode 配置，保证编解码一致
17const BINCODE: bincode::config::Configuration = bincode::config::standard();
18
19/// 线路消息：Raft 协议 或 客户端请求/响应。
20// 可克隆以便重试发送；可序列化以走 TCP
21// 三种载荷共用同一帧格式，由接收端按变体分发
22#[derive(Clone, Debug, Serialize, Deserialize)]
23// 节点间与客户端共用一条 TCP 语义
24pub enum WireMsg {
25    // 节点间 Raft 协议消息（选举、复制、心跳等）
26    Raft(Envelope),
27    // 客户端写入/读取请求，id 用于匹配 ClientReply
28    Client { id: Uuid, request: Request },
29    // 服务端对客户端请求的回复（成功响应或 Error）
30    ClientReply { id: Uuid, response: std::result::Result<Response, Error> },
31// 枚举定义结束
32}
33
34// 将线路消息编码为「4 字节小端长度 + body」完整帧
35pub fn encode(msg: &WireMsg) -> Result<Vec<u8>> {
36    // 先用 bincode 序列化消息体
37    let body = bincode::serde::encode_to_vec(msg, BINCODE)
38        // 序列化失败视为协议/数据非法
39        .map_err(|e| Error::InvalidData(e.to_string()))?;
40    // 预分配：长度前缀 4 字节 + body
41    let mut out = Vec::with_capacity(4 + body.len());
42    // 写入小端 u32 长度，便于接收端一次读头再读 body
43    out.extend_from_slice(&(body.len() as u32).to_le_bytes());
44    // 追加序列化后的消息体
45    out.extend_from_slice(&body);
46    // 返回完整可写帧
47    Ok(out)
48// encode 结束
49}
50
51// 仅解码 body（不含长度前缀），供 read_msg 在读完长度后调用
52pub fn decode(bytes: &[u8]) -> Result<WireMsg> {
53    // borrow_decode 返回 (值, 消耗字节数)，这里只取值
54    Ok(bincode::serde::borrow_decode_from_slice(bytes, BINCODE)
55        // 反序列化失败映射为 InvalidData
56        .map_err(|e| Error::InvalidData(e.to_string()))?
57        // 丢弃已消耗长度，只保留消息
58        .0)
59// decode 结束
60}
61
62// 向流写入一帧并 flush，确保对端能及时读到
63pub fn write_msg(w: &mut impl Write, msg: &WireMsg) -> Result<()> {
64    // 先编码为长度前缀帧
65    let frame = encode(msg)?;
66    // 写满整个帧
67    w.write_all(&frame).map_err(|e| Error::IO(e.to_string()))?;
68    // 冲刷缓冲，避免消息滞留在用户态
69    w.flush().map_err(|e| Error::IO(e.to_string()))?;
70    // 写路径成功完成
71    Ok(())
72// write_msg 结束
73}
74
75// 从流阻塞读取一帧并解码为 WireMsg
76pub fn read_msg(r: &mut impl Read) -> Result<WireMsg> {
77    // 先读 4 字节小端长度头
78    let mut len_buf = [0u8; 4];
79    // 读头不足则按 IO 错误返回（对端关闭/超时）
80    r.read_exact(&mut len_buf).map_err(|e| Error::IO(e.to_string()))?;
81    // 解析 body 长度
82    let len = u32::from_le_bytes(len_buf) as usize;
83    // 防止恶意/损坏帧占用过大内存（上限 64MB）
84    if len > 64 * 1024 * 1024 {
85        // 超限直接拒绝，避免 OOM
86        return Err(Error::InvalidData(format!("frame too large: {len}")));
87    // 长度校验分支结束
88    }
89    // 按长度分配并精确读取 body
90    let mut body = vec![0u8; len];
91    // body 读不全同样映射为 IO 错误
92    r.read_exact(&mut body).map_err(|e| Error::IO(e.to_string()))?;
93    // 解码 body 为业务消息
94    decode(&body)
95// read_msg 结束
96}
raft_rust/net/codec.rs

raft_rust/net/
codec.rs
