GRIN/grin-node
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Refactor Readable trait (#3309)

Browse Source 
Currently Writable accepts trait Write as a type parameter but Readable
takes Read as a trait object, which is not symmetrical and also less performant. This PR changes Readable trait and all places where it's used
This commit is contained in:
hashmap
2020-04-30 17:42:19 +02:00
committed by GitHub
parent 9e51e86538
commit a82041d0ed
20 changed files with 98 additions and 100 deletions
Download Patch File Download Diff File Expand all files Collapse all files
p2p/src/conn.rs
+6 -10
View File
@@ -48,9 +48,9 @@ const BODY_IO_TIMEOUT: Duration = Duration::from_millis(60000);
/// A trait to be implemented in order to receive messages from the
/// connection. Allows providing an optional response.
pub trait MessageHandler: Send + 'static {
fn consume<'a>(
fn consume<'a, R: Read>(
&self,
msg: Message<'a>,
msg: Message<'a, R>,
stopped: Arc<AtomicBool>,
tracker: Arc<Tracker>,
) -> Result<Option<Msg>, Error>;
@@ -88,18 +88,14 @@ macro_rules! try_header {
/// A message as received by the connection. Provides access to the message
/// header lazily consumes the message body, handling its deserialization.
pub struct Message<'a> {
pub struct Message<'a, R: Read> {
pub header: MsgHeader,
stream: &'a mut dyn Read,
stream: &'a mut R,
version: ProtocolVersion,
}
impl<'a> Message<'a> {
fn from_header(
header: MsgHeader,
stream: &'a mut dyn Read,
version: ProtocolVersion,
) -> Message<'a> {
impl<'a, R: Read> Message<'a, R> {
fn from_header(header: MsgHeader, stream: &'a mut R, version: ProtocolVersion) -> Self {
Message {
header,
stream,
p2p/src/msg.rs
+24 -24
View File
@@ -146,21 +146,21 @@ impl Msg {
///
/// Note: We return a MsgHeaderWrapper here as we may encounter an unknown msg type.
///
pub fn read_header(
stream: &mut dyn Read,
pub fn read_header<R: Read>(
stream: &mut R,
version: ProtocolVersion,
) -> Result<MsgHeaderWrapper, Error> {
let mut head = vec![0u8; MsgHeader::LEN];
stream.read_exact(&mut head)?;
let header = ser::deserialize::<MsgHeaderWrapper>(&mut &head[..], version)?;
let header: MsgHeaderWrapper = ser::deserialize(&mut &head[..], version)?;
Ok(header)
}
/// Read a single item from the provided stream, always blocking until we
/// have a result (or timeout).
/// Returns the item and the total bytes read.
pub fn read_item<T: Readable>(
stream: &mut dyn Read,
pub fn read_item<T: Readable, R: Read>(
stream: &mut R,
version: ProtocolVersion,
) -> Result<(T, u64), Error> {
let mut reader = StreamingReader::new(stream, version);
@@ -170,9 +170,9 @@ pub fn read_item<T: Readable>(
/// Read a message body from the provided stream, always blocking
/// until we have a result (or timeout).
pub fn read_body<T: Readable>(
pub fn read_body<T: Readable, R: Read>(
h: &MsgHeader,
stream: &mut dyn Read,
stream: &mut R,
version: ProtocolVersion,
) -> Result<T, Error> {
let mut body = vec![0u8; h.msg_len as usize];
@@ -181,15 +181,15 @@ pub fn read_body<T: Readable>(
}
/// Read (an unknown) message from the provided stream and discard it.
pub fn read_discard(msg_len: u64, stream: &mut dyn Read) -> Result<(), Error> {
pub fn read_discard<R: Read>(msg_len: u64, stream: &mut R) -> Result<(), Error> {
let mut buffer = vec![0u8; msg_len as usize];
stream.read_exact(&mut buffer)?;
Ok(())
}
/// Reads a full message from the underlying stream.
pub fn read_message<T: Readable>(
stream: &mut dyn Read,
pub fn read_message<T: Readable, R: Read>(
stream: &mut R,
version: ProtocolVersion,
msg_type: Type,
) -> Result<T, Error> {
@@ -208,8 +208,8 @@ pub fn read_message<T: Readable>(
}
}
pub fn write_message(
stream: &mut dyn Write,
pub fn write_message<W: Write>(
stream: &mut W,
msg: &Msg,
tracker: Arc<Tracker>,
) -> Result<(), Error> {
@@ -285,7 +285,7 @@ impl Writeable for MsgHeader {
}
impl Readable for MsgHeaderWrapper {
fn read(reader: &mut dyn Reader) -> Result<MsgHeaderWrapper, ser::Error> {
fn read<R: Reader>(reader: &mut R) -> Result<MsgHeaderWrapper, ser::Error> {
let m = magic();
reader.expect_u8(m[0])?;
reader.expect_u8(m[1])?;
@@ -371,7 +371,7 @@ impl Writeable for Hand {
}
impl Readable for Hand {
fn read(reader: &mut dyn Reader) -> Result<Hand, ser::Error> {
fn read<R: Reader>(reader: &mut R) -> Result<Hand, ser::Error> {
let version = ProtocolVersion::read(reader)?;
let (capab, nonce) = ser_multiread!(reader, read_u32, read_u64);
let capabilities = Capabilities::from_bits_truncate(capab);
@@ -422,7 +422,7 @@ impl Writeable for Shake {
}
impl Readable for Shake {
fn read(reader: &mut dyn Reader) -> Result<Shake, ser::Error> {
fn read<R: Reader>(reader: &mut R) -> Result<Shake, ser::Error> {
let version = ProtocolVersion::read(reader)?;
let capab = reader.read_u32()?;
let capabilities = Capabilities::from_bits_truncate(capab);
@@ -453,7 +453,7 @@ impl Writeable for GetPeerAddrs {
}
impl Readable for GetPeerAddrs {
fn read(reader: &mut dyn Reader) -> Result<GetPeerAddrs, ser::Error> {
fn read<R: Reader>(reader: &mut R) -> Result<GetPeerAddrs, ser::Error> {
let capab = reader.read_u32()?;
let capabilities = Capabilities::from_bits_truncate(capab);
Ok(GetPeerAddrs { capabilities })
@@ -478,7 +478,7 @@ impl Writeable for PeerAddrs {
}
impl Readable for PeerAddrs {
fn read(reader: &mut dyn Reader) -> Result<PeerAddrs, ser::Error> {
fn read<R: Reader>(reader: &mut R) -> Result<PeerAddrs, ser::Error> {
let peer_count = reader.read_u32()?;
if peer_count > MAX_PEER_ADDRS {
return Err(ser::Error::TooLargeReadErr);
@@ -510,7 +510,7 @@ impl Writeable for PeerError {
}
impl Readable for PeerError {
fn read(reader: &mut dyn Reader) -> Result<PeerError, ser::Error> {
fn read<R: Reader>(reader: &mut R) -> Result<PeerError, ser::Error> {
let code = reader.read_u32()?;
let msg = reader.read_bytes_len_prefix()?;
let message = String::from_utf8(msg).map_err(|_| ser::Error::CorruptedData)?;
@@ -538,7 +538,7 @@ impl Writeable for Locator {
}
impl Readable for Locator {
fn read(reader: &mut dyn Reader) -> Result<Locator, ser::Error> {
fn read<R: Reader>(reader: &mut R) -> Result<Locator, ser::Error> {
let len = reader.read_u8()?;
if len > (MAX_LOCATORS as u8) {
return Err(ser::Error::TooLargeReadErr);
@@ -583,7 +583,7 @@ impl Writeable for Ping {
}
impl Readable for Ping {
fn read(reader: &mut dyn Reader) -> Result<Ping, ser::Error> {
fn read<R: Reader>(reader: &mut R) -> Result<Ping, ser::Error> {
let total_difficulty = Difficulty::read(reader)?;
let height = reader.read_u64()?;
Ok(Ping {
@@ -610,7 +610,7 @@ impl Writeable for Pong {
}
impl Readable for Pong {
fn read(reader: &mut dyn Reader) -> Result<Pong, ser::Error> {
fn read<R: Reader>(reader: &mut R) -> Result<Pong, ser::Error> {
let total_difficulty = Difficulty::read(reader)?;
let height = reader.read_u64()?;
Ok(Pong {
@@ -635,7 +635,7 @@ impl Writeable for BanReason {
}
impl Readable for BanReason {
fn read(reader: &mut dyn Reader) -> Result<BanReason, ser::Error> {
fn read<R: Reader>(reader: &mut R) -> Result<BanReason, ser::Error> {
let ban_reason_i32 = match reader.read_i32() {
Ok(h) => h,
Err(_) => 0,
@@ -665,7 +665,7 @@ impl Writeable for TxHashSetRequest {
}
impl Readable for TxHashSetRequest {
fn read(reader: &mut dyn Reader) -> Result<TxHashSetRequest, ser::Error> {
fn read<R: Reader>(reader: &mut R) -> Result<TxHashSetRequest, ser::Error> {
Ok(TxHashSetRequest {
hash: Hash::read(reader)?,
height: reader.read_u64()?,
@@ -693,7 +693,7 @@ impl Writeable for TxHashSetArchive {
}
impl Readable for TxHashSetArchive {
fn read(reader: &mut dyn Reader) -> Result<TxHashSetArchive, ser::Error> {
fn read<R: Reader>(reader: &mut R) -> Result<TxHashSetArchive, ser::Error> {
let hash = Hash::read(reader)?;
let (height, bytes) = ser_multiread!(reader, read_u64, read_u64);
p2p/src/protocol.rs
+3 -3
View File
@@ -25,7 +25,7 @@ use chrono::prelude::Utc;
use rand::{thread_rng, Rng};
use std::cmp;
use std::fs::{self, File, OpenOptions};
use std::io::BufWriter;
use std::io::{BufWriter, Read};
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::Arc;
use std::time::Instant;
@@ -51,9 +51,9 @@ impl Protocol {
}
impl MessageHandler for Protocol {
fn consume(
fn consume<R: Read>(
&self,
mut msg: Message,
mut msg: Message<R>,
stopped: Arc<AtomicBool>,
tracker: Arc<Tracker>,
) -> Result<Option<Msg>, Error> {
p2p/src/store.rs
+1 -1
View File
@@ -75,7 +75,7 @@ impl Writeable for PeerData {
}
impl Readable for PeerData {
fn read(reader: &mut dyn Reader) -> Result<PeerData, ser::Error> {
fn read<R: Reader>(reader: &mut R) -> Result<PeerData, ser::Error> {
let addr = PeerAddr::read(reader)?;
let capab = reader.read_u32()?;
let ua = reader.read_bytes_len_prefix()?;
p2p/src/types.rs
+1 -1
View File
@@ -138,7 +138,7 @@ impl Writeable for PeerAddr {
}
impl Readable for PeerAddr {
fn read(reader: &mut dyn Reader) -> Result<PeerAddr, ser::Error> {
fn read<R: Reader>(reader: &mut R) -> Result<PeerAddr, ser::Error> {
let v4_or_v6 = reader.read_u8()?;
if v4_or_v6 == 0 {
let ip = reader.read_fixed_bytes(4)?;