tokio_core::Framed for tokio_tungstenite?

[bitdivision]

I'm attempting to implement websockets for the rumqtt MQTT library and I'm running into some issues using this crate. My main problem is mentioned here:
AtherEnergy/rumqtt#70

But what I'd like to do is create an instance of tokio_core::Framed (to keep consistency with the rest of rumqtt) from a WebSocketStream. I am unsure how I could implement Read / AsyncRead for WebSocketStream, and was wondering if you had any suggestions?

I can't really see a way to do it with the current interfaces, since as far as I can tell, WebSocketStream is essentially equivalent to Framed, but without the ability to add encoders and decoders

I'm extremely new to tokio, so forgive me if I've completely misunderstood anything here.

[daniel-abramov]

Well, the Framed instance does not belong to tokio_core anymore (your link points to the old version of tokio). It's part of tokio_io as far as I know. To create a Framed instance you have to implement Encoder and Decoder traits for the corresponding objects.

Encoder trait: https://docs.rs/tokio-io/0.1/tokio_io/codec/trait.Encoder.html
Decoder trait: https://docs.rs/tokio-io/0.1/tokio_io/codec/trait.Decoder.html

In theory, based on the interface of the trait I would assume that it's quite easy to implement both of them for the WebSocketStream, as it basically has the same semantics.

The reason why we did not implement Encoder and Decoder is that both of them were primarily used by the users of tokio-proto crate. The current implementation of tokio-tungstenite is only dependant on tokio_core and tokio_io, i.e. the one can extend it easy to use tokio-proto if needed. If we developed it the other way around, it would be less flexible.

I'm not sure if we're going to implement both of these traits in the near future, I'll have to check the current state of tokio, it seems that the current direction of tokio went in the direction opposite of tokio-proto. There is already a [relatively new] issue about compatibility with new tokio versions (#33), so I'll try to check the current state and adapt tokio-tungstenite to work with the latest changes when I have time.

[jeffesquivels]

Hi! I'm also interested in being able to use Encoder and Decoder (they seem to have been moved to the tokio-codec crate now) to be able to convert to/from WebSocket's binary messages from/to my application's messages.

I'm very new to Rust and Tokio, so I'm not sure if I understood correctly, but it seems like this would require me to implement a Codec struct for which I would need to implement the Encoder and Decoder traits, and then I should apply that Codec to WebSocketStream through the tokio-codec::Framed::new() associated function.

The main problem I found with this is that it would require WebSocketStream to implement AsyncRead and AsyncWrite, which it currently doesn't; furthermore, as @bitdivision commented originally, it seems like WebSocketStream is equivalent to tokio-codec::Framed in the sense that it also implements Stream and Sink from an underlying type that implements AsyncRead and AsyncWrite, so I don't know if what I want to do is actually possible.

Any idea if this is possible? And if so, how could be implemented? Thanks!

[daniel-abramov]

@jeffesquivels Framed and WebSocketStream are indeed similar (if not the same). When you create a Framed object, you specify the stream and the object which is capable of doing encoding / decoding. Basically the WebSocketStream does the same job -- you give the stream to it and the object itself represents something that can encode/decode the stream.

If you really want to have the Framed type, then the best thing to do would probably adding the method into_framed() to WebSocketStream which constructs the corresponding Framed object. Actually the only reason to use Framed is an interoperability with the rest of the system in case if you have some codebase which makes use of Framed. I don't see any other scenarios, where Framed might be useful.

The reason why Framed should be constructed from WebSocketStream is that there are basically at least 2 ways to create a WebSocketStream: with and without the handshake. Depending on that, the underlying logic either performs a handshake or reads / writes messages right away. That's why there are 2 available functions in WebSocketStream: from_raw_socket() and from_partially_read().

As for the Encoder and Decoder traits, you cannot implement them right away, because they work with an abstract buffer and they write to some abstract buffer as well, while the current WebSocket type encapsulates an abstract stream within itself. I.e. if you want to implement these traits, you probably have to look at tungstenite-rs (our core of tokio-tungstenite), there you can find all functions which perform the reading / writing, but currently the FrameSocket contains the stream inside it, i.e. manages it in order to make it efficient. This part of tungstenite-rs may be written in a different way / refactored, so that the functions become more abstract and the buffer is passed to them.

