如何使用 tokio-tungstenite 的 send_all 方法发送 websocket 消息流

Question

我正在使用优秀的 rust websocket 库 tokio-tungstenite，它只是 tungstenite 的 tokio 版本。

我可以使用 connect_async 方法轻松连接到 websocket 流，在返回的 websocket 流中，我将其拆分如下：

connect_async(url).and_then(move |(ws_stream, _)| {

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

我可以使用这样的发送消息轻松地向接收器发送一条消息：

sink.send(tungstenite::Message::Text("my message".to_string()))
.map_err(|e| ())
.wait();

这只是一个人为的例子。我的问题是我想向接收器发送消息流，我了解到最好使用 send_all 方法来完成。我的尝试是：

let my_messsages : Vec<tungstenite::Message> = vec![
tungstenite::Message::Text("message_1".to_string()), 
tungstenite::Message::Text("message_2".to_string()) ];

let send_stream: tokio_tungstenite::WebSocketStream
<tokio_tungstenite::tungstenite::Message> 
= futures::stream::iter_ok(my_messages);

sink.send_all(stream).map_err(|e| ()).wait();

但它不起作用，因为 futures::stream::iter_ok 不是构造我需要的流的正确方法，这是我卡住的地方，无法从文档中看到如何继续。

编辑：最初我没有为 send_stream 指定类型注释，它给了我这个错误：

type annotations needed for `futures::stream::iter_ok::IterOk
<std::vec::IntoIter<tungstenite::protocol::message::Message>, E>`

cannot infer type for `E`

当前错误也与发送流的类型注释有关。

Answer 1

你的问题就在这一行：

let send_stream:tokio_tungstenite::WebSocketStream<tokio_tungstenite::tungstenite::Message> 
    = futures::stream::iter_ok(my_messages);

futures::stream::iter_ok的定义如下：

pub fn iter_ok<I, E>(i: I) -> IterOk<<I as IntoIterator>::IntoIter, E> 
where
    I: IntoIterator, 

请注意，这个新流是IterOk，而不是tungstenite::WebsocketStream。通过强制类型（使用您的注释），您正在强制编译器为 IterOk 寻找 Into<WebsocketStream> 的实现，但没有。

因此，这失败了。

这种类型提示完全没有必要，因为 send_all 是为 impl Sink<SinkItem = _, SinkError = _> 定义的。

不用太意外，换行为：

let send_stream = futures::stream::iter_ok(my_messages)
  .map_err(|_:()| tungstenite::error::Error::Utf8)

应该可以完美运行，正如您的接收器所期望的 tungstenite::Message 并且这个新流具有它作为一个元素（因为它是从 Vec<tungstenite::Message> 构建的）。 map_err 纯粹是为了强制类型，并且在实践中永远不会被调用。