Merge branch 'master' into spec-v0.12

beacon_node/eth2_libp2p/src/rpc/codec/base.rs

@@ -0,0 +1,217 @@
+//! This handles the various supported encoding mechanism for the Eth 2.0 RPC.
+
+use crate::rpc::{RPCCodedResponse, RPCRequest, RPCResponse};
+use libp2p::bytes::BufMut;
+use libp2p::bytes::BytesMut;
+use std::marker::PhantomData;
+use tokio_util::codec::{Decoder, Encoder};
+use types::EthSpec;
+
+pub trait OutboundCodec<TItem>: Encoder<TItem> + Decoder {
+    type ErrorType;
+
+    fn decode_error(
+        &mut self,
+        src: &mut BytesMut,
+    ) -> Result<Option<Self::ErrorType>, <Self as Decoder>::Error>;
+}
+
+/* Global Inbound Codec */
+// This deals with Decoding RPC Requests from other peers and encoding our responses
+
+pub struct BaseInboundCodec<TCodec, TSpec>
+where
+    TCodec: Encoder<RPCCodedResponse<TSpec>> + Decoder,
+    TSpec: EthSpec,
+{
+    /// Inner codec for handling various encodings
+    inner: TCodec,
+    phantom: PhantomData<TSpec>,
+}
+
+impl<TCodec, TSpec> BaseInboundCodec<TCodec, TSpec>
+where
+    TCodec: Encoder<RPCCodedResponse<TSpec>> + Decoder,
+    TSpec: EthSpec,
+{
+    pub fn new(codec: TCodec) -> Self {
+        BaseInboundCodec {
+            inner: codec,
+            phantom: PhantomData,
+        }
+    }
+}
+
+/* Global Outbound Codec */
+// This deals with Decoding RPC Responses from other peers and encoding our requests
+pub struct BaseOutboundCodec<TOutboundCodec, TSpec>
+where
+    TOutboundCodec: OutboundCodec<RPCRequest<TSpec>>,
+    TSpec: EthSpec,
+{
+    /// Inner codec for handling various encodings.
+    inner: TOutboundCodec,
+    /// Keeps track of the current response code for a chunk.
+    current_response_code: Option<u8>,
+    phantom: PhantomData<TSpec>,
+}
+
+impl<TOutboundCodec, TSpec> BaseOutboundCodec<TOutboundCodec, TSpec>
+where
+    TSpec: EthSpec,
+    TOutboundCodec: OutboundCodec<RPCRequest<TSpec>>,
+{
+    pub fn new(codec: TOutboundCodec) -> Self {
+        BaseOutboundCodec {
+            inner: codec,
+            current_response_code: None,
+            phantom: PhantomData,
+        }
+    }
+}
+
+/* Implementation of the Encoding/Decoding for the global codecs */
+
+/* Base Inbound Codec */
+
+// This Encodes RPC Responses sent to external peers
+impl<TCodec, TSpec> Encoder<RPCCodedResponse<TSpec>> for BaseInboundCodec<TCodec, TSpec>
+where
+    TSpec: EthSpec,
+    TCodec: Decoder + Encoder<RPCCodedResponse<TSpec>>,
+{
+    type Error = <TCodec as Encoder<RPCCodedResponse<TSpec>>>::Error;
+
+    fn encode(
+        &mut self,
+        item: RPCCodedResponse<TSpec>,
+        dst: &mut BytesMut,
+    ) -> Result<(), Self::Error> {
+        dst.clear();
+        dst.reserve(1);
+        dst.put_u8(
+            item.as_u8()
+                .expect("Should never encode a stream termination"),
+        );
+        self.inner.encode(item, dst)
+    }
+}
+
+// This Decodes RPC Requests from external peers
+impl<TCodec, TSpec> Decoder for BaseInboundCodec<TCodec, TSpec>
+where
+    TSpec: EthSpec,
+    TCodec: Encoder<RPCCodedResponse<TSpec>> + Decoder<Item = RPCRequest<TSpec>>,
+{
+    type Item = RPCRequest<TSpec>;
+    type Error = <TCodec as Decoder>::Error;
+
+    fn decode(&mut self, src: &mut BytesMut) -> Result<Option<Self::Item>, Self::Error> {
+        self.inner.decode(src)
+    }
+}
+
+/* Base Outbound Codec */
+
+// This Encodes RPC Requests sent to external peers
+impl<TCodec, TSpec> Encoder<RPCRequest<TSpec>> for BaseOutboundCodec<TCodec, TSpec>
+where
+    TSpec: EthSpec,
+    TCodec: OutboundCodec<RPCRequest<TSpec>> + Encoder<RPCRequest<TSpec>>,
+{
+    type Error = <TCodec as Encoder<RPCRequest<TSpec>>>::Error;
+
+    fn encode(&mut self, item: RPCRequest<TSpec>, dst: &mut BytesMut) -> Result<(), Self::Error> {
+        self.inner.encode(item, dst)
+    }
+}
+
+// This decodes RPC Responses received from external peers
+impl<TCodec, TSpec> Decoder for BaseOutboundCodec<TCodec, TSpec>
+where
+    TSpec: EthSpec,
+    TCodec:
+        OutboundCodec<RPCRequest<TSpec>, ErrorType = String> + Decoder<Item = RPCResponse<TSpec>>,
+{
+    type Item = RPCCodedResponse<TSpec>;
+    type Error = <TCodec as Decoder>::Error;
+
+    fn decode(&mut self, src: &mut BytesMut) -> Result<Option<Self::Item>, Self::Error> {
+        // if we have only received the response code, wait for more bytes
+        if src.len() <= 1 {
+            return Ok(None);
+        }
+        // using the response code determine which kind of payload needs to be decoded.
+        let response_code = self.current_response_code.unwrap_or_else(|| {
+            let resp_code = src.split_to(1)[0];
+            self.current_response_code = Some(resp_code);
+            resp_code
+        });
+
+        let inner_result = {
+            if RPCCodedResponse::<TSpec>::is_response(response_code) {
+                // decode an actual response and mutates the buffer if enough bytes have been read
+                // returning the result.
+                self.inner
+                    .decode(src)
+                    .map(|r| r.map(RPCCodedResponse::Success))
+            } else {
+                // decode an error
+                self.inner
+                    .decode_error(src)
+                    .map(|r| r.map(|resp| RPCCodedResponse::from_error(response_code, resp)))
+            }
+        };
+        // if the inner decoder was capable of decoding a chunk, we need to reset the current
+        // response code for the next chunk
+        if let Ok(Some(_)) = inner_result {
+            self.current_response_code = None;
+        }
+        // return the result
+        inner_result
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::super::ssz::*;
+    use super::super::ssz_snappy::*;
+    use super::*;
+    use crate::rpc::protocol::*;
+
+    #[test]
+    fn test_decode_status_message() {
+        let message = hex::decode("ff060000734e615070590032000006e71e7b54989925efd6c9cbcb8ceb9b5f71216f5137282bf6a1e3b50f64e42d6c7fb347abe07eb0db8200000005029e2800").unwrap();
+        let mut buf = BytesMut::new();
+        buf.extend_from_slice(&message);
+
+        type Spec = types::MainnetEthSpec;
+
+        let snappy_protocol_id =
+            ProtocolId::new(Protocol::Status, Version::V1, Encoding::SSZSnappy);
+        let ssz_protocol_id = ProtocolId::new(Protocol::Status, Version::V1, Encoding::SSZ);
+
+        let mut snappy_outbound_codec =
+            SSZSnappyOutboundCodec::<Spec>::new(snappy_protocol_id, 1_048_576);
+        let mut ssz_outbound_codec = SSZOutboundCodec::<Spec>::new(ssz_protocol_id, 1_048_576);
+
+        // decode message just as snappy message
+        let snappy_decoded_message = snappy_outbound_codec.decode(&mut buf.clone());
+        // decode message just a ssz message
+        let ssz_decoded_message = ssz_outbound_codec.decode(&mut buf.clone());
+
+        // build codecs for entire chunk
+        let mut snappy_base_outbound_codec = BaseOutboundCodec::new(snappy_outbound_codec);
+        let mut ssz_base_outbound_codec = BaseOutboundCodec::new(ssz_outbound_codec);
+
+        // decode message as ssz snappy chunk
+        let snappy_decoded_chunk = snappy_base_outbound_codec.decode(&mut buf.clone());
+        // decode message just a ssz chunk
+        let ssz_decoded_chunk = ssz_base_outbound_codec.decode(&mut buf.clone());
+
+        let _ = dbg!(snappy_decoded_message);
+        let _ = dbg!(ssz_decoded_message);
+        let _ = dbg!(snappy_decoded_chunk);
+        let _ = dbg!(ssz_decoded_chunk);
+    }
+}

beacon_node/eth2_libp2p/src/rpc/codec/mod.rs

@@ -0,0 +1,69 @@
+pub(crate) mod base;
+pub(crate) mod ssz;
+pub(crate) mod ssz_snappy;
+
+use self::base::{BaseInboundCodec, BaseOutboundCodec};
+use self::ssz::{SSZInboundCodec, SSZOutboundCodec};
+use self::ssz_snappy::{SSZSnappyInboundCodec, SSZSnappyOutboundCodec};
+use crate::rpc::protocol::RPCError;
+use crate::rpc::{RPCCodedResponse, RPCRequest};
+use libp2p::bytes::BytesMut;
+use tokio_util::codec::{Decoder, Encoder};
+use types::EthSpec;
+
+// Known types of codecs
+pub enum InboundCodec<TSpec: EthSpec> {
+    SSZSnappy(BaseInboundCodec<SSZSnappyInboundCodec<TSpec>, TSpec>),
+    SSZ(BaseInboundCodec<SSZInboundCodec<TSpec>, TSpec>),
+}
+
+pub enum OutboundCodec<TSpec: EthSpec> {
+    SSZSnappy(BaseOutboundCodec<SSZSnappyOutboundCodec<TSpec>, TSpec>),
+    SSZ(BaseOutboundCodec<SSZOutboundCodec<TSpec>, TSpec>),
+}
+
+impl<T: EthSpec> Encoder<RPCCodedResponse<T>> for InboundCodec<T> {
+    type Error = RPCError;
+
+    fn encode(&mut self, item: RPCCodedResponse<T>, dst: &mut BytesMut) -> Result<(), Self::Error> {
+        match self {
+            InboundCodec::SSZ(codec) => codec.encode(item, dst),
+            InboundCodec::SSZSnappy(codec) => codec.encode(item, dst),
+        }
+    }
+}
+
+impl<TSpec: EthSpec> Decoder for InboundCodec<TSpec> {
+    type Item = RPCRequest<TSpec>;
+    type Error = RPCError;
+
+    fn decode(&mut self, src: &mut BytesMut) -> Result<Option<Self::Item>, Self::Error> {
+        match self {
+            InboundCodec::SSZ(codec) => codec.decode(src),
+            InboundCodec::SSZSnappy(codec) => codec.decode(src),
+        }
+    }
+}
+
+impl<TSpec: EthSpec> Encoder<RPCRequest<TSpec>> for OutboundCodec<TSpec> {
+    type Error = RPCError;
+
+    fn encode(&mut self, item: RPCRequest<TSpec>, dst: &mut BytesMut) -> Result<(), Self::Error> {
+        match self {
+            OutboundCodec::SSZ(codec) => codec.encode(item, dst),
+            OutboundCodec::SSZSnappy(codec) => codec.encode(item, dst),
+        }
+    }
+}
+
+impl<T: EthSpec> Decoder for OutboundCodec<T> {
+    type Item = RPCCodedResponse<T>;
+    type Error = RPCError;
+
+    fn decode(&mut self, src: &mut BytesMut) -> Result<Option<Self::Item>, Self::Error> {
+        match self {
+            OutboundCodec::SSZ(codec) => codec.decode(src),
+            OutboundCodec::SSZSnappy(codec) => codec.decode(src),
+        }
+    }
+}

beacon_node/eth2_libp2p/src/rpc/codec/ssz.rs

@@ -0,0 +1,326 @@
+use crate::rpc::methods::*;
+use crate::rpc::{
+    codec::base::OutboundCodec,
+    protocol::{
+        Encoding, Protocol, ProtocolId, RPCError, Version, BLOCKS_BY_ROOT_REQUEST_MAX,
+        BLOCKS_BY_ROOT_REQUEST_MIN, SIGNED_BEACON_BLOCK_MAX, SIGNED_BEACON_BLOCK_MIN,
+    },
+};
+use crate::rpc::{RPCCodedResponse, RPCRequest, RPCResponse};
+use libp2p::bytes::{BufMut, Bytes, BytesMut};
+use ssz::{Decode, Encode};
+use ssz_types::VariableList;
+use std::marker::PhantomData;
+use tokio_util::codec::{Decoder, Encoder};
+use types::{EthSpec, SignedBeaconBlock};
+use unsigned_varint::codec::UviBytes;
+
+/* Inbound Codec */
+
+pub struct SSZInboundCodec<TSpec: EthSpec> {
+    inner: UviBytes,
+    protocol: ProtocolId,
+    phantom: PhantomData<TSpec>,
+}
+
+impl<TSpec: EthSpec> SSZInboundCodec<TSpec> {
+    pub fn new(protocol: ProtocolId, max_packet_size: usize) -> Self {
+        let mut uvi_codec = UviBytes::default();
+        uvi_codec.set_max_len(max_packet_size);
+
+        // this encoding only applies to ssz.
+        debug_assert_eq!(protocol.encoding, Encoding::SSZ);
+
+        SSZInboundCodec {
+            inner: uvi_codec,
+            protocol,
+            phantom: PhantomData,
+        }
+    }
+}
+
+// Encoder for inbound streams: Encodes RPC Responses sent to peers.
+impl<TSpec: EthSpec> Encoder<RPCCodedResponse<TSpec>> for SSZInboundCodec<TSpec> {
+    type Error = RPCError;
+
+    fn encode(
+        &mut self,
+        item: RPCCodedResponse<TSpec>,
+        dst: &mut BytesMut,
+    ) -> Result<(), Self::Error> {
+        let bytes = match item {
+            RPCCodedResponse::Success(resp) => match resp {
+                RPCResponse::Status(res) => res.as_ssz_bytes(),
+                RPCResponse::BlocksByRange(res) => res.as_ssz_bytes(),
+                RPCResponse::BlocksByRoot(res) => res.as_ssz_bytes(),
+                RPCResponse::Pong(res) => res.data.as_ssz_bytes(),
+                RPCResponse::MetaData(res) => res.as_ssz_bytes(),
+            },
+            RPCCodedResponse::InvalidRequest(err) => err.as_ssz_bytes(),
+            RPCCodedResponse::ServerError(err) => err.as_ssz_bytes(),
+            RPCCodedResponse::Unknown(err) => err.as_ssz_bytes(),
+            RPCCodedResponse::StreamTermination(_) => {
+                unreachable!("Code error - attempting to encode a stream termination")
+            }
+        };
+        if !bytes.is_empty() {
+            // length-prefix and return
+            return self
+                .inner
+                .encode(Bytes::from(bytes), dst)
+                .map_err(RPCError::from);
+        } else {
+            // payload is empty, add a 0-byte length prefix
+            dst.reserve(1);
+            dst.put_u8(0);
+        }
+        Ok(())
+    }
+}
+
+// Decoder for inbound streams: Decodes RPC requests from peers
+impl<TSpec: EthSpec> Decoder for SSZInboundCodec<TSpec> {
+    type Item = RPCRequest<TSpec>;
+    type Error = RPCError;
+
+    fn decode(&mut self, src: &mut BytesMut) -> Result<Option<Self::Item>, Self::Error> {
+        match self.inner.decode(src).map_err(RPCError::from) {
+            Ok(Some(packet)) => match self.protocol.message_name {
+                Protocol::Status => match self.protocol.version {
+                    Version::V1 => {
+                        if packet.len() == <StatusMessage as Encode>::ssz_fixed_len() {
+                            Ok(Some(RPCRequest::Status(StatusMessage::from_ssz_bytes(
+                                &packet,
+                            )?)))
+                        } else {
+                            Err(RPCError::InvalidData)
+                        }
+                    }
+                },
+                Protocol::Goodbye => match self.protocol.version {
+                    Version::V1 => {
+                        if packet.len() == <GoodbyeReason as Encode>::ssz_fixed_len() {
+                            Ok(Some(RPCRequest::Goodbye(GoodbyeReason::from_ssz_bytes(
+                                &packet,
+                            )?)))
+                        } else {
+                            Err(RPCError::InvalidData)
+                        }
+                    }
+                },
+                Protocol::BlocksByRange => match self.protocol.version {
+                    Version::V1 => {
+                        if packet.len() == <BlocksByRangeRequest as Encode>::ssz_fixed_len() {
+                            Ok(Some(RPCRequest::BlocksByRange(
+                                BlocksByRangeRequest::from_ssz_bytes(&packet)?,
+                            )))
+                        } else {
+                            Err(RPCError::InvalidData)
+                        }
+                    }
+                },
+                Protocol::BlocksByRoot => match self.protocol.version {
+                    Version::V1 => {
+                        if packet.len() >= *BLOCKS_BY_ROOT_REQUEST_MIN
+                            && packet.len() <= *BLOCKS_BY_ROOT_REQUEST_MAX
+                        {
+                            Ok(Some(RPCRequest::BlocksByRoot(BlocksByRootRequest {
+                                block_roots: VariableList::from_ssz_bytes(&packet)?,
+                            })))
+                        } else {
+                            Err(RPCError::InvalidData)
+                        }
+                    }
+                },
+                Protocol::Ping => match self.protocol.version {
+                    Version::V1 => {
+                        if packet.len() == <Ping as Encode>::ssz_fixed_len() {
+                            Ok(Some(RPCRequest::Ping(Ping {
+                                data: u64::from_ssz_bytes(&packet)?,
+                            })))
+                        } else {
+                            Err(RPCError::InvalidData)
+                        }
+                    }
+                },
+                Protocol::MetaData => match self.protocol.version {
+                    Version::V1 => {
+                        if packet.len() > 0 {
+                            Err(RPCError::InvalidData)
+                        } else {
+                            Ok(Some(RPCRequest::MetaData(PhantomData)))
+                        }
+                    }
+                },
+            },
+            Ok(None) => Ok(None),
+            Err(e) => Err(e),
+        }
+    }
+}
+
+/* Outbound Codec: Codec for initiating RPC requests */
+
+pub struct SSZOutboundCodec<TSpec: EthSpec> {
+    inner: UviBytes,
+    protocol: ProtocolId,
+    phantom: PhantomData<TSpec>,
+}
+
+impl<TSpec: EthSpec> SSZOutboundCodec<TSpec> {
+    pub fn new(protocol: ProtocolId, max_packet_size: usize) -> Self {
+        let mut uvi_codec = UviBytes::default();
+        uvi_codec.set_max_len(max_packet_size);
+
+        // this encoding only applies to ssz.
+        debug_assert_eq!(protocol.encoding, Encoding::SSZ);
+
+        SSZOutboundCodec {
+            inner: uvi_codec,
+            protocol,
+            phantom: PhantomData,
+        }
+    }
+}
+
+// Encoder for outbound streams: Encodes RPC Requests to peers
+impl<TSpec: EthSpec> Encoder<RPCRequest<TSpec>> for SSZOutboundCodec<TSpec> {
+    type Error = RPCError;
+
+    fn encode(&mut self, item: RPCRequest<TSpec>, dst: &mut BytesMut) -> Result<(), Self::Error> {
+        let bytes = match item {
+            RPCRequest::Status(req) => req.as_ssz_bytes(),
+            RPCRequest::Goodbye(req) => req.as_ssz_bytes(),
+            RPCRequest::BlocksByRange(req) => req.as_ssz_bytes(),
+            RPCRequest::BlocksByRoot(req) => req.block_roots.as_ssz_bytes(),
+            RPCRequest::Ping(req) => req.as_ssz_bytes(),
+            RPCRequest::MetaData(_) => return Ok(()), // no metadata to encode
+        };
+        // length-prefix
+        self.inner
+            .encode(libp2p::bytes::Bytes::from(bytes), dst)
+            .map_err(RPCError::from)
+    }
+}
+
+// Decoder for outbound streams: Decodes RPC responses from peers.
+//
+// The majority of the decoding has now been pushed upstream due to the changing specification.
+// We prefer to decode blocks and attestations with extra knowledge about the chain to perform
+// faster verification checks before decoding entire blocks/attestations.
+impl<TSpec: EthSpec> Decoder for SSZOutboundCodec<TSpec> {
+    type Item = RPCResponse<TSpec>;
+    type Error = RPCError;
+
+    fn decode(&mut self, src: &mut BytesMut) -> Result<Option<Self::Item>, Self::Error> {
+        if src.len() == 1 && src[0] == 0_u8 {
+            // the object is empty. We return the empty object if this is the case
+            // clear the buffer and return an empty object
+            src.clear();
+            match self.protocol.message_name {
+                Protocol::Status => match self.protocol.version {
+                    Version::V1 => Err(RPCError::IncompleteStream), // cannot have an empty HELLO message. The stream has terminated unexpectedly
+                },
+                Protocol::Goodbye => Err(RPCError::InvalidData),
+                Protocol::BlocksByRange => match self.protocol.version {
+                    Version::V1 => Err(RPCError::IncompleteStream), // cannot have an empty block message.
+                },
+                Protocol::BlocksByRoot => match self.protocol.version {
+                    Version::V1 => Err(RPCError::IncompleteStream), // cannot have an empty block message.
+                },
+                Protocol::Ping => match self.protocol.version {
+                    Version::V1 => Err(RPCError::IncompleteStream), // cannot have an empty block message.
+                },
+                Protocol::MetaData => match self.protocol.version {
+                    Version::V1 => Err(RPCError::IncompleteStream), // cannot have an empty block message.
+                },
+            }
+        } else {
+            match self.inner.decode(src).map_err(RPCError::from) {
+                Ok(Some(mut packet)) => {
+                    // take the bytes from the buffer
+                    let raw_bytes = packet.split();
+
+                    match self.protocol.message_name {
+                        Protocol::Status => match self.protocol.version {
+                            Version::V1 => {
+                                if raw_bytes.len() == <StatusMessage as Encode>::ssz_fixed_len() {
+                                    Ok(Some(RPCResponse::Status(StatusMessage::from_ssz_bytes(
+                                        &raw_bytes,
+                                    )?)))
+                                } else {
+                                    Err(RPCError::InvalidData)
+                                }
+                            }
+                        },
+                        Protocol::Goodbye => Err(RPCError::InvalidData),
+                        Protocol::BlocksByRange => match self.protocol.version {
+                            Version::V1 => {
+                                if raw_bytes.len() >= *SIGNED_BEACON_BLOCK_MIN
+                                    && raw_bytes.len() <= *SIGNED_BEACON_BLOCK_MAX
+                                {
+                                    Ok(Some(RPCResponse::BlocksByRange(Box::new(
+                                        SignedBeaconBlock::from_ssz_bytes(&raw_bytes)?,
+                                    ))))
+                                } else {
+                                    Err(RPCError::InvalidData)
+                                }
+                            }
+                        },
+                        Protocol::BlocksByRoot => match self.protocol.version {
+                            Version::V1 => {
+                                if raw_bytes.len() >= *SIGNED_BEACON_BLOCK_MIN
+                                    && raw_bytes.len() <= *SIGNED_BEACON_BLOCK_MAX
+                                {
+                                    Ok(Some(RPCResponse::BlocksByRoot(Box::new(
+                                        SignedBeaconBlock::from_ssz_bytes(&raw_bytes)?,
+                                    ))))
+                                } else {
+                                    Err(RPCError::InvalidData)
+                                }
+                            }
+                        },
+                        Protocol::Ping => match self.protocol.version {
+                            Version::V1 => {
+                                if raw_bytes.len() == <Ping as Encode>::ssz_fixed_len() {
+                                    Ok(Some(RPCResponse::Pong(Ping {
+                                        data: u64::from_ssz_bytes(&raw_bytes)?,
+                                    })))
+                                } else {
+                                    Err(RPCError::InvalidData)
+                                }
+                            }
+                        },
+                        Protocol::MetaData => match self.protocol.version {
+                            Version::V1 => {
+                                if raw_bytes.len() == <MetaData<TSpec> as Encode>::ssz_fixed_len() {
+                                    Ok(Some(RPCResponse::MetaData(MetaData::from_ssz_bytes(
+                                        &raw_bytes,
+                                    )?)))
+                                } else {
+                                    Err(RPCError::InvalidData)
+                                }
+                            }
+                        },
+                    }
+                }
+                Ok(None) => Ok(None), // waiting for more bytes
+                Err(e) => Err(e),
+            }
+        }
+    }
+}
+
+impl<TSpec: EthSpec> OutboundCodec<RPCRequest<TSpec>> for SSZOutboundCodec<TSpec> {
+    type ErrorType = String;
+
+    fn decode_error(&mut self, src: &mut BytesMut) -> Result<Option<Self::ErrorType>, RPCError> {
+        match self.inner.decode(src).map_err(RPCError::from) {
+            Ok(Some(packet)) => Ok(Some(
+                String::from_utf8_lossy(&<Vec<u8>>::from_ssz_bytes(&packet)?).into(),
+            )),
+            Ok(None) => Ok(None),
+            Err(e) => Err(e),
+        }
+    }
+}

beacon_node/eth2_libp2p/src/rpc/codec/ssz_snappy.rs

@@ -0,0 +1,424 @@
+use crate::rpc::methods::*;
+use crate::rpc::{
+    codec::base::OutboundCodec,
+    protocol::{
+        Encoding, Protocol, ProtocolId, RPCError, Version, BLOCKS_BY_ROOT_REQUEST_MAX,
+        BLOCKS_BY_ROOT_REQUEST_MIN, SIGNED_BEACON_BLOCK_MAX, SIGNED_BEACON_BLOCK_MIN,
+    },
+};
+use crate::rpc::{RPCCodedResponse, RPCRequest, RPCResponse};
+use libp2p::bytes::BytesMut;
+use snap::read::FrameDecoder;
+use snap::write::FrameEncoder;
+use ssz::{Decode, Encode};
+use ssz_types::VariableList;
+use std::io::Cursor;
+use std::io::ErrorKind;
+use std::io::{Read, Write};
+use std::marker::PhantomData;
+use tokio_util::codec::{Decoder, Encoder};
+use types::{EthSpec, SignedBeaconBlock};
+use unsigned_varint::codec::Uvi;
+
+/* Inbound Codec */
+
+pub struct SSZSnappyInboundCodec<TSpec: EthSpec> {
+    protocol: ProtocolId,
+    inner: Uvi<usize>,
+    len: Option<usize>,
+    /// Maximum bytes that can be sent in one req/resp chunked responses.
+    max_packet_size: usize,
+    phantom: PhantomData<TSpec>,
+}
+
+impl<T: EthSpec> SSZSnappyInboundCodec<T> {
+    pub fn new(protocol: ProtocolId, max_packet_size: usize) -> Self {
+        let uvi_codec = Uvi::default();
+        // this encoding only applies to ssz_snappy.
+        debug_assert_eq!(protocol.encoding, Encoding::SSZSnappy);
+
+        SSZSnappyInboundCodec {
+            inner: uvi_codec,
+            protocol,
+            len: None,
+            phantom: PhantomData,
+            max_packet_size,
+        }
+    }
+}
+
+// Encoder for inbound streams: Encodes RPC Responses sent to peers.
+impl<TSpec: EthSpec> Encoder<RPCCodedResponse<TSpec>> for SSZSnappyInboundCodec<TSpec> {
+    type Error = RPCError;
+
+    fn encode(
+        &mut self,
+        item: RPCCodedResponse<TSpec>,
+        dst: &mut BytesMut,
+    ) -> Result<(), Self::Error> {
+        let bytes = match item {
+            RPCCodedResponse::Success(resp) => match resp {
+                RPCResponse::Status(res) => res.as_ssz_bytes(),
+                RPCResponse::BlocksByRange(res) => res.as_ssz_bytes(),
+                RPCResponse::BlocksByRoot(res) => res.as_ssz_bytes(),
+                RPCResponse::Pong(res) => res.data.as_ssz_bytes(),
+                RPCResponse::MetaData(res) => res.as_ssz_bytes(),
+            },
+            RPCCodedResponse::InvalidRequest(err) => err.as_ssz_bytes(),
+            RPCCodedResponse::ServerError(err) => err.as_ssz_bytes(),
+            RPCCodedResponse::Unknown(err) => err.as_ssz_bytes(),
+            RPCCodedResponse::StreamTermination(_) => {
+                unreachable!("Code error - attempting to encode a stream termination")
+            }
+        };
+        // SSZ encoded bytes should be within `max_packet_size`
+        if bytes.len() > self.max_packet_size {
+            return Err(RPCError::InternalError(
+                "attempting to encode data > max_packet_size".into(),
+            ));
+        }
+        // Inserts the length prefix of the uncompressed bytes into dst
+        // encoded as a unsigned varint
+        self.inner
+            .encode(bytes.len(), dst)
+            .map_err(RPCError::from)?;
+
+        let mut writer = FrameEncoder::new(Vec::new());
+        writer.write_all(&bytes).map_err(RPCError::from)?;
+        writer.flush().map_err(RPCError::from)?;
+
+        // Write compressed bytes to `dst`
+        dst.extend_from_slice(writer.get_ref());
+        Ok(())
+    }
+}
+
+// Decoder for inbound streams: Decodes RPC requests from peers
+impl<TSpec: EthSpec> Decoder for SSZSnappyInboundCodec<TSpec> {
+    type Item = RPCRequest<TSpec>;
+    type Error = RPCError;
+
+    fn decode(&mut self, src: &mut BytesMut) -> Result<Option<Self::Item>, Self::Error> {
+        if self.len.is_none() {
+            // Decode the length of the uncompressed bytes from an unsigned varint
+            match self.inner.decode(src).map_err(RPCError::from)? {
+                Some(length) => {
+                    self.len = Some(length);
+                }
+                None => return Ok(None), // need more bytes to decode length
+            }
+        };
+
+        let length = self.len.expect("length should be Some");
+
+        // Should not attempt to decode rpc chunks with length > max_packet_size
+        if length > self.max_packet_size {
+            return Err(RPCError::InvalidData);
+        }
+        let mut reader = FrameDecoder::new(Cursor::new(&src));
+        let mut decoded_buffer = vec![0; length];
+
+        match reader.read_exact(&mut decoded_buffer) {
+            Ok(()) => {
+                // `n` is how many bytes the reader read in the compressed stream
+                let n = reader.get_ref().position();
+                self.len = None;
+                let _read_bytes = src.split_to(n as usize);
+                match self.protocol.message_name {
+                    Protocol::Status => match self.protocol.version {
+                        Version::V1 => {
+                            if decoded_buffer.len() == <StatusMessage as Encode>::ssz_fixed_len() {
+                                Ok(Some(RPCRequest::Status(StatusMessage::from_ssz_bytes(
+                                    &decoded_buffer,
+                                )?)))
+                            } else {
+                                Err(RPCError::InvalidData)
+                            }
+                        }
+                    },
+                    Protocol::Goodbye => match self.protocol.version {
+                        Version::V1 => {
+                            if decoded_buffer.len() == <GoodbyeReason as Encode>::ssz_fixed_len() {
+                                Ok(Some(RPCRequest::Goodbye(GoodbyeReason::from_ssz_bytes(
+                                    &decoded_buffer,
+                                )?)))
+                            } else {
+                                Err(RPCError::InvalidData)
+                            }
+                        }
+                    },
+                    Protocol::BlocksByRange => match self.protocol.version {
+                        Version::V1 => {
+                            if decoded_buffer.len()
+                                == <BlocksByRangeRequest as Encode>::ssz_fixed_len()
+                            {
+                                Ok(Some(RPCRequest::BlocksByRange(
+                                    BlocksByRangeRequest::from_ssz_bytes(&decoded_buffer)?,
+                                )))
+                            } else {
+                                Err(RPCError::InvalidData)
+                            }
+                        }
+                    },
+                    Protocol::BlocksByRoot => match self.protocol.version {
+                        Version::V1 => {
+                            if decoded_buffer.len() >= *BLOCKS_BY_ROOT_REQUEST_MIN
+                                && decoded_buffer.len() <= *BLOCKS_BY_ROOT_REQUEST_MAX
+                            {
+                                Ok(Some(RPCRequest::BlocksByRoot(BlocksByRootRequest {
+                                    block_roots: VariableList::from_ssz_bytes(&decoded_buffer)?,
+                                })))
+                            } else {
+                                Err(RPCError::InvalidData)
+                            }
+                        }
+                    },
+                    Protocol::Ping => match self.protocol.version {
+                        Version::V1 => {
+                            if decoded_buffer.len() == <Ping as Encode>::ssz_fixed_len() {
+                                Ok(Some(RPCRequest::Ping(Ping {
+                                    data: u64::from_ssz_bytes(&decoded_buffer)?,
+                                })))
+                            } else {
+                                Err(RPCError::InvalidData)
+                            }
+                        }
+                    },
+                    Protocol::MetaData => match self.protocol.version {
+                        Version::V1 => {
+                            if decoded_buffer.len() > 0 {
+                                Err(RPCError::InvalidData)
+                            } else {
+                                Ok(Some(RPCRequest::MetaData(PhantomData)))
+                            }
+                        }
+                    },
+                }
+            }
+            Err(e) => match e.kind() {
+                // Haven't received enough bytes to decode yet
+                // TODO: check if this is the only Error variant where we return `Ok(None)`
+                ErrorKind::UnexpectedEof => {
+                    return Ok(None);
+                }
+                _ => return Err(e).map_err(RPCError::from),
+            },
+        }
+    }
+}
+
+/* Outbound Codec: Codec for initiating RPC requests */
+pub struct SSZSnappyOutboundCodec<TSpec: EthSpec> {
+    inner: Uvi<usize>,
+    len: Option<usize>,
+    protocol: ProtocolId,
+    /// Maximum bytes that can be sent in one req/resp chunked responses.
+    max_packet_size: usize,
+    phantom: PhantomData<TSpec>,
+}
+
+impl<TSpec: EthSpec> SSZSnappyOutboundCodec<TSpec> {
+    pub fn new(protocol: ProtocolId, max_packet_size: usize) -> Self {
+        let uvi_codec = Uvi::default();
+        // this encoding only applies to ssz_snappy.
+        debug_assert_eq!(protocol.encoding, Encoding::SSZSnappy);
+
+        SSZSnappyOutboundCodec {
+            inner: uvi_codec,
+            protocol,
+            max_packet_size,
+            len: None,
+            phantom: PhantomData,
+        }
+    }
+}
+
+// Encoder for outbound streams: Encodes RPC Requests to peers
+impl<TSpec: EthSpec> Encoder<RPCRequest<TSpec>> for SSZSnappyOutboundCodec<TSpec> {
+    type Error = RPCError;
+
+    fn encode(&mut self, item: RPCRequest<TSpec>, dst: &mut BytesMut) -> Result<(), Self::Error> {
+        let bytes = match item {
+            RPCRequest::Status(req) => req.as_ssz_bytes(),
+            RPCRequest::Goodbye(req) => req.as_ssz_bytes(),
+            RPCRequest::BlocksByRange(req) => req.as_ssz_bytes(),
+            RPCRequest::BlocksByRoot(req) => req.block_roots.as_ssz_bytes(),
+            RPCRequest::Ping(req) => req.as_ssz_bytes(),
+            RPCRequest::MetaData(_) => return Ok(()), // no metadata to encode
+        };
+        // SSZ encoded bytes should be within `max_packet_size`
+        if bytes.len() > self.max_packet_size {
+            return Err(RPCError::InternalError(
+                "attempting to encode data > max_packet_size",
+            ));
+        }
+
+        // Inserts the length prefix of the uncompressed bytes into dst
+        // encoded as a unsigned varint
+        self.inner
+            .encode(bytes.len(), dst)
+            .map_err(RPCError::from)?;
+
+        let mut writer = FrameEncoder::new(Vec::new());
+        writer.write_all(&bytes).map_err(RPCError::from)?;
+        writer.flush().map_err(RPCError::from)?;
+
+        // Write compressed bytes to `dst`
+        dst.extend_from_slice(writer.get_ref());
+        Ok(())
+    }
+}
+
+// Decoder for outbound streams: Decodes RPC responses from peers.
+//
+// The majority of the decoding has now been pushed upstream due to the changing specification.
+// We prefer to decode blocks and attestations with extra knowledge about the chain to perform
+// faster verification checks before decoding entire blocks/attestations.
+impl<TSpec: EthSpec> Decoder for SSZSnappyOutboundCodec<TSpec> {
+    type Item = RPCResponse<TSpec>;
+    type Error = RPCError;
+
+    fn decode(&mut self, src: &mut BytesMut) -> Result<Option<Self::Item>, Self::Error> {
+        if self.len.is_none() {
+            // Decode the length of the uncompressed bytes from an unsigned varint
+            match self.inner.decode(src).map_err(RPCError::from)? {
+                Some(length) => {
+                    self.len = Some(length as usize);
+                }
+                None => return Ok(None), // need more bytes to decode length
+            }
+        };
+
+        let length = self.len.expect("length should be Some");
+
+        // Should not attempt to decode rpc chunks with length > max_packet_size
+        if length > self.max_packet_size {
+            return Err(RPCError::InvalidData);
+        }
+        let mut reader = FrameDecoder::new(Cursor::new(&src));
+        let mut decoded_buffer = vec![0; length];
+        match reader.read_exact(&mut decoded_buffer) {
+            Ok(()) => {
+                // `n` is how many bytes the reader read in the compressed stream
+                let n = reader.get_ref().position();
+                self.len = None;
+                let _read_byts = src.split_to(n as usize);
+                match self.protocol.message_name {
+                    Protocol::Status => match self.protocol.version {
+                        Version::V1 => {
+                            if decoded_buffer.len() == <StatusMessage as Encode>::ssz_fixed_len() {
+                                Ok(Some(RPCResponse::Status(StatusMessage::from_ssz_bytes(
+                                    &decoded_buffer,
+                                )?)))
+                            } else {
+                                Err(RPCError::InvalidData)
+                            }
+                        }
+                    },
+                    Protocol::Goodbye => Err(RPCError::InvalidData),
+                    Protocol::BlocksByRange => match self.protocol.version {
+                        Version::V1 => {
+                            if decoded_buffer.len() >= *SIGNED_BEACON_BLOCK_MIN
+                                && decoded_buffer.len() <= *SIGNED_BEACON_BLOCK_MAX
+                            {
+                                Ok(Some(RPCResponse::BlocksByRange(Box::new(
+                                    SignedBeaconBlock::from_ssz_bytes(&decoded_buffer)?,
+                                ))))
+                            } else {
+                                Err(RPCError::InvalidData)
+                            }
+                        }
+                    },
+                    Protocol::BlocksByRoot => match self.protocol.version {
+                        Version::V1 => {
+                            if decoded_buffer.len() >= *SIGNED_BEACON_BLOCK_MIN
+                                && decoded_buffer.len() <= *SIGNED_BEACON_BLOCK_MAX
+                            {
+                                Ok(Some(RPCResponse::BlocksByRoot(Box::new(
+                                    SignedBeaconBlock::from_ssz_bytes(&decoded_buffer)?,
+                                ))))
+                            } else {
+                                Err(RPCError::InvalidData)
+                            }
+                        }
+                    },
+                    Protocol::Ping => match self.protocol.version {
+                        Version::V1 => {
+                            if decoded_buffer.len() == <Ping as Encode>::ssz_fixed_len() {
+                                Ok(Some(RPCResponse::Pong(Ping {
+                                    data: u64::from_ssz_bytes(&decoded_buffer)?,
+                                })))
+                            } else {
+                                Err(RPCError::InvalidData)
+                            }
+                        }
+                    },
+                    Protocol::MetaData => match self.protocol.version {
+                        Version::V1 => {
+                            if decoded_buffer.len() == <MetaData<TSpec> as Encode>::ssz_fixed_len()
+                            {
+                                Ok(Some(RPCResponse::MetaData(MetaData::from_ssz_bytes(
+                                    &decoded_buffer,
+                                )?)))
+                            } else {
+                                Err(RPCError::InvalidData)
+                            }
+                        }
+                    },
+                }
+            }
+            Err(e) => match e.kind() {
+                // Haven't received enough bytes to decode yet
+                // TODO: check if this is the only Error variant where we return `Ok(None)`
+                ErrorKind::UnexpectedEof => {
+                    return Ok(None);
+                }
+                _ => return Err(e).map_err(RPCError::from),
+            },
+        }
+    }
+}
+
+impl<TSpec: EthSpec> OutboundCodec<RPCRequest<TSpec>> for SSZSnappyOutboundCodec<TSpec> {
+    type ErrorType = String;
+
+    fn decode_error(&mut self, src: &mut BytesMut) -> Result<Option<Self::ErrorType>, RPCError> {
+        if self.len.is_none() {
+            // Decode the length of the uncompressed bytes from an unsigned varint
+            match self.inner.decode(src).map_err(RPCError::from)? {
+                Some(length) => {
+                    self.len = Some(length as usize);
+                }
+                None => return Ok(None), // need more bytes to decode length
+            }
+        };
+
+        let length = self.len.expect("length should be Some");
+
+        // Should not attempt to decode rpc chunks with length > max_packet_size
+        if length > self.max_packet_size {
+            return Err(RPCError::InvalidData);
+        }
+        let mut reader = FrameDecoder::new(Cursor::new(&src));
+        let mut decoded_buffer = vec![0; length];
+        match reader.read_exact(&mut decoded_buffer) {
+            Ok(()) => {
+                // `n` is how many bytes the reader read in the compressed stream
+                let n = reader.get_ref().position();
+                self.len = None;
+                let _read_bytes = src.split_to(n as usize);
+                Ok(Some(
+                    String::from_utf8_lossy(&<Vec<u8>>::from_ssz_bytes(&decoded_buffer)?).into(),
+                ))
+            }
+            Err(e) => match e.kind() {
+                // Haven't received enough bytes to decode yet
+                // TODO: check if this is the only Error variant where we return `Ok(None)`
+                ErrorKind::UnexpectedEof => {
+                    return Ok(None);
+                }
+                _ => return Err(e).map_err(RPCError::from),
+            },
+        }
+    }
+}

beacon_node/eth2_libp2p/src/rpc/methods.rs

@@ -0,0 +1,412 @@
+//! Available RPC methods types and ids.
+
+use crate::types::EnrBitfield;
+use serde::Serialize;
+use ssz_derive::{Decode, Encode};
+use ssz_types::{
+    typenum::{U1024, U256},
+    VariableList,
+};
+use std::ops::Deref;
+use types::{Epoch, EthSpec, Hash256, SignedBeaconBlock, Slot};
+
+/// Maximum number of blocks in a single request.
+pub type MaxRequestBlocks = U1024;
+pub const MAX_REQUEST_BLOCKS: u64 = 1024;
+
+/// Maximum length of error message.
+type MaxErrorLen = U256;
+
+/// Wrapper over SSZ List to represent error message in rpc responses.
+#[derive(Debug, Clone)]
+pub struct ErrorType(VariableList<u8, MaxErrorLen>);
+
+impl From<String> for ErrorType {
+    fn from(s: String) -> Self {
+        Self(VariableList::from(s.as_bytes().to_vec()))
+    }
+}
+
+impl From<&str> for ErrorType {
+    fn from(s: &str) -> Self {
+        Self(VariableList::from(s.as_bytes().to_vec()))
+    }
+}
+
+impl Deref for ErrorType {
+    type Target = VariableList<u8, MaxErrorLen>;
+    fn deref(&self) -> &Self::Target {
+        &self.0
+    }
+}
+
+impl ToString for ErrorType {
+    fn to_string(&self) -> String {
+        match std::str::from_utf8(self.0.deref()) {
+            Ok(s) => s.to_string(),
+            Err(_) => format!("{:?}", self.0.deref()), // Display raw bytes if not a UTF-8 string
+        }
+    }
+}
+
+/* Request/Response data structures for RPC methods */
+
+/* Requests */
+
+/// Identifier of a request.
+///
+// NOTE: The handler stores the `RequestId` to inform back of responses and errors, but it's execution
+// is independent of the contents on this type.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum RequestId {
+    Router,
+    Sync(usize),
+    Behaviour,
+}
+
+/// The STATUS request/response handshake message.
+#[derive(Encode, Decode, Clone, Debug, PartialEq)]
+pub struct StatusMessage {
+    /// The fork version of the chain we are broadcasting.
+    pub fork_digest: [u8; 4],
+
+    /// Latest finalized root.
+    pub finalized_root: Hash256,
+
+    /// Latest finalized epoch.
+    pub finalized_epoch: Epoch,
+
+    /// The latest block root.
+    pub head_root: Hash256,
+
+    /// The slot associated with the latest block root.
+    pub head_slot: Slot,
+}
+
+/// The PING request/response message.
+#[derive(Encode, Decode, Clone, Debug, PartialEq)]
+pub struct Ping {
+    /// The metadata sequence number.
+    pub data: u64,
+}
+
+/// The METADATA response structure.
+#[derive(Encode, Decode, Clone, Debug, PartialEq, Serialize)]
+#[serde(bound = "T: EthSpec")]
+pub struct MetaData<T: EthSpec> {
+    /// A sequential counter indicating when data gets modified.
+    pub seq_number: u64,
+    /// The persistent subnet bitfield.
+    pub attnets: EnrBitfield<T>,
+}
+
+/// The reason given for a `Goodbye` message.
+///
+/// Note: any unknown `u64::into(n)` will resolve to `Goodbye::Unknown` for any unknown `n`,
+/// however `GoodbyeReason::Unknown.into()` will go into `0_u64`. Therefore de-serializing then
+/// re-serializing may not return the same bytes.
+#[derive(Debug, Clone, PartialEq)]
+pub enum GoodbyeReason {
+    /// This node has shutdown.
+    ClientShutdown = 1,
+
+    /// Incompatible networks.
+    IrrelevantNetwork = 2,
+
+    /// Error/fault in the RPC.
+    Fault = 3,
+
+    /// Unknown reason.
+    Unknown = 0,
+}
+
+impl From<u64> for GoodbyeReason {
+    fn from(id: u64) -> GoodbyeReason {
+        match id {
+            1 => GoodbyeReason::ClientShutdown,
+            2 => GoodbyeReason::IrrelevantNetwork,
+            3 => GoodbyeReason::Fault,
+            _ => GoodbyeReason::Unknown,
+        }
+    }
+}
+
+impl Into<u64> for GoodbyeReason {
+    fn into(self) -> u64 {
+        self as u64
+    }
+}
+
+impl ssz::Encode for GoodbyeReason {
+    fn is_ssz_fixed_len() -> bool {
+        <u64 as ssz::Encode>::is_ssz_fixed_len()
+    }
+
+    fn ssz_fixed_len() -> usize {
+        <u64 as ssz::Encode>::ssz_fixed_len()
+    }
+
+    fn ssz_bytes_len(&self) -> usize {
+        0_u64.ssz_bytes_len()
+    }
+
+    fn ssz_append(&self, buf: &mut Vec<u8>) {
+        let conv: u64 = self.clone().into();
+        conv.ssz_append(buf)
+    }
+}
+
+impl ssz::Decode for GoodbyeReason {
+    fn is_ssz_fixed_len() -> bool {
+        <u64 as ssz::Decode>::is_ssz_fixed_len()
+    }
+
+    fn ssz_fixed_len() -> usize {
+        <u64 as ssz::Decode>::ssz_fixed_len()
+    }
+
+    fn from_ssz_bytes(bytes: &[u8]) -> Result<Self, ssz::DecodeError> {
+        u64::from_ssz_bytes(bytes).and_then(|n| Ok(n.into()))
+    }
+}
+
+/// Request a number of beacon block roots from a peer.
+#[derive(Encode, Decode, Clone, Debug, PartialEq)]
+pub struct BlocksByRangeRequest {
+    /// The starting slot to request blocks.
+    pub start_slot: u64,
+
+    /// The number of blocks from the start slot.
+    pub count: u64,
+
+    /// The step increment to receive blocks.
+    ///
+    /// A value of 1 returns every block.
+    /// A value of 2 returns every second block.
+    /// A value of 3 returns every third block and so on.
+    pub step: u64,
+}
+
+/// Request a number of beacon block bodies from a peer.
+#[derive(Encode, Decode, Clone, Debug, PartialEq)]
+pub struct BlocksByRootRequest {
+    /// The list of beacon block bodies being requested.
+    pub block_roots: VariableList<Hash256, MaxRequestBlocks>,
+}
+
+/* RPC Handling and Grouping */
+// Collection of enums and structs used by the Codecs to encode/decode RPC messages
+
+#[derive(Debug, Clone, PartialEq)]
+pub enum RPCResponse<T: EthSpec> {
+    /// A HELLO message.
+    Status(StatusMessage),
+
+    /// A response to a get BLOCKS_BY_RANGE request. A None response signifies the end of the
+    /// batch.
+    BlocksByRange(Box<SignedBeaconBlock<T>>),
+
+    /// A response to a get BLOCKS_BY_ROOT request.
+    BlocksByRoot(Box<SignedBeaconBlock<T>>),
+
+    /// A PONG response to a PING request.
+    Pong(Ping),
+
+    /// A response to a META_DATA request.
+    MetaData(MetaData<T>),
+}
+
+/// Indicates which response is being terminated by a stream termination response.
+#[derive(Debug, Clone)]
+pub enum ResponseTermination {
+    /// Blocks by range stream termination.
+    BlocksByRange,
+
+    /// Blocks by root stream termination.
+    BlocksByRoot,
+}
+
+/// The structured response containing a result/code indicating success or failure
+/// and the contents of the response
+#[derive(Debug, Clone)]
+pub enum RPCCodedResponse<T: EthSpec> {
+    /// The response is a successful.
+    Success(RPCResponse<T>),
+
+    /// The response was invalid.
+    InvalidRequest(ErrorType),
+
+    /// The response indicates a server error.
+    ServerError(ErrorType),
+
+    /// There was an unknown response.
+    Unknown(ErrorType),
+
+    /// Received a stream termination indicating which response is being terminated.
+    StreamTermination(ResponseTermination),
+}
+
+/// The code assigned to an erroneous `RPCResponse`.
+#[derive(Debug, Clone, Copy)]
+pub enum RPCResponseErrorCode {
+    InvalidRequest,
+    ServerError,
+    Unknown,
+}
+
+impl<T: EthSpec> RPCCodedResponse<T> {
+    /// Used to encode the response in the codec.
+    pub fn as_u8(&self) -> Option<u8> {
+        match self {
+            RPCCodedResponse::Success(_) => Some(0),
+            RPCCodedResponse::InvalidRequest(_) => Some(1),
+            RPCCodedResponse::ServerError(_) => Some(2),
+            RPCCodedResponse::Unknown(_) => Some(255),
+            RPCCodedResponse::StreamTermination(_) => None,
+        }
+    }
+
+    /// Tells the codec whether to decode as an RPCResponse or an error.
+    pub fn is_response(response_code: u8) -> bool {
+        match response_code {
+            0 => true,
+            _ => false,
+        }
+    }
+
+    /// Builds an RPCCodedResponse from a response code and an ErrorMessage
+    pub fn from_error(response_code: u8, err: String) -> Self {
+        match response_code {
+            1 => RPCCodedResponse::InvalidRequest(err.into()),
+            2 => RPCCodedResponse::ServerError(err.into()),
+            _ => RPCCodedResponse::Unknown(err.into()),
+        }
+    }
+
+    /// Builds an RPCCodedResponse from a response code and an ErrorMessage
+    pub fn from_error_code(response_code: RPCResponseErrorCode, err: String) -> Self {
+        match response_code {
+            RPCResponseErrorCode::InvalidRequest => RPCCodedResponse::InvalidRequest(err.into()),
+            RPCResponseErrorCode::ServerError => RPCCodedResponse::ServerError(err.into()),
+            RPCResponseErrorCode::Unknown => RPCCodedResponse::Unknown(err.into()),
+        }
+    }
+
+    /// Specifies which response allows for multiple chunks for the stream handler.
+    pub fn multiple_responses(&self) -> bool {
+        match self {
+            RPCCodedResponse::Success(resp) => match resp {
+                RPCResponse::Status(_) => false,
+                RPCResponse::BlocksByRange(_) => true,
+                RPCResponse::BlocksByRoot(_) => true,
+                RPCResponse::Pong(_) => false,
+                RPCResponse::MetaData(_) => false,
+            },
+            RPCCodedResponse::InvalidRequest(_) => true,
+            RPCCodedResponse::ServerError(_) => true,
+            RPCCodedResponse::Unknown(_) => true,
+            // Stream terminations are part of responses that have chunks
+            RPCCodedResponse::StreamTermination(_) => true,
+        }
+    }
+
+    /// Returns true if this response is an error. Used to terminate the stream after an error is
+    /// sent.
+    pub fn is_error(&self) -> bool {
+        match self {
+            RPCCodedResponse::Success(_) => false,
+            _ => true,
+        }
+    }
+
+    pub fn error_code(&self) -> Option<RPCResponseErrorCode> {
+        match self {
+            RPCCodedResponse::Success(_) => None,
+            RPCCodedResponse::StreamTermination(_) => None,
+            RPCCodedResponse::InvalidRequest(_) => Some(RPCResponseErrorCode::InvalidRequest),
+            RPCCodedResponse::ServerError(_) => Some(RPCResponseErrorCode::ServerError),
+            RPCCodedResponse::Unknown(_) => Some(RPCResponseErrorCode::Unknown),
+        }
+    }
+}
+
+impl std::fmt::Display for RPCResponseErrorCode {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        let repr = match self {
+            RPCResponseErrorCode::InvalidRequest => "The request was invalid",
+            RPCResponseErrorCode::ServerError => "Server error occurred",
+            RPCResponseErrorCode::Unknown => "Unknown error occurred",
+        };
+        f.write_str(repr)
+    }
+}
+
+impl std::fmt::Display for StatusMessage {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        write!(f, "Status Message: Fork Digest: {:?}, Finalized Root: {}, Finalized Epoch: {}, Head Root: {}, Head Slot: {}", self.fork_digest, self.finalized_root, self.finalized_epoch, self.head_root, self.head_slot)
+    }
+}
+
+impl<T: EthSpec> std::fmt::Display for RPCResponse<T> {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        match self {
+            RPCResponse::Status(status) => write!(f, "{}", status),
+            RPCResponse::BlocksByRange(block) => {
+                write!(f, "BlocksByRange: Block slot: {}", block.message.slot)
+            }
+            RPCResponse::BlocksByRoot(block) => {
+                write!(f, "BlocksByRoot: BLock slot: {}", block.message.slot)
+            }
+            RPCResponse::Pong(ping) => write!(f, "Pong: {}", ping.data),
+            RPCResponse::MetaData(metadata) => write!(f, "Metadata: {}", metadata.seq_number),
+        }
+    }
+}
+
+impl<T: EthSpec> std::fmt::Display for RPCCodedResponse<T> {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        match self {
+            RPCCodedResponse::Success(res) => write!(f, "{}", res),
+            RPCCodedResponse::InvalidRequest(err) => write!(f, "Invalid Request: {:?}", err),
+            RPCCodedResponse::ServerError(err) => write!(f, "Server Error: {:?}", err),
+            RPCCodedResponse::Unknown(err) => write!(f, "Unknown Error: {:?}", err),
+            RPCCodedResponse::StreamTermination(_) => write!(f, "Stream Termination"),
+        }
+    }
+}
+
+impl std::fmt::Display for GoodbyeReason {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        match self {
+            GoodbyeReason::ClientShutdown => write!(f, "Client Shutdown"),
+            GoodbyeReason::IrrelevantNetwork => write!(f, "Irrelevant Network"),
+            GoodbyeReason::Fault => write!(f, "Fault"),
+            GoodbyeReason::Unknown => write!(f, "Unknown Reason"),
+        }
+    }
+}
+
+impl std::fmt::Display for BlocksByRangeRequest {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        write!(
+            f,
+            "Start Slot: {}, Count: {}, Step: {}",
+            self.start_slot, self.count, self.step
+        )
+    }
+}
+
+impl slog::Value for RequestId {
+    fn serialize(
+        &self,
+        record: &slog::Record,
+        key: slog::Key,
+        serializer: &mut dyn slog::Serializer,
+    ) -> slog::Result {
+        match self {
+            RequestId::Behaviour => slog::Value::serialize("Behaviour", record, key, serializer),
+            RequestId::Router => slog::Value::serialize("Router", record, key, serializer),
+            RequestId::Sync(ref id) => slog::Value::serialize(id, record, key, serializer),
+        }
+    }
+}

