split outbound and inbound codecs encoded types (#2410)

Splits the inbound and outbound requests, for maintainability.

beacon_node/eth2_libp2p/src/behaviour/handler/mod.rs

@@ -38,7 +38,7 @@ impl<TSpec: EthSpec> BehaviourHandler<TSpec> {
 pub enum BehaviourHandlerIn<TSpec: EthSpec> {
     Delegate(DelegateIn<TSpec>),
     /// Start the shutdown process.
-    Shutdown(Option<(RequestId, RPCRequest<TSpec>)>),
+    Shutdown(Option<(RequestId, OutboundRequest<TSpec>)>),
 }
 
 impl<TSpec: EthSpec> ProtocolsHandler for BehaviourHandler<TSpec> {

beacon_node/eth2_libp2p/src/behaviour/mod.rs

@@ -595,7 +595,7 @@ impl<TSpec: EthSpec> Behaviour<TSpec> {
         trace!(self.log, "Sending Ping"; "request_id" => id, "peer_id" => %peer_id);
 
         self.eth2_rpc
-            .send_request(peer_id, id, RPCRequest::Ping(ping));
+            .send_request(peer_id, id, OutboundRequest::Ping(ping));
     }
 
     /// Sends a Pong response to the peer.
@@ -610,7 +610,7 @@ impl<TSpec: EthSpec> Behaviour<TSpec> {
 
     /// Sends a METADATA request to a peer.
     fn send_meta_data_request(&mut self, peer_id: PeerId) {
-        let event = RPCRequest::MetaData(PhantomData);
+        let event = OutboundRequest::MetaData(PhantomData);
         self.eth2_rpc
             .send_request(peer_id, RequestId::Behaviour, event);
     }
@@ -749,17 +749,17 @@ impl<TSpec: EthSpec> Behaviour<TSpec> {
                 let peer_request_id = (handler_id, id);
                 match request {
                     /* Behaviour managed protocols: Ping and Metadata */
-                    RPCRequest::Ping(ping) => {
+                    InboundRequest::Ping(ping) => {
                         // inform the peer manager and send the response
                         self.peer_manager.ping_request(&peer_id, ping.data);
                         // send a ping response
                         self.pong(peer_request_id, peer_id);
                     }
-                    RPCRequest::MetaData(_) => {
+                    InboundRequest::MetaData(_) => {
                         // send the requested meta-data
                         self.send_meta_data_response((handler_id, id), peer_id);
                     }
-                    RPCRequest::Goodbye(reason) => {
+                    InboundRequest::Goodbye(reason) => {
                         // queue for disconnection without a goodbye message
                         debug!(
                             self.log, "Peer sent Goodbye";
@@ -775,18 +775,18 @@ impl<TSpec: EthSpec> Behaviour<TSpec> {
                         // inform the application layer early.
                     }
                     /* Protocols propagated to the Network */
-                    RPCRequest::Status(msg) => {
+                    InboundRequest::Status(msg) => {
                         // inform the peer manager that we have received a status from a peer
                         self.peer_manager.peer_statusd(&peer_id);
                         // propagate the STATUS message upwards
                         self.propagate_request(peer_request_id, peer_id, Request::Status(msg))
                     }
-                    RPCRequest::BlocksByRange(req) => self.propagate_request(
+                    InboundRequest::BlocksByRange(req) => self.propagate_request(
                         peer_request_id,
                         peer_id,
                         Request::BlocksByRange(req),
                     ),
-                    RPCRequest::BlocksByRoot(req) => {
+                    InboundRequest::BlocksByRoot(req) => {
                         self.propagate_request(peer_request_id, peer_id, Request::BlocksByRoot(req))
                     }
                 }
@@ -834,7 +834,7 @@ impl<TSpec: EthSpec> Behaviour<TSpec> {
                 peer_id,
                 handler: NotifyHandler::Any,
                 event: BehaviourHandlerIn::Shutdown(
-                    reason.map(|reason| (RequestId::Behaviour, RPCRequest::Goodbye(reason))),
+                    reason.map(|reason| (RequestId::Behaviour, OutboundRequest::Goodbye(reason))),
                 ),
             });
         }
@@ -878,7 +878,7 @@ impl<TSpec: EthSpec> Behaviour<TSpec> {
                             handler: NotifyHandler::Any,
                             event: BehaviourHandlerIn::Shutdown(Some((
                                 RequestId::Behaviour,
-                                RPCRequest::Goodbye(reason),
+                                OutboundRequest::Goodbye(reason),
                             ))),
                         });
                     }
@@ -1293,12 +1293,12 @@ pub enum Request {
     BlocksByRoot(BlocksByRootRequest),
 }
 
-impl<TSpec: EthSpec> std::convert::From<Request> for RPCRequest<TSpec> {
-    fn from(req: Request) -> RPCRequest<TSpec> {
+impl<TSpec: EthSpec> std::convert::From<Request> for OutboundRequest<TSpec> {
+    fn from(req: Request) -> OutboundRequest<TSpec> {
         match req {
-            Request::BlocksByRoot(r) => RPCRequest::BlocksByRoot(r),
-            Request::BlocksByRange(r) => RPCRequest::BlocksByRange(r),
-            Request::Status(s) => RPCRequest::Status(s),
+            Request::BlocksByRoot(r) => OutboundRequest::BlocksByRoot(r),
+            Request::BlocksByRange(r) => OutboundRequest::BlocksByRange(r),
+            Request::Status(s) => OutboundRequest::Status(s),
         }
     }
 }

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

@@ -1,7 +1,7 @@
 //! This handles the various supported encoding mechanism for the Eth 2.0 RPC.
 
 use crate::rpc::methods::ErrorType;
-use crate::rpc::{RPCCodedResponse, RPCRequest, RPCResponse};
+use crate::rpc::{InboundRequest, OutboundRequest, RPCCodedResponse, RPCResponse};
 use libp2p::bytes::BufMut;
 use libp2p::bytes::BytesMut;
 use std::marker::PhantomData;
@@ -47,7 +47,7 @@ where
 // This deals with Decoding RPC Responses from other peers and encoding our requests
 pub struct BaseOutboundCodec<TOutboundCodec, TSpec>
 where
-    TOutboundCodec: OutboundCodec<RPCRequest<TSpec>>,
+    TOutboundCodec: OutboundCodec<OutboundRequest<TSpec>>,
     TSpec: EthSpec,
 {
     /// Inner codec for handling various encodings.
@@ -60,7 +60,7 @@ where
 impl<TOutboundCodec, TSpec> BaseOutboundCodec<TOutboundCodec, TSpec>
 where
     TSpec: EthSpec,
-    TOutboundCodec: OutboundCodec<RPCRequest<TSpec>>,
+    TOutboundCodec: OutboundCodec<OutboundRequest<TSpec>>,
 {
     pub fn new(codec: TOutboundCodec) -> Self {
         BaseOutboundCodec {
@@ -102,9 +102,9 @@ where
 impl<TCodec, TSpec> Decoder for BaseInboundCodec<TCodec, TSpec>
 where
     TSpec: EthSpec,
-    TCodec: Encoder<RPCCodedResponse<TSpec>> + Decoder<Item = RPCRequest<TSpec>>,
+    TCodec: Encoder<RPCCodedResponse<TSpec>> + Decoder<Item = InboundRequest<TSpec>>,
 {
-    type Item = RPCRequest<TSpec>;
+    type Item = InboundRequest<TSpec>;
     type Error = <TCodec as Decoder>::Error;
 
     fn decode(&mut self, src: &mut BytesMut) -> Result<Option<Self::Item>, Self::Error> {
@@ -115,14 +115,18 @@ where
 /* Base Outbound Codec */
 
 // This Encodes RPC Requests sent to external peers
-impl<TCodec, TSpec> Encoder<RPCRequest<TSpec>> for BaseOutboundCodec<TCodec, TSpec>
+impl<TCodec, TSpec> Encoder<OutboundRequest<TSpec>> for BaseOutboundCodec<TCodec, TSpec>
 where
     TSpec: EthSpec,
-    TCodec: OutboundCodec<RPCRequest<TSpec>> + Encoder<RPCRequest<TSpec>>,
+    TCodec: OutboundCodec<OutboundRequest<TSpec>> + Encoder<OutboundRequest<TSpec>>,
 {
-    type Error = <TCodec as Encoder<RPCRequest<TSpec>>>::Error;
+    type Error = <TCodec as Encoder<OutboundRequest<TSpec>>>::Error;
 
-    fn encode(&mut self, item: RPCRequest<TSpec>, dst: &mut BytesMut) -> Result<(), Self::Error> {
+    fn encode(
+        &mut self,
+        item: OutboundRequest<TSpec>,
+        dst: &mut BytesMut,
+    ) -> Result<(), Self::Error> {
         self.inner.encode(item, dst)
     }
 }
@@ -131,7 +135,7 @@ where
 impl<TCodec, TSpec> Decoder for BaseOutboundCodec<TCodec, TSpec>
 where
     TSpec: EthSpec,
-    TCodec: OutboundCodec<RPCRequest<TSpec>, CodecErrorType = ErrorType>
+    TCodec: OutboundCodec<OutboundRequest<TSpec>, CodecErrorType = ErrorType>
         + Decoder<Item = RPCResponse<TSpec>>,
 {
     type Item = RPCCodedResponse<TSpec>;

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

@@ -4,7 +4,7 @@ pub(crate) mod ssz_snappy;
 use self::base::{BaseInboundCodec, BaseOutboundCodec};
 use self::ssz_snappy::{SSZSnappyInboundCodec, SSZSnappyOutboundCodec};
 use crate::rpc::protocol::RPCError;
-use crate::rpc::{RPCCodedResponse, RPCRequest};
+use crate::rpc::{InboundRequest, OutboundRequest, RPCCodedResponse};
 use libp2p::bytes::BytesMut;
 use tokio_util::codec::{Decoder, Encoder};
 use types::EthSpec;
@@ -29,7 +29,7 @@ impl<T: EthSpec> Encoder<RPCCodedResponse<T>> for InboundCodec<T> {
 }
 
 impl<TSpec: EthSpec> Decoder for InboundCodec<TSpec> {
-    type Item = RPCRequest<TSpec>;
+    type Item = InboundRequest<TSpec>;
     type Error = RPCError;
 
     fn decode(&mut self, src: &mut BytesMut) -> Result<Option<Self::Item>, Self::Error> {
@@ -39,10 +39,14 @@ impl<TSpec: EthSpec> Decoder for InboundCodec<TSpec> {
     }
 }
 
-impl<TSpec: EthSpec> Encoder<RPCRequest<TSpec>> for OutboundCodec<TSpec> {
+impl<TSpec: EthSpec> Encoder<OutboundRequest<TSpec>> for OutboundCodec<TSpec> {
     type Error = RPCError;
 
-    fn encode(&mut self, item: RPCRequest<TSpec>, dst: &mut BytesMut) -> Result<(), Self::Error> {
+    fn encode(
+        &mut self,
+        item: OutboundRequest<TSpec>,
+        dst: &mut BytesMut,
+    ) -> Result<(), Self::Error> {
         match self {
             OutboundCodec::SSZSnappy(codec) => codec.encode(item, dst),
         }

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

@@ -1,9 +1,8 @@
-use crate::rpc::methods::*;
 use crate::rpc::{
     codec::base::OutboundCodec,
     protocol::{Encoding, Protocol, ProtocolId, RPCError, Version, ERROR_TYPE_MAX, ERROR_TYPE_MIN},
 };
-use crate::rpc::{RPCCodedResponse, RPCRequest, RPCResponse};
+use crate::rpc::{methods::*, InboundRequest, OutboundRequest, RPCCodedResponse, RPCResponse};
 use libp2p::bytes::BytesMut;
 use snap::read::FrameDecoder;
 use snap::write::FrameEncoder;
@@ -90,7 +89,7 @@ impl<TSpec: EthSpec> Encoder<RPCCodedResponse<TSpec>> for SSZSnappyInboundCodec<
 
 // Decoder for inbound streams: Decodes RPC requests from peers
 impl<TSpec: EthSpec> Decoder for SSZSnappyInboundCodec<TSpec> {
-    type Item = RPCRequest<TSpec>;
+    type Item = InboundRequest<TSpec>;
     type Error = RPCError;
 
     fn decode(&mut self, src: &mut BytesMut) -> Result<Option<Self::Item>, Self::Error> {
@@ -133,27 +132,29 @@ impl<TSpec: EthSpec> Decoder for SSZSnappyInboundCodec<TSpec> {
                 // since we have already checked `length` above.
                 match self.protocol.message_name {
                     Protocol::Status => match self.protocol.version {
-                        Version::V1 => Ok(Some(RPCRequest::Status(StatusMessage::from_ssz_bytes(
-                            &decoded_buffer,
-                        )?))),
+                        Version::V1 => Ok(Some(InboundRequest::Status(
+                            StatusMessage::from_ssz_bytes(&decoded_buffer)?,
+                        ))),
                     },
                     Protocol::Goodbye => match self.protocol.version {
-                        Version::V1 => Ok(Some(RPCRequest::Goodbye(
+                        Version::V1 => Ok(Some(InboundRequest::Goodbye(
                             GoodbyeReason::from_ssz_bytes(&decoded_buffer)?,
                         ))),
                     },
                     Protocol::BlocksByRange => match self.protocol.version {
-                        Version::V1 => Ok(Some(RPCRequest::BlocksByRange(
+                        Version::V1 => Ok(Some(InboundRequest::BlocksByRange(
                             BlocksByRangeRequest::from_ssz_bytes(&decoded_buffer)?,
                         ))),
                     },
                     Protocol::BlocksByRoot => match self.protocol.version {
-                        Version::V1 => Ok(Some(RPCRequest::BlocksByRoot(BlocksByRootRequest {
-                            block_roots: VariableList::from_ssz_bytes(&decoded_buffer)?,
-                        }))),
+                        Version::V1 => {
+                            Ok(Some(InboundRequest::BlocksByRoot(BlocksByRootRequest {
+                                block_roots: VariableList::from_ssz_bytes(&decoded_buffer)?,
+                            })))
+                        }
                     },
                     Protocol::Ping => match self.protocol.version {
-                        Version::V1 => Ok(Some(RPCRequest::Ping(Ping {
+                        Version::V1 => Ok(Some(InboundRequest::Ping(Ping {
                             data: u64::from_ssz_bytes(&decoded_buffer)?,
                         }))),
                     },
@@ -163,7 +164,7 @@ impl<TSpec: EthSpec> Decoder for SSZSnappyInboundCodec<TSpec> {
                             if !decoded_buffer.is_empty() {
                                 Err(RPCError::InvalidData)
                             } else {
-                                Ok(Some(RPCRequest::MetaData(PhantomData)))
+                                Ok(Some(InboundRequest::MetaData(PhantomData)))
                             }
                         }
                     },
@@ -201,17 +202,21 @@ impl<TSpec: EthSpec> SSZSnappyOutboundCodec<TSpec> {
 }
 
 // Encoder for outbound streams: Encodes RPC Requests to peers
-impl<TSpec: EthSpec> Encoder<RPCRequest<TSpec>> for SSZSnappyOutboundCodec<TSpec> {
+impl<TSpec: EthSpec> Encoder<OutboundRequest<TSpec>> for SSZSnappyOutboundCodec<TSpec> {
     type Error = RPCError;
 
-    fn encode(&mut self, item: RPCRequest<TSpec>, dst: &mut BytesMut) -> Result<(), Self::Error> {
+    fn encode(
+        &mut self,
+        item: OutboundRequest<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
+            OutboundRequest::Status(req) => req.as_ssz_bytes(),
+            OutboundRequest::Goodbye(req) => req.as_ssz_bytes(),
+            OutboundRequest::BlocksByRange(req) => req.as_ssz_bytes(),
+            OutboundRequest::BlocksByRoot(req) => req.block_roots.as_ssz_bytes(),
+            OutboundRequest::Ping(req) => req.as_ssz_bytes(),
+            OutboundRequest::MetaData(_) => return Ok(()), // no metadata to encode
         };
         // SSZ encoded bytes should be within `max_packet_size`
         if bytes.len() > self.max_packet_size {
@@ -318,7 +323,7 @@ impl<TSpec: EthSpec> Decoder for SSZSnappyOutboundCodec<TSpec> {
     }
 }
 
-impl<TSpec: EthSpec> OutboundCodec<RPCRequest<TSpec>> for SSZSnappyOutboundCodec<TSpec> {
+impl<TSpec: EthSpec> OutboundCodec<OutboundRequest<TSpec>> for SSZSnappyOutboundCodec<TSpec> {
     type CodecErrorType = ErrorType;
 
     fn decode_error(

beacon_node/eth2_libp2p/src/rpc/handler.rs

@@ -2,9 +2,10 @@
 #![allow(clippy::cognitive_complexity)]
 
 use super::methods::{RPCCodedResponse, RPCResponseErrorCode, RequestId, ResponseTermination};
-use super::protocol::{Protocol, RPCError, RPCProtocol, RPCRequest};
+use super::protocol::{Protocol, RPCError, RPCProtocol};
 use super::{RPCReceived, RPCSend};
-use crate::rpc::protocol::{InboundFramed, OutboundFramed};
+use crate::rpc::outbound::{OutboundFramed, OutboundRequest};
+use crate::rpc::protocol::InboundFramed;
 use fnv::FnvHashMap;
 use futures::prelude::*;
 use futures::{Sink, SinkExt};
@@ -90,7 +91,7 @@ where
     events_out: SmallVec<[HandlerEvent<TSpec>; 4]>,
 
     /// Queue of outbound substreams to open.
-    dial_queue: SmallVec<[(RequestId, RPCRequest<TSpec>); 4]>,
+    dial_queue: SmallVec<[(RequestId, OutboundRequest<TSpec>); 4]>,
 
     /// Current number of concurrent outbound substreams being opened.
     dial_negotiated: u32,
@@ -186,7 +187,7 @@ pub enum OutboundSubstreamState<TSpec: EthSpec> {
         /// The framed negotiated substream.
         substream: Box<OutboundFramed<NegotiatedSubstream, TSpec>>,
         /// Keeps track of the actual request sent.
-        request: RPCRequest<TSpec>,
+        request: OutboundRequest<TSpec>,
     },
     /// Closing an outbound substream>
     Closing(Box<OutboundFramed<NegotiatedSubstream, TSpec>>),
@@ -221,7 +222,7 @@ where
     }
 
     /// Initiates the handler's shutdown process, sending an optional last message to the peer.
-    pub fn shutdown(&mut self, final_msg: Option<(RequestId, RPCRequest<TSpec>)>) {
+    pub fn shutdown(&mut self, final_msg: Option<(RequestId, OutboundRequest<TSpec>)>) {
         if matches!(self.state, HandlerState::Active) {
             if !self.dial_queue.is_empty() {
                 debug!(self.log, "Starting handler shutdown"; "unsent_queued_requests" => self.dial_queue.len());
@@ -247,7 +248,7 @@ where
     }
 
     /// Opens an outbound substream with a request.
-    fn send_request(&mut self, id: RequestId, req: RPCRequest<TSpec>) {
+    fn send_request(&mut self, id: RequestId, req: OutboundRequest<TSpec>) {
         match self.state {
             HandlerState::Active => {
                 self.dial_queue.push((id, req));
@@ -303,8 +304,8 @@ where
     type OutEvent = HandlerEvent<TSpec>;
     type Error = RPCError;
     type InboundProtocol = RPCProtocol<TSpec>;
-    type OutboundProtocol = RPCRequest<TSpec>;
-    type OutboundOpenInfo = (RequestId, RPCRequest<TSpec>); // Keep track of the id and the request
+    type OutboundProtocol = OutboundRequest<TSpec>;
+    type OutboundOpenInfo = (RequestId, OutboundRequest<TSpec>); // Keep track of the id and the request
     type InboundOpenInfo = ();
 
     fn listen_protocol(&self) -> SubstreamProtocol<Self::InboundProtocol, ()> {