| @ -1,83 +0,0 @@ | |||||
| use std::fmt; | |||||
| use std::io; | |||||
| use bytes::BytesMut; | |||||
| use tokio_io::codec::length_delimited::{Builder, Framed}; | |||||
| use tokio_io::{AsyncRead, AsyncWrite}; | |||||
| use crate::proto::peer; | |||||
| use crate::proto::{ProtoDecode, ProtoDecoder, ProtoEncode, ProtoEncoder, ServerRequest, ServerResponse}; | |||||
| fn decode_frame<'a, T>(frame_type: &str, bytes: &'a mut BytesMut) -> io::Result<T> | |||||
| where | |||||
| T: ProtoDecode + fmt::Debug, | |||||
| { | |||||
| let mut decoder = ProtoDecoder::new(&*bytes); | |||||
| let frame = decoder.decode()?; | |||||
| if decoder.has_remaining() { | |||||
| warn!( | |||||
| "Received {} with trailing bytes. Frame:\n{:?}Bytes:{:?}", | |||||
| frame_type, | |||||
| frame, | |||||
| decoder.bytes() | |||||
| ); | |||||
| } | |||||
| Ok(frame) | |||||
| } | |||||
| fn encode_frame<T: ProtoEncode>(frame: &T) -> io::Result<BytesMut> { | |||||
| let mut bytes = BytesMut::new(); | |||||
| frame.encode(&mut ProtoEncoder::new(&mut bytes))?; | |||||
| Ok(bytes) | |||||
| } | |||||
| /// Wraps a raw byte async I/O object, providing it with the ability to read and | |||||
| /// write entire frames at once. | |||||
| /// The returned stream and sink of frames is intended to be combined (using | |||||
| /// `Stream::and_then` and `Sink::with`) with the following decoding and | |||||
| /// encoding functions to create a stream/sink of decoded messages. | |||||
| pub fn new_framed<T: AsyncRead + AsyncWrite>(io: T) -> Framed<T, BytesMut> { | |||||
| Builder::new() | |||||
| .length_field_length(4) | |||||
| .little_endian() | |||||
| .new_framed(io) | |||||
| } | |||||
| /// Decodes a server response from the given byte buffer, which should contain | |||||
| /// exactly the bytes encoding the returned response. | |||||
| /// Intended to be used on the result of `new_framed` using `Stream::and_then`. | |||||
| pub fn decode_server_response(bytes: &mut BytesMut) -> io::Result<ServerResponse> { | |||||
| decode_frame("server response", bytes) | |||||
| } | |||||
| /// Encodes the given server response into a byte buffer, then returns it. | |||||
| /// Intended to be used on a sink of BytesMut objects, using `Sink::with`. | |||||
| pub fn encode_server_response(response: &ServerResponse) -> io::Result<BytesMut> { | |||||
| encode_frame(response) | |||||
| } | |||||
| /// Decodes a server request from the given byte buffer, which should contain | |||||
| /// exactly the bytes encoding the returned response. | |||||
| /// Intended to be used on the result of `new_framed` using `Stream::and_then`. | |||||
| pub fn decode_server_request(bytes: &mut BytesMut) -> io::Result<ServerRequest> { | |||||
| decode_frame("server request", bytes) | |||||
| } | |||||
| /// Encodes the given server request into a byte buffer, then returns it. | |||||
| /// Intended to be used on a sink of BytesMut objects, using `Sink::with`. | |||||
| pub fn encode_server_request(request: &ServerRequest) -> io::Result<BytesMut> { | |||||
| encode_frame(request) | |||||
| } | |||||
| /// Decodes a peer message from the given byte buffer, which should contain | |||||
| /// exactly the bytes encoding the returned response. | |||||
| /// Intended to be used on the result of `new_framed` using `Stream::and_then`. | |||||
| pub fn decode_peer_message(bytes: &mut BytesMut) -> io::Result<peer::Message> { | |||||
| decode_frame("peer message", bytes) | |||||
| } | |||||
| /// Encodes the given peer message into a byte buffer, then returns it. | |||||
| /// Intended to be used on a sink of BytesMut objects, using `Sink::with`. | |||||
| pub fn encode_peer_message(message: &peer::Message) -> io::Result<BytesMut> { | |||||
| encode_frame(message) | |||||
| } | |||||