rust - 如何使用 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 .通过强制类型(使用您的注释)，您强制编译器寻找 Into<WebsocketStream> 的实现。对于 IterOk , 没有。

结果，这失败了。

这个类型提示是完全没有必要的 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是否纯粹用于强制类型并且在实践中永远不会被调用。