websocket - 在BusReader和Warp WebSocket接收器之间转发消息会留下未占用的缓冲区？

Question

我正在尝试创建一个websocket服务器(以及HTTP，因此使用warp)，该服务器通过websocket将消息从一个源(MQTT订阅)转发到许多客户端。除了客户端在广播第二条消息之前没有收到第一条Websocket消息之外，这似乎基本上可以正常工作。然后总是留下一条消息，直到最后再也没有收到最后一条消息。对我来说，问题似乎是在ws_connected函数中从未完全刷新过的发送缓冲区。
我使用futures::stream::iter将BusReader转换为流，然后将消息映射到WebSocket Sink所需的所需Ok(Message)类型。官方的warp websocket聊天示例使用类似的构造在流之间转发:https://github.com/seanmonstar/warp/blob/42fd14fdab8145d27ae770fe4b5c843a99bc2a44/examples/websockets_chat.rs#L62。
在这个简化的示例中，服务器通过总线广播值0-9。一个websocat客户端(和Firefox中的JS websocket客户端)接收到0-8消息-尽管总是在广播和服务器标准输出之后出现-但9个消息永远不会到达。 async_bus_print函数会按时接收所有值，这证明消息至少可以毫无问题地通过总线。
这是服务器进程的输出:

async bus_print started
0
async bus: "0"
1
async bus: "1"
2
async bus: "2"
3
async bus: "3"
4
async bus: "4"
5
async bus: "5"
6
async bus: "6"
7
async bus: "7"
8
async bus: "8"
9
async bus: "9"

有问题的代码:

use std::{sync::{Arc, RwLock}, thread};

use bus::{Bus, BusReader};
use futures::StreamExt;
use warp::ws::{Message, WebSocket};
use warp::Filter;

async fn ws_connected(ws: WebSocket, rx: BusReader<String>) {
    let (ws_tx, _ws_rx) = ws.split();

    thread::spawn(|| {
        futures::executor::block_on(async move {
            if let Err(e) = futures::stream::iter(rx.into_iter())
                .map(|ws_msg| Ok(Message::text(ws_msg)))
                .forward(ws_tx)
                .await
            {
                eprintln!("Goodbye, websocket user: {}", e);
            }
        });
    });
}

async fn async_bus_print(mut rx: BusReader<String>) {
    println!("async bus_print started");
    thread::spawn(||
        futures::executor::block_on(async move {
            while let Some(msg) = futures::stream::iter(rx.iter()).next().await {
                println!("async bus: {:#?}", msg);
            }
        })
    );
}

async fn bus_tx(tx: Arc<RwLock<Bus<String>>>) {
    for i in 0..10u8 {
        tx.write().unwrap().broadcast(format!("{}", i));
        println!("{}", i);
        tokio::time::delay_for(std::time::Duration::from_secs(1)).await;
    }
}

#[tokio::main]
async fn main() {
    let bus = Arc::new(RwLock::new(Bus::new(20)));
    let bus2 = Arc::clone(&bus);
    let rx = warp::any().map(move || bus2.write().unwrap().add_rx());
    let rx2 = bus.write().unwrap().add_rx();

    let ws = warp::path("ws")
        .and(warp::ws())
        .and(rx)
        .map(|ws: warp::ws::Ws, rx| ws.on_upgrade(move |socket| ws_connected(socket, rx)));

    futures::join!(
        async_bus_print(rx2),
        bus_tx(bus),
        warp::serve(ws).run(([127, 0, 0, 1], 3030)),
    );
}

如何跟踪并消除此“缓冲”问题？
希望我已经解释得足够好了。请让我知道是否可以提供更多信息。谢谢你的帮助。

Answer 1

尽管我仍然无法弄清数据未刷新的根本原因，但由于有一些乐于助人的人在Reddit上工作，我有一些更好的替代解决方案。
如果您不首先拆分WebSocket，它似乎可以正常工作:

async fn ws_connected(ws: WebSocket, rx: BusReader<String>) {
    thread::spawn(|| {
        futures::executor::block_on(async move {
            if let Err(e) = futures::stream::iter(rx.into_iter())
                .map(|ws_msg| Ok(Message::text(ws_msg)))
                .forward(ws)
                .await
            {
                eprintln!("Goodbye, websocket user: {}", e);
            }
        });
    });
}

如果您在代码中包含 send，则仍可以拆分流并将数据 forward(而不是 futures::SinkExt)分离到WebSocket的tx端:

async fn ws_connected(ws: WebSocket, mut rx: BusReader<String>) {
    use futures::SinkExt;

    let (mut ws_tx, ws_rx) = ws.split();

    while let Ok(msg) = rx.recv() {
        ws_tx.send(Message::text(msg)).await.unwrap();
        // `send` automatically flushes the sink
    }
}

最终，我认为tokio的 broadcast channel 比 bus更适合我的异步多生产者，多消费者 channel 通信需求:

use futures::StreamExt;
use tokio::sync::broadcast;
use warp::ws::{Message, WebSocket};
use warp::Filter;

async fn ws_connected(ws: WebSocket, recv: broadcast::Receiver<String>) {
    let (ws_tx, _ws_rx) = ws.split();

    recv.map(|s| Ok(Message::text(s.unwrap())))
        .forward(ws_tx)
        .await
        .unwrap();
}

async fn async_bus_print(mut recv: broadcast::Receiver<String>) {
    println!("async bus_print started");
    while let Some(msg) = recv.next().await {
        println!("async bus: {:#?}", msg.unwrap());
    }
}

async fn bus_tx(tx: broadcast::Sender<String>) {
    for i in 0..10u8 {
        tx.send(format!("{}", i)).unwrap();
        println!("{}", i);
        tokio::time::delay_for(std::time::Duration::from_secs(1)).await;
    }
}

#[tokio::main]
async fn main() {
    let (send, recv) = broadcast::channel::<String>(32);
    let send2 = send.clone();
    let rx = warp::any().map(move || send2.subscribe());

    let ws = warp::path("ws")
        .and(warp::ws())
        .and(rx)
        .map(|ws: warp::ws::Ws, recv| ws.on_upgrade(move |socket| ws_connected(socket, recv)));

    let (abp, tx, warp) = futures::join!(
        tokio::spawn(async_bus_print(recv)),
        tokio::spawn(bus_tx(send)),
        tokio::spawn(warp::serve(ws).run(([127, 0, 0, 1], 3030))),
    );

    abp.unwrap();
    tx.unwrap();
    warp.unwrap();
}