tauri_plugin_py_runtime/bridge/
protocol.rs

1//! v2 帧协议：消息类型、帧编解码与载荷表示。
2//!
3//! 架构位置：前端 -> Tauri 命令 -> Bridge -> Writer/**FrameCodec** -> Python 进程。
4//! 本模块只描述线上格式，不涉及调用状态。Python 端对应 `tauri_runtime.protocol` 中的 pydantic
5//! 模型，两端以 `python-sdk/tests/fixtures/golden_frames.json` 逐字节对齐。
6//!
7//! 线上格式：`[u32 BE 内层长度][u32 BE 头长度][头 JSON][body]`。
8//!
9//! - 外层长度只计其后的字节，上限为 `max_frame_size`（默认 [`DEFAULT_MAX_FRAME_SIZE`]）；
10//! - 头是以 `type` 字段区分的 JSON 对象（见 [`Message`]），未知字段被忽略；
11//! - JSON 载荷内联在头的 `value` 字段，二进制载荷放在 body 并在头中标记 `raw: true`；
12//!   未设置 `raw` 时 body 必须为空（见 [`Inline::join`]）。
13//!
14//! 不变量：头无法解码的帧只影响该帧本身，可以跳过；超限帧意味着对端异常，按致命错误处理
15//! （见 [`FrameCodec`]）。
16
17use std::collections::BTreeMap;
18
19use bytes::{BufMut, Bytes, BytesMut};
20use serde::{Deserialize, Serialize};
21use serde_json::Value;
22use tokio_util::codec::{Decoder, LengthDelimitedCodec, LengthDelimitedCodecError};
23
24use crate::error::{codes, Error};
25
26/// 当前协议版本，握手时 `hello.protocol` 必须与之相等，否则启动失败并返回 `protocol_mismatch`。
27pub const PROTOCOL_VERSION: u64 = 2;
28/// 默认的单帧大小上限（64 MiB），按外层长度前缀之后的字节数计算。
29pub const DEFAULT_MAX_FRAME_SIZE: usize = 64 * 1024 * 1024;
30
31/// 帧编解码过程中的错误。
32#[derive(Debug, thiserror::Error)]
33pub enum ProtocolError {
34  /// 底层 I/O 错误，包括 EOF 时帧不完整；对连接而言是致命的。
35  #[error("io error: {0}")]
36  Io(#[from] std::io::Error),
37  /// 长度前缀超过上限（携带该上限）；对连接而言是致命的。
38  #[error("frame exceeds limit of {0} bytes")]
39  TooLarge(usize),
40  /// 帧完整但内容非法（头 JSON 错误、未知 `type`、字段缺失或类型错误、载荷不一致），可跳过。
41  #[error("malformed frame: {0}")]
42  Malformed(String),
43}
44
45impl From<ProtocolError> for Error {
46  fn from(e: ProtocolError) -> Self {
47    match e {
48      ProtocolError::TooLarge(_) => Error::new(codes::FRAME_TOO_LARGE, e.to_string()),
49      _ => Error::new(codes::INTERNAL, e.to_string()),
50    }
51  }
52}
53
54/// `ready` 消息中携带的宿主应用信息，取自 Tauri 的 `PackageInfo`。
55#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
56pub struct AppInfo {
57  /// 应用名称。
58  pub name: String,
59  /// 应用版本号。
60  pub version: String,
61}
62
63/// 消息头中的载荷字段：JSON 载荷内联为 `value`，二进制载荷放在帧 body 中并以 `raw: true` 标记。
64///
65/// 两个字段均为默认值时不会出现在序列化结果中；与 body 组合成 [`Payload`] 见 [`Inline::join`]。
66#[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize)]
67pub struct Inline {
68  /// 内联的 JSON 载荷；缺省表示无载荷（`null`）。
69  #[serde(default, skip_serializing_if = "Option::is_none")]
70  pub value: Option<Value>,
71  /// 为 `true` 时载荷是帧 body 中的原始字节，此时 `value` 必须缺省。
72  #[serde(default, skip_serializing_if = "std::ops::Not::not")]
73  pub raw: bool,
74}
75
76/// 帧头，即以 `type` 字段区分的 JSON 对象。
77///
78/// 字段声明顺序即线上顺序，与 Python 模型一致（golden frames 依赖于此）；反序列化时忽略未知字段。
79#[derive(Debug, Clone, PartialEq, Serialize, Deserialize, strum::IntoStaticStr)]
80#[serde(tag = "type", rename_all = "snake_case")]
81#[strum(serialize_all = "snake_case")]
82pub enum Message {
83  /// Python → Rust：握手首帧，声明协议版本、SDK 版本与已注册的命令。
84  Hello {
85    protocol: u64,
86    #[serde(default)]
87    sdk: String,
88    #[serde(default)]
89    commands: Vec<String>,
90  },
91  /// Rust → Python：确认握手，Python 收到后执行 `ready` 钩子。
92  Ready {
93    protocol: u64,
94    app: AppInfo,
95  },
96  /// Rust → Python：请求退出，`timeoutMs` 是 Python 取消调用并执行 `shutdown` 钩子的总预算。
97  Shutdown {
98    #[serde(rename = "timeoutMs")]
99    timeout_ms: u64,
100  },
101  /// Rust → Python：命令调用请求。
102  Request {
103    /// Rust 侧分配的调用 id，在单个 [`crate::bridge::Bridge`] 内唯一。
104    id: u64,
105    cmd: String,
106    /// 参数名 → channel id。
107    #[serde(default, skip_serializing_if = "BTreeMap::is_empty")]
108    channels: BTreeMap<String, u64>,
109    /// 发起调用的 webview 标签。
110    #[serde(default, skip_serializing_if = "Option::is_none")]
111    window: Option<String>,
112    #[serde(flatten)]
113    payload: Inline,
114  },
115  /// Python → Rust：调用成功的结果。
116  Response {
117    id: u64,
118    #[serde(flatten)]
119    payload: Inline,
120  },
121  /// Python → Rust：调用失败；缺省的 `code` 按 `internal` 处理。
122  Error {
123    id: u64,
124    #[serde(default, skip_serializing_if = "Option::is_none")]
125    code: Option<String>,
126    #[serde(default, skip_serializing_if = "Option::is_none")]
127    message: Option<String>,
128    #[serde(default, skip_serializing_if = "Option::is_none")]
129    data: Option<Value>,
130  },
131  /// Rust → Python：取消调用，`reason` 为 `"client"` 或 `"timeout"`。
132  Cancel {
133    id: u64,
134    reason: String,
135  },
136  /// 双向事件。Python → Rust 的事件携带 `target`，Rust → Python 的事件携带 `window`。
137  Event {
138    event: String,
139    /// 目标 webview 标签；缺省时广播。
140    #[serde(default, skip_serializing_if = "Option::is_none")]
141    target: Option<String>,
142    /// 发送事件的 webview 标签。
143    #[serde(default, skip_serializing_if = "Option::is_none")]
144    window: Option<String>,
145    #[serde(flatten)]
146    payload: Inline,
147  },
148  /// Python → Rust：向前端 `Channel` 推送一条数据。
149  ChannelData {
150    channel: u64,
151    #[serde(flatten)]
152    payload: Inline,
153  },
154  /// Python → Rust：关闭 channel，此后该 channel 的数据被丢弃。
155  ChannelClose {
156    channel: u64,
157  },
158}
159
160/// 与线上表示无关的载荷值。
161#[derive(Debug, Clone, PartialEq)]
162pub enum Payload {
163  /// 无载荷，对前端表现为 `null`。
164  None,
165  /// JSON 值。
166  Json(Value),
167  /// 原始字节，前端收到 `Uint8Array` / `ArrayBuffer`。
168  Raw(Vec<u8>),
169}
170
171impl Payload {
172  /// 拆分为消息头中的载荷字段和帧 body。
173  pub fn split(self) -> (Inline, Vec<u8>) {
174    match self {
175      Self::None => (Inline::default(), Vec::new()),
176      Self::Json(value) => (Inline { value: Some(value), raw: false }, Vec::new()),
177      Self::Raw(bytes) => (Inline { value: None, raw: true }, bytes),
178    }
179  }
180}
181
182impl Inline {
183  /// 将载荷字段与帧 body 组合为 [`Payload`]，是 [`Payload::split`] 的逆操作。
184  ///
185  /// # Errors
186  ///
187  /// 以下情况返回 [`ProtocolError::Malformed`]（转换为 [`Error`] 后错误码为 `internal`）：
188  /// `raw` 为 `true` 却同时带有 `value`，或未设置 `raw` 而 body 非空。
189  pub fn join(self, body: Vec<u8>) -> Result<Payload, ProtocolError> {
190    match (self.raw, self.value) {
191      (true, None) => Ok(Payload::Raw(body)),
192      (false, value) if body.is_empty() => Ok(value.map_or(Payload::None, Payload::Json)),
193      (true, Some(_)) => Err(ProtocolError::Malformed("raw payload with an inline value".into())),
194      (false, _) => Err(ProtocolError::Malformed("frame body without the raw flag".into())),
195    }
196  }
197}
198
199/// 一个完整的协议帧：消息头加上可选的二进制 body。
200#[derive(Debug, Clone, PartialEq)]
201pub struct Frame {
202  /// 帧头。
203  pub message: Message,
204  /// 帧 body，仅在载荷 `raw: true` 时非空。
205  pub body: Vec<u8>,
206}
207
208impl Frame {
209  /// 构造 body 为空的帧。
210  pub fn new(message: Message) -> Self {
211    Self { message, body: Vec::new() }
212  }
213
214  /// 编码为 `[u32 BE 头长度][头 JSON][body]`，不含外层长度前缀（由 [`length_codec`] 添加）。
215  ///
216  /// # Panics
217  ///
218  /// 仅当 [`Message`] 序列化失败时 panic；所有消息都可序列化为 JSON，实际不会发生。
219  pub fn encode(&self) -> Bytes {
220    let header = serde_json::to_vec(&self.message).expect("messages always serialize");
221    let mut buf = BytesMut::with_capacity(4 + header.len() + self.body.len());
222    buf.put_u32(header.len() as u32);
223    buf.put_slice(&header);
224    buf.put_slice(&self.body);
225    buf.freeze()
226  }
227
228  /// 从去掉外层长度前缀的帧数据解码。
229  ///
230  /// # Errors
231  ///
232  /// 数据不足 4 字节、头长度超出帧长度或头 JSON 无法解码为 [`Message`] 时返回
233  /// [`ProtocolError::Malformed`]。
234  pub fn decode(buf: &[u8]) -> Result<Self, ProtocolError> {
235    let (len, rest) = buf
236      .split_first_chunk::<4>()
237      .ok_or_else(|| ProtocolError::Malformed("frame shorter than 4 bytes".into()))?;
238    let header_len = u32::from_be_bytes(*len) as usize;
239    if header_len > rest.len() {
240      return Err(ProtocolError::Malformed("header length exceeds frame".into()));
241    }
242    let (header, body) = rest.split_at(header_len);
243    let message =
244      serde_json::from_slice(header).map_err(|e| ProtocolError::Malformed(format!("invalid header: {e}")))?;
245    Ok(Self { message, body: body.to_vec() })
246  }
247}
248
249/// 外层分帧编解码器：u32 大端长度前缀，帧长上限为 `max_frame_size`。
250pub fn length_codec(max_frame_size: usize) -> LengthDelimitedCodec {
251  LengthDelimitedCodec::builder()
252    .big_endian()
253    .length_field_type::<u32>()
254    .max_frame_length(max_frame_size)
255    .new_codec()
256}
257
258/// 基于 [`length_codec`] 的帧解码器，用于读取 Python 的 stdout。
259///
260/// 对于完整但非法的帧（JSON 错误、未知消息类型、字段缺失），解码结果为 `Ok(Err(_))`：
261/// 此时数据流仍对齐在帧边界上，可以继续读取。`Err(_)` 是致命错误，其中包括长度前缀超限时的
262/// [`ProtocolError::TooLarge`]：双方在发送前都会检查大小，出现超限帧说明对端已损坏，因此不跳过。
263pub struct FrameCodec {
264  inner: LengthDelimitedCodec,
265  max_frame_size: usize,
266}
267
268impl FrameCodec {
269  /// 创建帧长上限为 `max_frame_size` 的解码器。
270  pub fn new(max_frame_size: usize) -> Self {
271    Self { inner: length_codec(max_frame_size), max_frame_size }
272  }
273}
274
275impl Decoder for FrameCodec {
276  type Item = Result<Frame, ProtocolError>;
277  type Error = ProtocolError;
278
279  fn decode(&mut self, src: &mut BytesMut) -> Result<Option<Self::Item>, ProtocolError> {
280    let chunk = self.inner.decode(src).map_err(|e| {
281      if e.get_ref().is_some_and(|inner| inner.is::<LengthDelimitedCodecError>()) {
282        ProtocolError::TooLarge(self.max_frame_size)
283      } else {
284        ProtocolError::Io(e)
285      }
286    })?;
287    Ok(chunk.map(|buf| Frame::decode(&buf)))
288  }
289}
290
291#[cfg(test)]
292mod tests {
293  use super::*;
294  use serde_json::json;
295  use tokio_util::codec::Encoder;
296
297  fn wire(frame: &Frame) -> Vec<u8> {
298    let mut buf = BytesMut::new();
299    length_codec(DEFAULT_MAX_FRAME_SIZE).encode(frame.encode(), &mut buf).unwrap();
300    buf.to_vec()
301  }
302
303  fn raw_wire(header: &[u8]) -> Vec<u8> {
304    [&((4 + header.len()) as u32).to_be_bytes()[..], &(header.len() as u32).to_be_bytes(), header].concat()
305  }
306
307  fn decode_all(data: &[u8], max: usize) -> Vec<Result<Frame, ProtocolError>> {
308    let mut codec = FrameCodec::new(max);
309    let mut buf = BytesMut::from(data);
310    std::iter::from_fn(|| codec.decode(&mut buf).unwrap()).collect()
311  }
312
313  fn event(name: &str) -> Frame {
314    Frame::new(Message::Event { event: name.into(), target: None, window: None, payload: Inline::default() })
315  }
316
317  fn raw_response(id: u64, body: &[u8]) -> Frame {
318    let (payload, body) = Payload::Raw(body.to_vec()).split();
319    Frame { message: Message::Response { id, payload }, body }
320  }
321
322  /// 必须与 `python-sdk/tests/test_protocol.py` 中的 `GOLDEN_FRAMES` 保持一致。
323  fn golden_frames() -> Vec<(&'static str, Frame)> {
324    let json = |value: Value| Inline { value: Some(value), raw: false };
325    vec![
326      ("hello", Frame::new(Message::Hello { protocol: 2, sdk: "0.1.0".into(), commands: vec!["greet".into()] })),
327      (
328        "ready",
329        Frame::new(Message::Ready { protocol: 2, app: AppInfo { name: "app".into(), version: "1.0.0".into() } }),
330      ),
331      ("shutdown", Frame::new(Message::Shutdown { timeout_ms: 3000 })),
332      (
333        "request",
334        Frame::new(Message::Request {
335          id: 7,
336          cmd: "greet".into(),
337          channels: [("onItem".into(), 3)].into(),
338          window: Some("main".into()),
339          payload: json(json!({"n": [1, 2.5, null], "name": "中文"})),
340        }),
341      ),
342      ("response_raw", raw_response(7, &[0, 255])),
343      (
344        "error",
345        Frame::new(Message::Error {
346          id: 7,
347          code: Some("custom".into()),
348          message: Some("bad".into()),
349          data: Some(json!({"x": 1})),
350        }),
351      ),
352      ("cancel", Frame::new(Message::Cancel { id: 7, reason: "timeout".into() })),
353      (
354        "event",
355        Frame::new(Message::Event {
356          event: "tick".into(),
357          target: Some("main".into()),
358          window: None,
359          payload: Inline::default(),
360        }),
361      ),
362      ("channel_data", Frame::new(Message::ChannelData { channel: 3, payload: json(json!(1)) })),
363      ("channel_close", Frame::new(Message::ChannelClose { channel: 3 })),
364    ]
365  }
366
367  #[test]
368  fn golden_frames_match_python() {
369    let fixture: BTreeMap<String, String> = serde_json::from_str(include_str!(concat!(
370      env!("CARGO_MANIFEST_DIR"),
371      "/python-sdk/tests/fixtures/golden_frames.json"
372    )))
373    .unwrap();
374    let frames = golden_frames();
375    assert_eq!(fixture.keys().collect::<Vec<_>>(), {
376      let mut names: Vec<_> = frames.iter().map(|(name, _)| *name).collect();
377      names.sort();
378      names
379    });
380    for (name, frame) in frames {
381      let hex = &fixture[name];
382      let data: Vec<u8> = (0..hex.len()).step_by(2).map(|i| u8::from_str_radix(&hex[i..i + 2], 16).unwrap()).collect();
383      assert_eq!(wire(&frame), data, "{name}: encoding differs");
384      let decoded = decode_all(&data, DEFAULT_MAX_FRAME_SIZE).pop().unwrap().unwrap();
385      assert_eq!(decoded, frame, "{name}: decoding differs");
386    }
387  }
388
389  #[test]
390  fn layout() {
391    let data = wire(&raw_response(2, b"xy"));
392    let header = br#"{"type":"response","id":2,"raw":true}"#;
393    assert_eq!(&data[..4], &((4 + header.len() + 2) as u32).to_be_bytes());
394    assert_eq!(&data[4..8], &(header.len() as u32).to_be_bytes());
395    assert_eq!(&data[8..], [&header[..], b"xy"].concat().as_slice());
396  }
397
398  #[test]
399  fn partial_input_waits_and_truncated_eof_is_fatal() {
400    let data = wire(&raw_response(1, &[0, 255]));
401    let mut codec = FrameCodec::new(DEFAULT_MAX_FRAME_SIZE);
402    let mut buf = BytesMut::from(&data[..data.len() - 1]);
403    assert!(codec.decode(&mut buf).unwrap().is_none());
404    assert!(codec.decode_eof(&mut buf).is_err());
405    assert!(codec.decode_eof(&mut BytesMut::new()).unwrap().is_none());
406  }
407
408  #[test]
409  fn oversized_frame_is_fatal() {
410    let mut buf = BytesMut::from(&wire(&raw_response(1, &[1; 100]))[..]);
411    let err = FrameCodec::new(64).decode(&mut buf).unwrap_err();
412    assert!(matches!(err, ProtocolError::TooLarge(64)), "{err:?}");
413    assert_eq!(Error::from(err).code, codes::FRAME_TOO_LARGE);
414  }
415
416  #[test]
417  fn malformed_frames_are_skipped() {
418    let bad: [&[u8]; 6] = [
419      br#"{"type":"bogus"}"#,
420      br#"{"type":"response"}"#,
421      br#"{"type":"response","id":"1"}"#,
422      br#"{"id":1}"#,
423      b"[1]",
424      b"{not json",
425    ];
426    let mut data: Vec<u8> = bad.iter().flat_map(|h| raw_wire(h)).collect();
427    data.extend([0, 0, 0, 4, 0, 0, 0, 9]);
428    data.extend(wire(&event("ok")));
429    let frames = decode_all(&data, 1024);
430    assert_eq!(frames.len(), bad.len() + 2);
431    for frame in &frames[..bad.len() + 1] {
432      assert!(matches!(frame, Err(ProtocolError::Malformed(_))), "{frame:?}");
433    }
434    assert_eq!(frames.last().unwrap().as_ref().unwrap(), &event("ok"));
435  }
436
437  #[test]
438  fn payload_roundtrip_and_validation() {
439    for payload in [Payload::None, Payload::Json(json!({"a": [1, null]})), Payload::Raw(vec![1, 2])] {
440      let (inline, body) = payload.clone().split();
441      assert_eq!(inline.join(body).unwrap(), payload);
442    }
443    assert!(Inline { value: Some(json!(1)), raw: false }.join(vec![1]).is_err());
444    assert!(Inline { value: Some(json!(1)), raw: true }.join(vec![1]).is_err());
445  }
446
447  #[test]
448  fn absent_fields_are_omitted() {
449    let frame = Frame::new(Message::Request {
450      id: 1,
451      cmd: "hi".into(),
452      channels: BTreeMap::new(),
453      window: None,
454      payload: Inline::default(),
455    });
456    assert_eq!(&wire(&frame)[8..], br#"{"type":"request","id":1,"cmd":"hi"}"#);
457    let message: Message = serde_json::from_str(r#"{"type":"error","id":1,"extra":true}"#).unwrap();
458    assert_eq!(message, Message::Error { id: 1, code: None, message: None, data: None });
459  }
460}
tauri_plugin_py_runtime/bridge/protocol.rs

tauri_plugin_py_runtime/bridge/
protocol.rs
