Implement Sink for WebSocketConnection

[mankinskin]

No description provided.

[yoshuawuyts]

Hi @mankinskin, thanks for opening this PR! Unfortunately I don't think this may be the right direction for the library; we don't use the Sink trait anywhere in the http-rs and async-rs ecosystems. We believe async iteration is generally preferable, with Sink currently unclear whether it'll ever be stabilized since it doesn't yet have a synchronous counterpart.

Either way, I don't think this PR quite fits the direction we envision for the ecosystem right now -- though the effort is very much appreciated, and we hope you don't feel discouraged from contributing again in the future!

[mankinskin]

hope you don't feel discouraged from contributing again in the future!

Not at all! Thanks for your detailed response.

We believe async iteration is generally preferable.

Is there any reason you prefer async iteration? I have found the Sink trait very useful usually, because it can be used to pass output handles to functions, for example. Also all kinds of channels use Sinks.
Also, the syntax is much more concise:

stream.map(f)
      .forward(sink)
      .await

vs

while let Some(item) = stream.next().await {
    sink.send(f(item)).await;
}

[yoshuawuyts]

Async iteration can be as short as calling for_each:

stream.map(f)
    .for_each(sink.send)
    .await;

This can all be done without a dedicated Sink trait.

[mankinskin]

Okay you are right about that. But what about the use with generic interfaces?

The reason I implemented this is because I would like to pass WebSocketConnection to a function:

// websocket.rs
async fn connection<M, Rx, Tx>(mut rx: Rx, tx: Tx)
    where M: Serialize + for<'de> Deserialize<'de> + Send,
          Rx: Stream<Item=M> + Send + 'static + Unpin,
          Tx: Sink<M> + Send + 'static,
{
    // process incoming items from rx and write results to tx
}

// tide_server.rs
async fn wss(request: Request<()>) -> tide::Result {
    WebSocket::new(async move |_, ws| {
        let (sink, stream) = ws.split(); // because Sink & Stream are implemented, ws implements StreamExt::split()!
        let stream = stream.map(|msg| msg.map(|msg| msg.into_data()));
        let sink = sink.with(async move |msg| Ok(Message::from(msg)) as Result<_, tide_websockets::Error>);

        websocket::connection(stream, sink).await; // this function works with any Stream and Sink!

        Ok(())
    })
    .call(request)
    .await
}

This is one example, where I can reuse websocket::connection for any channels implementing Stream and Sink. This is useful when you are switching the websocket library for example. But there are other examples where you may not want to code towards a single concrete type but may want to abstract over any types of Sinks.

[cryptoquick]

Async iteration can be as short as calling for_each:

stream.map(f)
    .for_each(sink.send)
    .await;

This can all be done without a dedicated Sink trait.

This doesn't work for me. It really helps to be able to treat them separately if sending stream iterators across threads. Otherwise the state locks overlap. I really don't understand why you keep rejecting these really helpful Sink PRs. It's a pattern that's quite common across the Rust ecosystem, and for good reason.

[mankinskin]

Yes I also don't really understand the reasoning behind not using Sink. It seems like the perfect application this trait was meant to be used for.

[cryptoquick]

I've published a fork including your work, @mankinskin . Hope this is of help to others in the community.

https://crates.io/crates/tide-websockets-sink