I'm also interested in being able to use Encoder and Decoder (they seem to have been moved to the tokio-codec crate now) to be able to convert to/from WebSocket's binary messages from/to my application's messages.

Actually you can use them in this way even with the current state of tokio-tungstenite. In order to convert something to the websocket message, just construct the message (see Message structure) and pass it to the WebSocketStream. Likewise you can read messages from an abstract stream which you pass to the constructor of the WebSocketStream.

[jeffesquivels]

@application-developer-DA thanks for your response! I didn't quite understand the last part (i.e. how it could be implemented in the current state), if you have any examples of projects that are using tokio-tungstenite in this way I would appreciate it, may be taking a look to their code would help me understand better.

I think my main problem is that one of my application's message is composed of one or more WebSocket messages, which means sometimes I will need to wait for more WS messages to arrive before I can return one application message. I was thinking may be the right approach is to have my own struct that implements the Stream trait by calling WebSocketStream's poll() underneath and just buffering the WS messages until I have enough of them to be able to return an application message. Do you think that would work with tokio-tungstenite? Thanks!

[daniel-abramov]

I didn't quite understand the last part (i.e. how it could be implemented in the current state), if you have any examples of projects that are using tokio-tungstenite in this way I would appreciate it, may be taking a look to their code would help me understand better.

I mean you could split the WebSocketStream into 2 parts (sending and receiving part) and use them the same way you would use Framed, i.e. reading incoming messages from Stream like part and writing to Sink. The only example I could "officially" provide at the moment is the one from our "examples" section: https://github.com/snapview/tokio-tungstenite/blob/master/examples/server.rs#L71

I think my main problem is that one of my application's message is composed of one or more WebSocket messages, which means sometimes I will need to wait for more WS messages to arrive before I can return one application message. I was thinking may be the right approach is to have my own struct that implements the Stream trait by calling WebSocketStream's poll() underneath and just buffering the WS messages until I have enough of them to be able to return an application message. Do you think that would work with tokio-tungstenite? Thanks!

Indeed, this is one of the approaches you may use! You may also want to to implement Sink for your wrapper.

[jeffesquivels]

I mean you could split the WebSocketStream into 2 parts (sending and receiving part) and use them the same way you would use Framed, i.e. reading incoming messages from Stream like part and writing to Sink. The only example I could "officially" provide at the moment is the one from our "examples" section: https://github.com/snapview/tokio-tungstenite/blob/master/examples/server.rs#L71

Ok, got it now.

Indeed, this is one of the approaches you may use! You may also want to to implement Sink for your wrapper.

Nice, thanks! I wanted to make sure trying this approach made sense; And yes, I will need Sink too for sure :-).

[svanharmelen]

@daniel-abramov I've noticed your comment in #260 and have been reading the discussion here, but I fail to fully understand what is possible and what is not 😊

I'm mainly looking to implement a codec to encode and decode data being send over a connected websocket into a custom message format.

As far as I understand, it's not possible to send raw binary data over the websocket as tokio-tungstenite already encodes and decodes the raw data into well-known websocket messages.

So I guess the next best thing is to put our raw data into a websocket binary message (so as binary data). But as this is a high throughput connection, I wonder what this means for the overhead? Looking at the websocket specs I guess that the encoding/decoding done by tungstenite-rs is pretty light weight, but I'm not sure...

Additionally I'm not sure how I can (if possible at all) use our own encoder/decoder as a "wrapper" around tokio-tungstenite or tungstenite-rs to encode/decode in a steaming fashion? I the most straight forward solution I can think of seems closer to using something like bincode or serde-binary instead of using an encoder/decoder.

So reading the messages from the stream as they come; check if it's a binary message; grab the payload and decode/de-serialize it using bincode or serde-binary... While that would probably work, if doesn't like it's the most efficient way to do this. So do you have any other suggestions that might be more efficient and stream-like?

@jeffesquivels can you maybe share what you ended up with?

Thanks for both your time!

[daniel-abramov]

I'm mainly looking to implement a codec to encode and decode data being send over a connected websocket into a custom message format.

That's totally possible (have always been 🙂).

As far as I understand, it's not possible to send raw binary data over the websocket as tokio-tungstenite already encodes and decodes the raw data into well-known websocket messages.

What do you mean by raw binary data? - With WebSockets, you can send any type of data over the connection. I.e. if test messages are not what you want, you could use the binary data message to send any arbitrary Vec<u8>.

