Rust中发送XeLaTeX编译日志时提示Channel Closed问题排查

我正在开发一款云端XeLaTeX在线编辑器，触发编译操作时希望通过SSE将XeLaTeX的编译日志输出到浏览器。编写了最小化测试Demo，但通过Channel发送编译日志时提示错误“channel closed”。

Rust代码示例 #

use std::fmt::Display;
use std::io::Read;
use std::process::{Command, Stdio};
use std::time::Duration;

use actix_web::{web, App, HttpRequest, HttpResponse, HttpServer};
use tokio::sync::mpsc;
use tokio::task;
use tokio::time::{interval, Interval};

async fn run_xelatex() -> String {
    let mut cmd = Command::new("xelatex")
        .arg("-interaction=batchmode")
        .arg("/Users/xiaoqiangjiang/Nutstore/document/dolphin-book-2023/src/dolphin-book-2023.tex")
        .stdout(Stdio::piped())
        .spawn()
        .expect("Failed to execute xelatex command");

    let mut output = String::new();
    cmd.stdout
        .take()
        .unwrap()
        .read_to_string(&mut output)
        .unwrap();

    output
}

async fn sse_endpoint(req: HttpRequest) -> HttpResponse {
    let (tx, rx) = tokio::sync::mpsc::unbounded_channel();
    task::spawn(async move {
        let output = run_xelatex().await;
        let message = format!("data: {}\n\n", output);
        let send_result = tx.send(message);
        match send_result {
            Ok(_) => {},
            Err(e) => {
                println!("send xelatex compile log error: {}", e);
            }
        }
    });

    let response = HttpResponse::Ok()
        .content_type("text/event-stream")
        .streaming(SseStream {
            counter: 4,
            interval: interval(Duration::from_secs(5000)),
            receiver: Some(rx),
        });

    response
}

/// SSE stream implementation
struct SseStream<T>
where
    T: Display,
{
    counter: usize,
    interval: Interval,
    receiver: Option<mpsc::UnboundedReceiver<T>>,
}

impl<T: std::fmt::Display> SseStream<T> {
    fn new() -> Self {
        Self {
            counter: 0,
            interval: interval(Duration::from_secs(1000)),
            receiver: None,
        }
    }
}

impl<T: std::fmt::Display> futures::Stream for SseStream<T> {
    type Item = Result<actix_web::web::Bytes, actix_web::Error>;

    fn poll_next(
        mut self: std::pin::Pin<&mut Self>,
        _: &mut std::task::Context<'_>,
    ) -> std::task::Poll<Option<Self::Item>> {
        self.counter += 1;
        // get reciver data and send
        if let Some(receiver) = &mut self.receiver {
            match receiver.try_recv() {
                Ok(data) => {
                    // Create the SSE event
                    let message = format!("data: {}\n\n", data);

                    // Return the event as a stream item
                    return std::task::Poll::Ready(Some(Ok(actix_web::web::Bytes::from(message))));
                }
                Err(_err) => {
                    // handle error
                    return std::task::Poll::Ready(None);
                }
            }
        } else {
            return std::task::Poll::Ready(None);
        }

        // Return the event as a stream item
        //std::task::Poll::Ready(Some(Ok(actix_web::web::Bytes::from(message))))
    }
}

#[actix_web::main]
async fn main() -> std::io::Result<()> {
    HttpServer::new(|| App::new().route("/sse", web::get().to(sse_endpoint)))
        .bind("127.0.0.1:8080")?
        .run()
        .await
}

Cargo.toml配置 #

[package]
name = "rust-learn"
version = "0.1.0"
edition = "2018"

[dependencies]
tokio = { version = "1.17.0", features = ["full"] }
serde = { version = "1.0.64", features = ["derive"] }
serde_json = "1.0.64"
actix-web = "4"
futures = "0.3"
eventsource = "0.5"
bytes = "1"

测试命令：curl -N http://localhost:8080/sse

问题根源分析 #

1. SSE流过早终止：SseStream的poll_next方法在首次调用时，无论是否收到数据都会直接返回Ready(None)，导致流被关闭。接收器rx随之销毁，后续tx.send就会触发"channel closed"错误。
2. 阻塞式读取子进程输出：run_xelatex使用标准库的read_to_string，属于阻塞IO操作，在tokio异步环境中会阻塞线程，导致编译过程中无法响应SSE流事件。
3. 错误处理逻辑错误：try_recv返回Err时直接终止流，但Err(mpsc::error::TryRecvError::Empty)仅表示当前无数据，并非通道关闭，此时不应终止流。

