Rust实现Postgres驱动：如何正确解码Socket返回的原始字节？

1. 先拆分Postgres协议的消息帧 #

Postgres的TCP通信不是纯文本流，而是遵循Frontend/Backend协议的结构化消息，每帧格式固定：

• 第1字节：消息类型标识（如'R'代表认证消息、'T'代表行描述、'D'代表数据行）
• 第2-5字节：4字节大端整数，代表整个消息的总长度（包含这4字节）
• 后续字节：对应消息类型的消息体

必须先按此规则拆分BytesMut，不能直接解码整个字节流。示例代码：

use bytes::BytesMut;
use byteorder::{BigEndian, ReadBytesExt};

fn split_postgres_messages(mut buf: BytesMut) -> Vec<(u8, BytesMut)> {
    let mut messages = Vec::new();
    while buf.len() >= 5 {
        let msg_type = buf[0];
        let len = buf[1..5].as_ref().read_u32::<BigEndian>().unwrap() as usize;
        if buf.len() >= len {
            let msg = buf.split_to(len);
            messages.push((msg_type, msg.split_off(1))); // 剥离消息类型字节，保留消息体
        } else {
            break; // 字节不足，等待后续数据
        }
    }
    messages
}

2. 针对不同消息类型处理编码 #

消息体中的字符串编码与数据库的server_encoding一致（默认多为UTF-8），控制字符（回车、空格、换行）均为ASCII兼容，不会出现乱码。乱码通常是因为误解码了协议中的二进制部分（如长度字段）。

示例：解码数据行消息（类型'D'） #

数据行消息体结构：1字节字段数 + 每个字段的4字节大端长度（-1表示NULL） + 字段数据。解码逻辑：

use byteorder::BigEndian;

fn decode_data_row(msg_body: &BytesMut) -> Vec<Option<String>> {
    let mut reader = msg_body.as_ref();
    let field_count = reader.read_u8().unwrap();
    let mut fields = Vec::with_capacity(field_count as usize);
    for _ in 0..field_count {
        let len = reader.read_i32::<BigEndian>().unwrap();
        if len == -1 {
            fields.push(None);
        } else {
            let field_bytes = reader.split_to(len as usize);
            // 按数据库编码解码，这里假设为UTF-8
            match String::from_utf8(field_bytes.to_vec()) {
                Ok(s) => fields.push(Some(s)),
                Err(e) => panic!("字段解码失败: {}", e), // 实际场景可根据编码调整错误处理
            }
        }
    }
    fields
}

3. 实现psql风格的格式化 #

你提到的管道符|并非Postgres返回的原始数据，是psql客户端自行添加的格式化元素。实现步骤：

• 先解析行描述消息（'T'）获取字段名与类型
• 计算每个字段的最大宽度，用于对齐
• 输出时手动添加分隔符、对齐空格与分隔线

简化示例：

// 假设已获取字段名列表fields和数据行rows
fn format_psql_style(fields: &[String], rows: &[Vec<Option<String>>]) {
    // 计算各列最大宽度
    let mut col_widths: Vec<usize> = fields.iter().map(|s| s.len()).collect();
    for row in rows {
        for (i, field) in row.iter().enumerate() {
            if let Some(s) = field {
                col_widths[i] = col_widths[i].max(s.len());
            }
        }
    }

    // 打印表头
    print!("|");
    for (i, field) in fields.iter().enumerate() {
        print!(" {:width$} |", field, width = col_widths[i]);
    }
    println!();

    // 打印分隔线
    print!("|");
    for &w in &col_widths {
        print!("{}|", "-".repeat(w + 2));
    }
    println!();

    // 打印数据行
    for row in rows {
        print!("|");
        for (i, field) in row.iter().enumerate() {
            let s = field.as_deref().unwrap_or("NULL");
            print!(" {:width$} |", s, width = col_widths[i]);
        }
        println!();
    }
}

关键注意事项 #

• 禁止直接解码整个BytesMut，必须先拆分协议消息帧
• 明确数据库编码，不要盲目尝试UTF-16解码（Postgres默认使用UTF-8）
• psql的格式化元素（管道符、对齐空格）需自行生成，并非协议返回内容

内容的提问来源于stack exchange，提问作者David Choi