So I guess the next best thing is to put our raw data into a websocket binary message (so as binary data). But as this is a high throughput connection, I wonder what this means for the overhead? Looking at the websocket specs I guess that the encoding/decoding done by tungstenite-rs is pretty light weight, but I'm not sure...

Right. I would say the throughput degradation is pretty negligible for the majority of use cases (especially if you don't have any alternative). That said, if the only thing that you need is to send a stream of binary data to the server and you have the possibility to choose any protocol, then the WebSockets might not necessarily be the best option here (I would say a plain TCP stream or QUIC would be a better choice here).

So reading the messages from the stream as they come; check if it's a binary message; grab the payload and decode/de-serialize it using bincode or serde-binary...

Yep, this sounds sensible. Though you could actually elegantly combine them as tokio-tungstenite implements the AsyncStream trait, if you have a generic encoder or decoder, you could potentially combine them together.

While that would probably work, if doesn't like it's the most efficient way to do this. So do you have any other suggestions that might be more efficient and stream-like?

I think w.r.t. the protocol, WebSockets might not necessary be the most performant option here (in general), irrespective of the library that you choose to use for the WebSocket. I.e. if you have a choice, you could choose a more well-suited protocol. However, depending on your requirements, using the approach that you described may not be that bad actually. I think the majority fo the users send some data over WebSockets that they end up serializing/deserializing in some format (e.g. usage of protocol buffers together with tungstenite is not unheard of).

[svanharmelen]

What do you mean by raw binary data? - With WebSockets, you can send any type of data over the connection. I.e. if test messages are not what you want, you could use the binary data message to send any arbitrary Vec.

I meant that I cannot "just" send raw bytes over the wire (like with a TcpStream), but that (like you also described) my raw bytes need to be encapsulated in a websocket binary data message.

Though you could actually elegantly combine them as tokio-tungstenite implements the AsyncStream trait, if you have a generic encoder or decoder, you could potentially combine them together

So this is probably the most important and interesting part (for me)... We already have an implementation of the Tokio encoder and decoder traits for our protocol, but I don't understand (I probably leak some general knowledge around this topic) how to combine those parts?

Is there a way to get a "stream of payloads" from tungstenite? Or do we need to write a little glue somewhere in between to somehow achieve that? Or...?

I understand that I might be asking for a lot, but any additional info or any pointers to an example would be highly appreciated!!

WebSockets might not necessary be the most performant option here (in general), irrespective of the library that you choose to use for the WebSocket. I.e. if you have a choice, you could choose a more well-suited protocol.

I'm afraid we have some constraints that cannot be met with a regular TCP stream or QUIC, so we are kind of forced to use websockets here 😏

[svanharmelen]

After discussing our requirements once more, we came tot the conclusion that we can use a TcpStream after all. So while I'm still interested to better understand the above (for possible future use cases), for now this isn't blocking for us anymore.

Thanks for triggering us to rethink/check our requirements, as the end result is much better now!

[daniel-abramov]

Sorry for the late reply, glad that you found a more appropriate solution meanwhile!

So this is probably the most important and interesting part (for me)... We already have an implementation of the Tokio encoder and decoder traits for our protocol, but I don't understand (I probably leak some general knowledge around this topic) how to combine those parts?

I see. Probably the easiest way would be wrapping the sink together with the encoder and the stream with the decoder.

WebSocketStream implements Stream and Sink, so you could use split() to get the reading and writing halves.

You could wrap your decoder along with the reading part of WebSocketStream that you get after the split() and implement the Stream trait for this wrapper, so that in the wrapper you first poll the read part and once it resolves, you pass it to your decoder implementation and return the result.

Or even something simpler like this:

let (read, write) = websocket.split();

let decoded_stream = read.map(|msg| decoder.decode(msg.into_bytes_mut()));

That way you turn a stream of Item=tungstenite::Result<tungstenite::Message> into the stream with the Item=your_crate::Result<YourDecoder::Item>.

[svanharmelen]

Thank you very much for the additional info @daniel-abramov! Very helpful and I'm sure I'm going to need/use this somewhere in the near future. Thanks!

[conorbros]

Just incase anyone comes upon this later, a small example of using this library with tokio codecs: https://github.com/conorbros/websocket-experiment