修复步骤与代码修改 #

1. 修正SSE Stream的poll_next逻辑

正确处理接收器状态：

• 有数据时返回数据；
• 无数据时让任务等待（返回Pending），而非终止流；
• 通道真正关闭时再返回Ready(None)终止流。

2. 改用异步方式读取子进程输出

使用tokio提供的异步进程API读取XeLaTeX输出，避免阻塞线程，同时支持实时推送日志片段。

3. 完善错误处理

区分try_recv的不同错误类型，仅在通道关闭时终止流。

修改后的完整代码 #

use std::fmt::Display;
use std::process::Stdio;

use actix_web::{web, App, HttpRequest, HttpResponse, HttpServer};
use futures::Stream;
use tokio::io::AsyncReadExt;
use tokio::process::Command;
use tokio::sync::mpsc;
use tokio::task;
use std::pin::Pin;
use std::task::{Context, Poll};

async fn run_xelatex(tx: mpsc::UnboundedSender<String>) {
    let mut cmd = Command::new("xelatex")
        .arg("-interaction=batchmode")
        .arg("/Users/xiaoqiangjiang/Nutstore/document/dolphin-book-2023/src/dolphin-book-2023.tex")
        .stdout(Stdio::piped())
        .spawn()
        .expect("Failed to execute xelatex command");

    let mut stdout = cmd.stdout.take().unwrap();
    let mut buffer = [0; 1024];

    // 异步读取子进程输出，逐段发送到通道
    while let Ok(n) = stdout.read(&mut buffer).await {
        if n == 0 {
            break;
        }
        let output = String::from_utf8_lossy(&buffer[..n]).to_string();
        if let Err(e) = tx.send(output) {
            eprintln!("Failed to send log chunk: {}", e);
            break;
        }
    }

    // 等待子进程结束
    let _ = cmd.wait().await;
}

async fn sse_endpoint(_req: HttpRequest) -> HttpResponse {
    let (tx, rx) = mpsc::unbounded_channel();
    task::spawn(async move {
        run_xelatex(tx).await;
    });

    HttpResponse::Ok()
        .content_type("text/event-stream")
        .streaming(SseStream { receiver: rx })
}

/// 简化后的SSE Stream实现
struct SseStream<T>
where
    T: Display,
{
    receiver: mpsc::UnboundedReceiver<T>,
}

impl<T: Display> Stream for SseStream<T> {
    type Item = Result<actix_web::web::Bytes, actix_web::Error>;

    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
        match self.receiver.poll_recv(cx) {
            Poll::Ready(Some(data)) => {
                // 按照SSE格式封装数据
                let message = format!("data: {}\n\n", data);
                Poll::Ready(Some(Ok(actix_web::web::Bytes::from(message))))
            }
            Poll::Ready(None) => {
                // 通道关闭，终止流
                Poll::Ready(None)
            }
            Poll::Pending => {
                // 当前无数据，等待下一次轮询
                Poll::Pending
            }
        }
    }
}

#[actix_web::main]
async fn main() -> std::io::Result<()> {
    HttpServer::new(|| App::new().route("/sse", web::get().to(sse_endpoint)))
        .bind("127.0.0.1:8080")?
        .run()
        .await
}

关键修改说明 #

• 异步读取子进程输出：使用tokio::process::Command和AsyncReadExt的read方法，异步读取XeLaTeX输出，避免阻塞线程，同时实现实时日志推送。
• 简化SSE Stream：直接利用mpsc::UnboundedReceiver的poll_recv处理异步等待，移除冗余的计数器和定时器，逻辑更清晰。
• 逐段发送日志：不再等待编译完成后一次性发送，读取到部分输出就立即推送，提升实时性。
• 正确处理通道状态：通过poll_recv的返回值区分"无数据"和"通道关闭"，避免过早终止SSE流。

重新执行curl -N http://localhost:8080/sse测试，即可正常接收XeLaTeX编译日志，不会再出现"channel closed"错误。

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