beacon_node/eth2_libp2p/src/rpc/mod.rs

@@ -0,0 +1,222 @@
+//! The Ethereum 2.0 Wire Protocol
+//!
+//! This protocol is a purpose built Ethereum 2.0 libp2p protocol. It's role is to facilitate
+//! direct peer-to-peer communication primarily for sending/receiving chain information for
+//! syncing.
+
+use handler::RPCHandler;
+use libp2p::core::{connection::ConnectionId, ConnectedPoint};
+use libp2p::swarm::{
+    protocols_handler::ProtocolsHandler, NetworkBehaviour, NetworkBehaviourAction, NotifyHandler,
+    PollParameters, SubstreamProtocol,
+};
+use libp2p::{Multiaddr, PeerId};
+use slog::{debug, o};
+use std::marker::PhantomData;
+use std::task::{Context, Poll};
+use std::time::Duration;
+use types::EthSpec;
+
+pub(crate) use handler::HandlerErr;
+pub(crate) use methods::{MetaData, Ping, RPCCodedResponse, RPCResponse};
+pub(crate) use protocol::{RPCProtocol, RPCRequest};
+
+pub use handler::SubstreamId;
+pub use methods::{
+    BlocksByRangeRequest, BlocksByRootRequest, GoodbyeReason, MaxRequestBlocks,
+    RPCResponseErrorCode, RequestId, ResponseTermination, StatusMessage, MAX_REQUEST_BLOCKS,
+};
+pub use protocol::{Protocol, RPCError};
+
+pub(crate) mod codec;
+mod handler;
+pub mod methods;
+mod protocol;
+
+/// RPC events sent from Lighthouse.
+#[derive(Debug, Clone)]
+pub enum RPCSend<T: EthSpec> {
+    /// A request sent from Lighthouse.
+    ///
+    /// The `RequestId` is given by the application making the request. These
+    /// go over *outbound* connections.
+    Request(RequestId, RPCRequest<T>),
+    /// A response sent from Lighthouse.
+    ///
+    /// The `SubstreamId` must correspond to the RPC-given ID of the original request received from the
+    /// peer. The second parameter is a single chunk of a response. These go over *inbound*
+    /// connections.
+    Response(SubstreamId, RPCCodedResponse<T>),
+}
+
+/// RPC events received from outside Lighthouse.
+#[derive(Debug, Clone)]
+pub enum RPCReceived<T: EthSpec> {
+    /// A request received from the outside.
+    ///
+    /// The `SubstreamId` is given by the `RPCHandler` as it identifies this request with the
+    /// *inbound* substream over which it is managed.
+    Request(SubstreamId, RPCRequest<T>),
+    /// A response received from the outside.
+    ///
+    /// The `RequestId` corresponds to the application given ID of the original request sent to the
+    /// peer. The second parameter is a single chunk of a response. These go over *outbound*
+    /// connections.
+    Response(RequestId, RPCResponse<T>),
+    /// Marks a request as completed
+    EndOfStream(RequestId, ResponseTermination),
+}
+
+impl<T: EthSpec> std::fmt::Display for RPCSend<T> {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        match self {
+            RPCSend::Request(id, req) => write!(f, "RPC Request(id: {:?}, {})", id, req),
+            RPCSend::Response(id, res) => write!(f, "RPC Response(id: {:?}, {})", id, res),
+        }
+    }
+}
+
+/// Messages sent to the user from the RPC protocol.
+pub struct RPCMessage<TSpec: EthSpec> {
+    /// The peer that sent the message.
+    pub peer_id: PeerId,
+    /// Handler managing this message.
+    pub conn_id: ConnectionId,
+    /// The message that was sent.
+    pub event: <RPCHandler<TSpec> as ProtocolsHandler>::OutEvent,
+}
+
+/// Implements the libp2p `NetworkBehaviour` trait and therefore manages network-level
+/// logic.
+pub struct RPC<TSpec: EthSpec> {
+    /// Queue of events to be processed.
+    events: Vec<NetworkBehaviourAction<RPCSend<TSpec>, RPCMessage<TSpec>>>,
+    /// Slog logger for RPC behaviour.
+    log: slog::Logger,
+}
+
+impl<TSpec: EthSpec> RPC<TSpec> {
+    pub fn new(log: slog::Logger) -> Self {
+        let log = log.new(o!("service" => "libp2p_rpc"));
+        RPC {
+            events: Vec::new(),
+            log,
+        }
+    }
+
+    /// Sends an RPC response.
+    ///
+    /// The peer must be connected for this to succeed.
+    pub fn send_response(
+        &mut self,
+        peer_id: PeerId,
+        id: (ConnectionId, SubstreamId),
+        event: RPCCodedResponse<TSpec>,
+    ) {
+        self.events.push(NetworkBehaviourAction::NotifyHandler {
+            peer_id,
+            handler: NotifyHandler::One(id.0),
+            event: RPCSend::Response(id.1, event),
+        });
+    }
+
+    /// Submits an RPC request.
+    ///
+    /// The peer must be connected for this to succeed.
+    pub fn send_request(
+        &mut self,
+        peer_id: PeerId,
+        request_id: RequestId,
+        event: RPCRequest<TSpec>,
+    ) {
+        self.events.push(NetworkBehaviourAction::NotifyHandler {
+            peer_id,
+            handler: NotifyHandler::Any,
+            event: RPCSend::Request(request_id, event),
+        });
+    }
+}
+
+impl<TSpec> NetworkBehaviour for RPC<TSpec>
+where
+    TSpec: EthSpec,
+{
+    type ProtocolsHandler = RPCHandler<TSpec>;
+    type OutEvent = RPCMessage<TSpec>;
+
+    fn new_handler(&mut self) -> Self::ProtocolsHandler {
+        RPCHandler::new(
+            SubstreamProtocol::new(RPCProtocol {
+                phantom: PhantomData,
+            }),
+            Duration::from_secs(30),
+            &self.log,
+        )
+    }
+
+    // handled by discovery
+    fn addresses_of_peer(&mut self, _peer_id: &PeerId) -> Vec<Multiaddr> {
+        Vec::new()
+    }
+
+    // Use connection established/closed instead of these currently
+    fn inject_connected(&mut self, peer_id: &PeerId) {
+        // find the peer's meta-data
+        debug!(self.log, "Requesting new peer's metadata"; "peer_id" => format!("{}",peer_id));
+        let rpc_event = RPCSend::Request(RequestId::Behaviour, RPCRequest::MetaData(PhantomData));
+        self.events.push(NetworkBehaviourAction::NotifyHandler {
+            peer_id: peer_id.clone(),
+            handler: NotifyHandler::Any,
+            event: rpc_event,
+        });
+    }
+
+    fn inject_disconnected(&mut self, _peer_id: &PeerId) {}
+
+    fn inject_connection_established(
+        &mut self,
+        _peer_id: &PeerId,
+        _: &ConnectionId,
+        _connected_point: &ConnectedPoint,
+    ) {
+    }
+
+    fn inject_connection_closed(
+        &mut self,
+        _peer_id: &PeerId,
+        _: &ConnectionId,
+        _connected_point: &ConnectedPoint,
+    ) {
+    }
+
+    fn inject_event(
+        &mut self,
+        peer_id: PeerId,
+        conn_id: ConnectionId,
+        event: <Self::ProtocolsHandler as ProtocolsHandler>::OutEvent,
+    ) {
+        // send the event to the user
+        self.events
+            .push(NetworkBehaviourAction::GenerateEvent(RPCMessage {
+                peer_id,
+                conn_id,
+                event,
+            }));
+    }
+
+    fn poll(
+        &mut self,
+        _cx: &mut Context,
+        _: &mut impl PollParameters,
+    ) -> Poll<
+        NetworkBehaviourAction<
+            <Self::ProtocolsHandler as ProtocolsHandler>::InEvent,
+            Self::OutEvent,
+        >,
+    > {
+        if !self.events.is_empty() {
+            return Poll::Ready(self.events.remove(0));
+        }
+        Poll::Pending
+    }
+}

