Refactor message serialization as a tokio codec. #22
Conversation
This allows us to organize all of the Bitcoin-Zcash specific parts of the protocol into a subtree.
This provides a significantly cleaner API to consumers, because it allows using adaptors that convert a TCP stream to a stream of messages, and potentially allows more efficient message handling.
|
(This is still a draft because there was one bit of cleanup I wanted to do on the |
hdevalence
force-pushed
the
tokio-codec
branch
from
d2d1650 to
2796a1a
Compare
| // XXX(HACK): this is inefficient and does an extra allocation. | ||
| // instead, we should have a size estimator for the message, reserve | ||
| // that much space, write the header (with zeroed checksum), then the body, | ||
| // then write the computed checksum in-place. for now, just do an extra alloc. |
There was a problem hiding this comment.
We could record this refactor in an issue but I think saving an extra alloc is much less important than other stuff right now. However when we do get around to doing it properly, the tokio codec setup will let us perform a single allocation per message.
| // ======== Decoding ========= | ||
|
|
||
| #[derive(Debug)] | ||
| enum DecodeState { |
There was a problem hiding this comment.
This tracks the decoder state; since decode can be called multiple times (see below) we need to track what phase (header/body) we're in and absorb the header contents (the decoder is responsible for removing parsed data from the buffer).
| type Error = Error; | ||
|
|
||
| #[instrument(skip(src))] | ||
| fn decode(&mut self, src: &mut BytesMut) -> Result<Option<Self::Item>, Self::Error> { |
There was a problem hiding this comment.
decode returns
Errto signal an error;Ok(None)to signal insufficient data;Ok(Some(msg))when an entire item is ready.
| ); | ||
|
|
||
| // Reserve buffer space for the expected body and the following header. | ||
| src.reserve(body_len + HEADER_LEN); |
There was a problem hiding this comment.
The tokio docs recommend reserving body_len + HEADER_LEN, so that no more allocations are required until after reading the next header.
| // Now that we know we have the full body, split off the body, | ||
| // and reset the decoder state for the next message. | ||
| let body = src.split_to(body_len); | ||
| self.state = DecodeState::Head; |
There was a problem hiding this comment.
After we remove the body, we have to reset the decoder state to DecodeState::Head or we'll try to read over the next message header as a body with the same type and checksum as the current message, causing a checksum error (I forgot this step at first 😓 )
There was a problem hiding this comment.
This would be good to have as a comment inline, just in case.
|
Fixed the constant placement and moved the header encoding to be prior to the body encoding. I agree that the file is larger than desirable but I'm not sure there's a great way to split up its contents. |
This is no longer required because the body reader methods have access to the version via the codec state.
…ion#22) ## What Make it so that creators can become stars/fans of eachother, "subscribe" and creators who `can_stream` can go live and their subscribers can join the "webinars" but also request the stage. ## Why Seems like it would be really cool.
Closes #20.
This lets us transform a TCP stream into an async stream of
Messagesrather than having to call send/recv directly.