tauri_plugin_py_runtime/bridge/
transport.rs

1//! 发往 Python 的帧发送端。
2//!
3//! 架构位置：前端 -> Tauri 命令 -> Bridge -> **Writer**/FrameCodec -> Python 进程（stdin）。
4//!
5//! 不变量：每个 Python 进程只有一个写任务独占其 stdin，所有帧都经由同一个队列串行写出，
6//! 因此并发调用的帧不会交错。帧在入队前完成编码并检查大小，超限帧不会进入队列。
7
8use bytes::Bytes;
9use futures_util::SinkExt;
10use tokio::{io::AsyncWrite, sync::mpsc};
11use tokio_util::codec::FramedWrite;
12
13use crate::{
14  bridge::protocol::{length_codec, Frame},
15  error::{codes, Error, Result},
16};
17
18/// 单写者句柄：可廉价克隆，所有克隆共享同一个写任务。
19///
20/// 帧由 [`Writer::send`] 编码，因此超限帧在入队前即被拒绝。队列无界、不提供背压；
21/// 写任务因 I/O 错误退出或接收端被丢弃后，后续发送返回 `runtime_unavailable`。
22#[derive(Clone)]
23pub struct Writer {
24  tx: mpsc::UnboundedSender<Bytes>,
25  max_frame_size: usize,
26}
27
28impl Writer {
29  /// 创建一个不绑定 I/O 的写句柄及其接收端，接收端得到的是不含外层长度前缀的帧数据。
30  ///
31  /// 主要用于测试，或由调用方自行驱动写出；生产环境使用 [`Writer::spawn`]。
32  pub fn channel(max_frame_size: usize) -> (Self, mpsc::UnboundedReceiver<Bytes>) {
33    let (tx, rx) = mpsc::unbounded_channel();
34    (Self { tx, max_frame_size }, rx)
35  }
36
37  /// 启动后台写任务，把帧加上 u32 大端长度前缀后依次写入 `stream`。
38  ///
39  /// 写入失败时记录错误并结束写任务，此后所有 [`Writer::send`] 都会失败。
40  pub fn spawn<W: AsyncWrite + Unpin + Send + 'static>(stream: W, max_frame_size: usize) -> Self {
41    let (writer, mut rx) = Self::channel(max_frame_size);
42    tauri::async_runtime::spawn(async move {
43      let mut sink = FramedWrite::new(stream, length_codec(max_frame_size));
44      while let Some(frame) = rx.recv().await {
45        if let Err(e) = sink.send(frame).await {
46          log::error!(target: "py-runtime", "failed to write frame: {e}");
47          break;
48        }
49      }
50    });
51    writer
52  }
53
54  /// 编码并排队发送一帧，不等待实际写出。
55  ///
56  /// # Errors
57  ///
58  /// - `frame_too_large`：编码后的帧超过 `max_frame_size`；
59  /// - `runtime_unavailable`：写任务已结束（Python 进程已退出或连接已关闭）。
60  pub fn send(&self, frame: Frame) -> Result<()> {
61    let data = frame.encode();
62    if data.len() > self.max_frame_size {
63      return Err(Error::new(
64        codes::FRAME_TOO_LARGE,
65        format!("frame of {} bytes exceeds limit of {}", data.len(), self.max_frame_size),
66      ));
67    }
68    self
69      .tx
70      .send(data)
71      .map_err(|_| Error::new(codes::RUNTIME_UNAVAILABLE, "python runtime is not running"))
72  }
73}
74
75#[cfg(test)]
76mod tests {
77  use super::*;
78  use crate::bridge::protocol::{FrameCodec, Message, DEFAULT_MAX_FRAME_SIZE};
79  use futures_util::StreamExt;
80  use tokio_util::codec::FramedRead;
81
82  fn close(channel: u64) -> Frame {
83    Frame::new(Message::ChannelClose { channel })
84  }
85
86  #[tokio::test]
87  async fn spawned_writer_writes_frames_in_order() {
88    let (a, b) = tokio::io::duplex(1024);
89    let writer = Writer::spawn(a, DEFAULT_MAX_FRAME_SIZE);
90    let mut frames = FramedRead::new(b, FrameCodec::new(DEFAULT_MAX_FRAME_SIZE));
91    for i in 0..3 {
92      writer.send(close(i)).unwrap();
93    }
94    for i in 0..3 {
95      assert_eq!(frames.next().await.unwrap().unwrap().unwrap(), close(i));
96    }
97  }
98
99  #[test]
100  fn rejects_oversized_frames() {
101    let (writer, _rx) = Writer::channel(16);
102    let mut frame = close(1);
103    frame.body = vec![0; 100];
104    assert_eq!(writer.send(frame).unwrap_err().code, "frame_too_large");
105  }
106
107  #[test]
108  fn send_after_close_is_unavailable() {
109    let (writer, rx) = Writer::channel(1024);
110    drop(rx);
111    assert_eq!(writer.send(close(1)).unwrap_err().code, "runtime_unavailable");
112  }
113}
tauri_plugin_py_runtime/bridge/transport.rs

tauri_plugin_py_runtime/bridge/
transport.rs