beacon_node/eth2_libp2p/src/rpc/protocol.rs

@@ -0,0 +1,500 @@
+#![allow(clippy::type_complexity)]
+
+use super::methods::*;
+use crate::rpc::{
+    codec::{
+        base::{BaseInboundCodec, BaseOutboundCodec},
+        ssz::{SSZInboundCodec, SSZOutboundCodec},
+        ssz_snappy::{SSZSnappyInboundCodec, SSZSnappyOutboundCodec},
+        InboundCodec, OutboundCodec,
+    },
+    methods::ResponseTermination,
+    MaxRequestBlocks, MAX_REQUEST_BLOCKS,
+};
+use futures::future::Ready;
+use futures::prelude::*;
+use futures::prelude::{AsyncRead, AsyncWrite};
+use libp2p::core::{InboundUpgrade, OutboundUpgrade, ProtocolName, UpgradeInfo};
+use ssz::Encode;
+use ssz_types::VariableList;
+use std::io;
+use std::marker::PhantomData;
+use std::pin::Pin;
+use std::time::Duration;
+use tokio_io_timeout::TimeoutStream;
+use tokio_util::{
+    codec::Framed,
+    compat::{Compat, FuturesAsyncReadCompatExt},
+};
+use types::{BeaconBlock, EthSpec, Hash256, MainnetEthSpec, Signature, SignedBeaconBlock};
+
+lazy_static! {
+    // Note: Hardcoding the `EthSpec` type for `SignedBeaconBlock` as min/max values is
+    // same across different `EthSpec` implementations.
+    pub static ref SIGNED_BEACON_BLOCK_MIN: usize = SignedBeaconBlock::<MainnetEthSpec> {
+        message: BeaconBlock::empty(&MainnetEthSpec::default_spec()),
+        signature: Signature::empty_signature(),
+    }
+    .as_ssz_bytes()
+    .len();
+    pub static ref SIGNED_BEACON_BLOCK_MAX: usize = SignedBeaconBlock::<MainnetEthSpec> {
+        message: BeaconBlock::full(&MainnetEthSpec::default_spec()),
+        signature: Signature::empty_signature(),
+    }
+    .as_ssz_bytes()
+    .len();
+    pub static ref BLOCKS_BY_ROOT_REQUEST_MIN: usize = BlocksByRootRequest {
+        block_roots: VariableList::<Hash256, MaxRequestBlocks>::from(Vec::<Hash256>::new())
+    }
+    .as_ssz_bytes()
+    .len();
+    pub static ref BLOCKS_BY_ROOT_REQUEST_MAX: usize = BlocksByRootRequest {
+        block_roots: VariableList::<Hash256, MaxRequestBlocks>::from(vec![
+            Hash256::zero();
+            MAX_REQUEST_BLOCKS
+                as usize
+        ])
+    }
+    .as_ssz_bytes()
+    .len();
+}
+
+/// The maximum bytes that can be sent across the RPC.
+const MAX_RPC_SIZE: usize = 1_048_576; // 1M
+/// The protocol prefix the RPC protocol id.
+const PROTOCOL_PREFIX: &str = "/eth2/beacon_chain/req";
+/// Time allowed for the first byte of a request to arrive before we time out (Time To First Byte).
+const TTFB_TIMEOUT: u64 = 5;
+/// The number of seconds to wait for the first bytes of a request once a protocol has been
+/// established before the stream is terminated.
+const REQUEST_TIMEOUT: u64 = 15;
+
+/// Protocol names to be used.
+#[derive(Debug, Clone, Copy)]
+pub enum Protocol {
+    /// The Status protocol name.
+    Status,
+    /// The Goodbye protocol name.
+    Goodbye,
+    /// The `BlocksByRange` protocol name.
+    BlocksByRange,
+    /// The `BlocksByRoot` protocol name.
+    BlocksByRoot,
+    /// The `Ping` protocol name.
+    Ping,
+    /// The `MetaData` protocol name.
+    MetaData,
+}
+
+/// RPC Versions
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub enum Version {
+    /// Version 1 of RPC
+    V1,
+}
+
+/// RPC Encondings supported.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub enum Encoding {
+    SSZ,
+    SSZSnappy,
+}
+
+impl std::fmt::Display for Protocol {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        let repr = match self {
+            Protocol::Status => "status",
+            Protocol::Goodbye => "goodbye",
+            Protocol::BlocksByRange => "beacon_blocks_by_range",
+            Protocol::BlocksByRoot => "beacon_blocks_by_root",
+            Protocol::Ping => "ping",
+            Protocol::MetaData => "metadata",
+        };
+        f.write_str(repr)
+    }
+}
+
+impl std::fmt::Display for Encoding {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        let repr = match self {
+            Encoding::SSZ => "ssz",
+            Encoding::SSZSnappy => "ssz_snappy",
+        };
+        f.write_str(repr)
+    }
+}
+
+impl std::fmt::Display for Version {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        let repr = match self {
+            Version::V1 => "1",
+        };
+        f.write_str(repr)
+    }
+}
+
+#[derive(Debug, Clone)]
+pub struct RPCProtocol<TSpec: EthSpec> {
+    pub phantom: PhantomData<TSpec>,
+}
+
+impl<TSpec: EthSpec> UpgradeInfo for RPCProtocol<TSpec> {
+    type Info = ProtocolId;
+    type InfoIter = Vec<Self::Info>;
+
+    /// The list of supported RPC protocols for Lighthouse.
+    fn protocol_info(&self) -> Self::InfoIter {
+        vec![
+            ProtocolId::new(Protocol::Status, Version::V1, Encoding::SSZSnappy),
+            ProtocolId::new(Protocol::Status, Version::V1, Encoding::SSZ),
+            ProtocolId::new(Protocol::Goodbye, Version::V1, Encoding::SSZSnappy),
+            ProtocolId::new(Protocol::Goodbye, Version::V1, Encoding::SSZ),
+            ProtocolId::new(Protocol::BlocksByRange, Version::V1, Encoding::SSZSnappy),
+            ProtocolId::new(Protocol::BlocksByRange, Version::V1, Encoding::SSZ),
+            ProtocolId::new(Protocol::BlocksByRoot, Version::V1, Encoding::SSZSnappy),
+            ProtocolId::new(Protocol::BlocksByRoot, Version::V1, Encoding::SSZ),
+            ProtocolId::new(Protocol::Ping, Version::V1, Encoding::SSZSnappy),
+            ProtocolId::new(Protocol::Ping, Version::V1, Encoding::SSZ),
+            ProtocolId::new(Protocol::MetaData, Version::V1, Encoding::SSZSnappy),
+            ProtocolId::new(Protocol::MetaData, Version::V1, Encoding::SSZ),
+        ]
+    }
+}
+
+/// Tracks the types in a protocol id.
+#[derive(Clone, Debug)]
+pub struct ProtocolId {
+    /// The RPC message type/name.
+    pub message_name: Protocol,
+
+    /// The version of the RPC.
+    pub version: Version,
+
+    /// The encoding of the RPC.
+    pub encoding: Encoding,
+
+    /// The protocol id that is formed from the above fields.
+    protocol_id: String,
+}
+
+/// An RPC protocol ID.
+impl ProtocolId {
+    pub fn new(message_name: Protocol, version: Version, encoding: Encoding) -> Self {
+        let protocol_id = format!(
+            "{}/{}/{}/{}",
+            PROTOCOL_PREFIX, message_name, version, encoding
+        );
+
+        ProtocolId {
+            message_name,
+            version,
+            encoding,
+            protocol_id,
+        }
+    }
+}
+
+impl ProtocolName for ProtocolId {
+    fn protocol_name(&self) -> &[u8] {
+        self.protocol_id.as_bytes()
+    }
+}
+
+/* Inbound upgrade */
+
+// The inbound protocol reads the request, decodes it and returns the stream to the protocol
+// handler to respond to once ready.
+
+pub type InboundOutput<TSocket, TSpec> = (RPCRequest<TSpec>, InboundFramed<TSocket, TSpec>);
+pub type InboundFramed<TSocket, TSpec> =
+    Framed<TimeoutStream<Compat<TSocket>>, InboundCodec<TSpec>>;
+type FnAndThen<TSocket, TSpec> = fn(
+    (
+        Option<Result<RPCRequest<TSpec>, RPCError>>,
+        InboundFramed<TSocket, TSpec>,
+    ),
+) -> Ready<Result<InboundOutput<TSocket, TSpec>, RPCError>>;
+type FnMapErr = fn(tokio::time::Elapsed) -> RPCError;
+
+impl<TSocket, TSpec> InboundUpgrade<TSocket> for RPCProtocol<TSpec>
+where
+    TSocket: AsyncRead + AsyncWrite + Unpin + Send + 'static,
+    TSpec: EthSpec,
+{
+    type Output = InboundOutput<TSocket, TSpec>;
+    type Error = RPCError;
+    type Future = Pin<Box<dyn Future<Output = Result<Self::Output, Self::Error>> + Send>>;
+
+    fn upgrade_inbound(self, socket: TSocket, protocol: ProtocolId) -> Self::Future {
+        let protocol_name = protocol.message_name;
+        // convert the socket to tokio compatible socket
+        let socket = socket.compat();
+        let codec = match protocol.encoding {
+            Encoding::SSZSnappy => {
+                let ssz_snappy_codec =
+                    BaseInboundCodec::new(SSZSnappyInboundCodec::new(protocol, MAX_RPC_SIZE));
+                InboundCodec::SSZSnappy(ssz_snappy_codec)
+            }
+            Encoding::SSZ => {
+                let ssz_codec = BaseInboundCodec::new(SSZInboundCodec::new(protocol, MAX_RPC_SIZE));
+                InboundCodec::SSZ(ssz_codec)
+            }
+        };
+        let mut timed_socket = TimeoutStream::new(socket);
+        timed_socket.set_read_timeout(Some(Duration::from_secs(TTFB_TIMEOUT)));
+
+        let socket = Framed::new(timed_socket, codec);
+
+        // MetaData requests should be empty, return the stream
+        Box::pin(match protocol_name {
+            Protocol::MetaData => {
+                future::Either::Left(future::ok((RPCRequest::MetaData(PhantomData), socket)))
+            }
+
+            _ => future::Either::Right(
+                tokio::time::timeout(Duration::from_secs(REQUEST_TIMEOUT), socket.into_future())
+                    .map_err(RPCError::from as FnMapErr)
+                    .and_then({
+                        |(req, stream)| match req {
+                            Some(Ok(request)) => future::ok((request, stream)),
+                            Some(Err(_)) | None => future::err(RPCError::IncompleteStream),
+                        }
+                    } as FnAndThen<TSocket, TSpec>),
+            ),
+        })
+    }
+}
+
+/* Outbound request */
+
+// Combines all the RPC requests into a single enum to implement `UpgradeInfo` and
+// `OutboundUpgrade`
+
+#[derive(Debug, Clone, PartialEq)]
+pub enum RPCRequest<TSpec: EthSpec> {
+    Status(StatusMessage),
+    Goodbye(GoodbyeReason),
+    BlocksByRange(BlocksByRangeRequest),
+    BlocksByRoot(BlocksByRootRequest),
+    Ping(Ping),
+    MetaData(PhantomData<TSpec>),
+}
+
+impl<TSpec: EthSpec> UpgradeInfo for RPCRequest<TSpec> {
+    type Info = ProtocolId;
+    type InfoIter = Vec<Self::Info>;
+
+    // add further protocols as we support more encodings/versions
+    fn protocol_info(&self) -> Self::InfoIter {
+        self.supported_protocols()
+    }
+}
+
+/// Implements the encoding per supported protocol for `RPCRequest`.
+impl<TSpec: EthSpec> RPCRequest<TSpec> {
+    pub fn supported_protocols(&self) -> Vec<ProtocolId> {
+        match self {
+            // add more protocols when versions/encodings are supported
+            RPCRequest::Status(_) => vec![
+                ProtocolId::new(Protocol::Status, Version::V1, Encoding::SSZSnappy),
+                ProtocolId::new(Protocol::Status, Version::V1, Encoding::SSZ),
+            ],
+            RPCRequest::Goodbye(_) => vec![
+                ProtocolId::new(Protocol::Goodbye, Version::V1, Encoding::SSZSnappy),
+                ProtocolId::new(Protocol::Goodbye, Version::V1, Encoding::SSZ),
+            ],
+            RPCRequest::BlocksByRange(_) => vec![
+                ProtocolId::new(Protocol::BlocksByRange, Version::V1, Encoding::SSZSnappy),
+                ProtocolId::new(Protocol::BlocksByRange, Version::V1, Encoding::SSZ),
+            ],
+            RPCRequest::BlocksByRoot(_) => vec![
+                ProtocolId::new(Protocol::BlocksByRoot, Version::V1, Encoding::SSZSnappy),
+                ProtocolId::new(Protocol::BlocksByRoot, Version::V1, Encoding::SSZ),
+            ],
+            RPCRequest::Ping(_) => vec![
+                ProtocolId::new(Protocol::Ping, Version::V1, Encoding::SSZSnappy),
+                ProtocolId::new(Protocol::Ping, Version::V1, Encoding::SSZ),
+            ],
+            RPCRequest::MetaData(_) => vec![
+                ProtocolId::new(Protocol::MetaData, Version::V1, Encoding::SSZSnappy),
+                ProtocolId::new(Protocol::MetaData, Version::V1, Encoding::SSZ),
+            ],
+        }
+    }
+
+    /* These functions are used in the handler for stream management */
+
+    /// Number of responses expected for this request.
+    pub fn expected_responses(&self) -> usize {
+        match self {
+            RPCRequest::Status(_) => 1,
+            RPCRequest::Goodbye(_) => 0,
+            RPCRequest::BlocksByRange(req) => req.count as usize,
+            RPCRequest::BlocksByRoot(req) => req.block_roots.len(),
+            RPCRequest::Ping(_) => 1,
+            RPCRequest::MetaData(_) => 1,
+        }
+    }
+
+    /// Gives the corresponding `Protocol` to this request.
+    pub fn protocol(&self) -> Protocol {
+        match self {
+            RPCRequest::Status(_) => Protocol::Status,
+            RPCRequest::Goodbye(_) => Protocol::Goodbye,
+            RPCRequest::BlocksByRange(_) => Protocol::BlocksByRange,
+            RPCRequest::BlocksByRoot(_) => Protocol::BlocksByRoot,
+            RPCRequest::Ping(_) => Protocol::Ping,
+            RPCRequest::MetaData(_) => Protocol::MetaData,
+        }
+    }
+
+    /// Returns the `ResponseTermination` type associated with the request if a stream gets
+    /// terminated.
+    pub fn stream_termination(&self) -> ResponseTermination {
+        match self {
+            // this only gets called after `multiple_responses()` returns true. Therefore, only
+            // variants that have `multiple_responses()` can have values.
+            RPCRequest::BlocksByRange(_) => ResponseTermination::BlocksByRange,
+            RPCRequest::BlocksByRoot(_) => ResponseTermination::BlocksByRoot,
+            RPCRequest::Status(_) => unreachable!(),
+            RPCRequest::Goodbye(_) => unreachable!(),
+            RPCRequest::Ping(_) => unreachable!(),
+            RPCRequest::MetaData(_) => unreachable!(),
+        }
+    }
+}
+
+/* RPC Response type - used for outbound upgrades */
+
+/* Outbound upgrades */
+
+pub type OutboundFramed<TSocket, TSpec> = Framed<Compat<TSocket>, OutboundCodec<TSpec>>;
+
+impl<TSocket, TSpec> OutboundUpgrade<TSocket> for RPCRequest<TSpec>
+where
+    TSpec: EthSpec + Send + 'static,
+    TSocket: AsyncRead + AsyncWrite + Unpin + Send + 'static,
+{
+    type Output = OutboundFramed<TSocket, TSpec>;
+    type Error = RPCError;
+    type Future = Pin<Box<dyn Future<Output = Result<Self::Output, Self::Error>> + Send>>;
+
+    fn upgrade_outbound(self, socket: TSocket, protocol: Self::Info) -> Self::Future {
+        // convert to a tokio compatible socket
+        let socket = socket.compat();
+        let codec = match protocol.encoding {
+            Encoding::SSZSnappy => {
+                let ssz_snappy_codec =
+                    BaseOutboundCodec::new(SSZSnappyOutboundCodec::new(protocol, MAX_RPC_SIZE));
+                OutboundCodec::SSZSnappy(ssz_snappy_codec)
+            }
+            Encoding::SSZ => {
+                let ssz_codec =
+                    BaseOutboundCodec::new(SSZOutboundCodec::new(protocol, MAX_RPC_SIZE));
+                OutboundCodec::SSZ(ssz_codec)
+            }
+        };
+
+        let mut socket = Framed::new(socket, codec);
+
+        let future = async { socket.send(self).await.map(|_| socket) };
+        Box::pin(future)
+    }
+}
+
+/// Error in RPC Encoding/Decoding.
+#[derive(Debug, Clone)]
+pub enum RPCError {
+    /// Error when decoding the raw buffer from ssz.
+    // NOTE: in the future a ssz::DecodeError should map to an InvalidData error
+    SSZDecodeError(ssz::DecodeError),
+    /// IO Error.
+    IoError(String),
+    /// The peer returned a valid response but the response indicated an error.
+    ErrorResponse(RPCResponseErrorCode, String),
+    /// Timed out waiting for a response.
+    StreamTimeout,
+    /// Peer does not support the protocol.
+    UnsupportedProtocol,
+    /// Stream ended unexpectedly.
+    IncompleteStream,
+    /// Peer sent invalid data.
+    InvalidData,
+    /// An error occurred due to internal reasons. Ex: timer failure.
+    InternalError(&'static str),
+    /// Negotiation with this peer timed out.
+    NegotiationTimeout,
+    /// Handler rejected this request.
+    HandlerRejected,
+}
+
+impl From<ssz::DecodeError> for RPCError {
+    #[inline]
+    fn from(err: ssz::DecodeError) -> Self {
+        RPCError::SSZDecodeError(err)
+    }
+}
+impl From<tokio::time::Elapsed> for RPCError {
+    fn from(_: tokio::time::Elapsed) -> Self {
+        RPCError::StreamTimeout
+    }
+}
+
+impl From<io::Error> for RPCError {
+    fn from(err: io::Error) -> Self {
+        RPCError::IoError(err.to_string())
+    }
+}
+
+// Error trait is required for `ProtocolsHandler`
+impl std::fmt::Display for RPCError {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        match *self {
+            RPCError::SSZDecodeError(ref err) => write!(f, "Error while decoding ssz: {:?}", err),
+            RPCError::InvalidData => write!(f, "Peer sent unexpected data"),
+            RPCError::IoError(ref err) => write!(f, "IO Error: {}", err),
+            RPCError::ErrorResponse(ref code, ref reason) => write!(
+                f,
+                "RPC response was an error: {} with reason: {}",
+                code, reason
+            ),
+            RPCError::StreamTimeout => write!(f, "Stream Timeout"),
+            RPCError::UnsupportedProtocol => write!(f, "Peer does not support the protocol"),
+            RPCError::IncompleteStream => write!(f, "Stream ended unexpectedly"),
+            RPCError::InternalError(ref err) => write!(f, "Internal error: {}", err),
+            RPCError::NegotiationTimeout => write!(f, "Negotiation timeout"),
+            RPCError::HandlerRejected => write!(f, "Handler rejected the request"),
+        }
+    }
+}
+
+impl std::error::Error for RPCError {
+    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
+        match *self {
+            // NOTE: this does have a source
+            RPCError::SSZDecodeError(_) => None,
+            RPCError::IoError(_) => None,
+            RPCError::StreamTimeout => None,
+            RPCError::UnsupportedProtocol => None,
+            RPCError::IncompleteStream => None,
+            RPCError::InvalidData => None,
+            RPCError::InternalError(_) => None,
+            RPCError::ErrorResponse(_, _) => None,
+            RPCError::NegotiationTimeout => None,
+            RPCError::HandlerRejected => None,
+        }
+    }
+}
+
+impl<TSpec: EthSpec> std::fmt::Display for RPCRequest<TSpec> {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        match self {
+            RPCRequest::Status(status) => write!(f, "Status Message: {}", status),
+            RPCRequest::Goodbye(reason) => write!(f, "Goodbye: {}", reason),
+            RPCRequest::BlocksByRange(req) => write!(f, "Blocks by range: {}", req),
+            RPCRequest::BlocksByRoot(req) => write!(f, "Blocks by root: {:?}", req),
+            RPCRequest::Ping(ping) => write!(f, "Ping: {}", ping.data),
+            RPCRequest::MetaData(_) => write!(f, "MetaData request"),
+        }
+    }
+}