如何通过期货通道将异步消息发送到接收器？

Question

我曾经能够做到这一点，但我正在努力应对0.3的期货。

下面是我从WebSocket获得的一个sink和stream:

let (mut sink, stream) = ws_stream.split();

我创建了一个在异步tokio任务之间通信的无界通道:

let (unbounded_sender, unbounded_receiver) = mpsc::unbounded::<tungstenite::Message>();

这是我被卡住的部分。我生成了一个异步任务，它应该连接无界接收器和接收器；我的想法是通过< code>unbounded_sender发送消息:

tokio::spawn(async {
    sink.send_all(&mut unbounded_receiver);
    Ok(())
});

对于＜code＞send_all＜/code＞，错误消息显示：

 " expected enum `tungstenite::protocol::message::Message`, found enum `std::result::Result`"

而且

"type mismatch resolving `<futures_channel::mpsc::UnboundedReceiver<tungstenite::protocol::message::Message> as futures_core::stream::Stream>::Item == std::result::Result<_, _>`"

而且

type mismatch resolving `<futures_channel::mpsc::UnboundedReceiver<tungstenite::protocol::message::Message> as futures_core::stream::Stream>::Item == std::result::Result<tungstenite::protocol::message::Message, tungstenite::error::Error>`

查看文档，我可以看到< code>unbounded_receiver实现了< code>TryStream特征，这是< code>send_all所需要的，所以我不确定该怎么做。

Answer 1

Futures现在需要无限通道来实现Futures_ core:：stream:：TryStream

这可以像将每条消息包装在Ok结果中一样简单地完成：

    sink.send_all(
        &mut unbounded_receiver.map(Ok::<tungstenite::Message, tungstenite::error::Error>),
    );