Rust Tonic框架如何实现gRPC的XML流式上传

tonic 原生不支持 xml 流式上传，需用 grpc streaming 传 raw bytes，再以 quick-xml 等流式解析器在业务层按 chunk 解析 xml 片段，维护解析状态实现事件驱动处理。

tonic 中如何实现「类 XML 流式上传」？

本质是：用 gRPC streaming（server_streaming 或 bidi_streaming）传原始字节，再在业务逻辑中按需解析 XML 片段（如 SAX 或 pull 解析）。不能依赖 tonic 自动绑定 XML。

• Protobuf message 中定义一个通用字段，例如 bytes payload，用于承载 XML 片段或完整文档
• 客户端分块发送 XML 内容（如按 <item>...</item> 切片），每块封装为一次 StreamingUploadRequest
• 服务端用 tonic::Streaming<streaminguploadrequest></streaminguploadrequest> 接收，并用 quick-xml（推荐 reader::events::BytesStart + Reader::read_event）做流式 XML 解析
• 避免一次性将整个流 collect 成 Vec<u8></u8>，否则失去流式意义，也易 OOM

service XmlUploadService {
  rpc UploadXml(stream StreamingUploadRequest) returns (StreamingUploadResponse);
}

message StreamingUploadRequest {
  bytes chunk = 1;  // raw XML fragment, e.g. "<record><id>1</id></record>"
  bool is_last = 2;
}

message StreamingUploadResponse {
  int32 parsed_count = 1;
  string status = 2;
}

为什么不能直接用 tonic + serde-xml？

serde-xml 和 quick-xml 都要求输入是完整 &[u8] 或 std::io::Read，而 tonic streaming 的 Streaming<t></t> 是异步迭代器，不是同步 reader。强行拼接会导致：

• 无法保证 XML well-formed：中间 chunk 可能截断标签（如 <user na> + <code>me="a">）
• quick-xml::Reader 不支持跨 chunk 恢复解析状态，必须自己缓存未闭合的 start tag
• 没有标准方式把 tonic::Streaming 转成 impl std::io::Read，因为它是 Stream<item result>></item>，非阻塞且带错误类型

真实可行的流式 XML 处理策略

不要试图让 XML 解析器“吞”整个 stream，而是设计成事件驱动：每收到一个 chunk，就喂给一个状态机，识别出完整起始/结束标签后触发回调。

• 用 quick-xml::events::BytesStart 和 BytesText 匹配常见结构，忽略注释、CDATA 等（除非业务强依赖）
• 维护一个栈记录当前嵌套路径（如 ["root", "items", "item"]），便于定位上下文
• 对每个完整 <record>...</record> 块启动异步处理（写 DB、转发 Kafka），而非等全部上传结束
• chunk 边界由客户端控制（建议固定大小如 64KB，或按语义切分），服务端不做粘包处理

let mut reader = Reader::from_reader(chunk.as_ref());
let mut buf = Vec::new();
while let Ok(event) = reader.read_event_into(&mut buf) {
    match event {
        BytesStart(ref e) if e.name().as_ref() == b"record" => {
            // 开始一条新 record
            records.push(Vec::new());
        }
        BytesText(e) => {
            if let Some(record) = records.last_mut() {
                record.extend_from_slice(&e.into_inner());
            }
        }
        BytesEnd(ref e) if e.name().as_ref() == b"record" => {
            // 触发解析 record 字节
            process_record(&records.pop().unwrap());
        }
        _ => {}
    }
    buf.clear();
}