Initial base codec

beacon_node/eth2-libp2p/src/rpc/codecs/base.rs

@@ -0,0 +1,442 @@
+
+///! This handles the various supported encoding mechanism for the Eth 2.0 RPC. 
+///!
+///! Currently supported encodings are:
+///! - ssz - Varint length-prefixed SSZ-encoding.
+
+
+pub struct BaseCodec<TCodec: Codec> {
+    /// Inner codec for handling various encodings
+    inner: TCodec,
+    /// Optimisation for decoding. True if the response code has been read and we are awaiting a
+    /// response.
+    read_response_code: bool,
+}
+
+
+impl<TCodec> Encoder for BaseOutboundCodec<TCodec>
+where 
+     TCodec: Encoder
+{
+    type Item = RPCRequest;
+    type Error = <TCodec as Encoder>::Error;
+
+    fn encode(
+        &mut self,
+        item: Self::Item,
+        dst: &mut BytesMut
+    ) -> Result<(), Self::Error> {
+        self.inner.encode(item, dst)
+    }
+}
+
+
+impl<TCodec> Encoder for BaseInboundCodec<TCodec>
+where 
+     TCodec: Encoder
+{
+    type Item = RPCResponse;
+    type Error = <TCodec as Encoder>::Error;
+
+    fn encode(
+        &mut self,
+        item: Self::Item,
+        dst: &mut BytesMut
+    ) -> Result<(), Self::Error> {
+
+        match item {
+            RPCResponse::Error(response) => {
+                match response = {
+                    ErrorResponse::EncodingError => {
+                        dst.clear();
+                        dst.reserve(1);
+                        dst.put(response as u8);
+                        return;
+                    }
+                    ErrorResponse::
+        self.inner.encode(item, dst)
+    }
+}
+
+
+impl<TCodec> Decoder for BaseCodec<TCodec>
+where 
+   TCodec: Decoder,
+   <TCodec as Decoder>::Error: From<RPCError>,
+{
+
+    type Item = RPCResponse;
+    type Error = <TCodec as Decoder>::Error;
+
+    fn decode(
+        &mut self,
+        src: &mut BytesMut
+    ) -> Result<Option<Self::Item>, Self::Error> {
+
+        if !self.read_response_code {
+            if src.len() < 1 {
+                return Err(io::Error::new(io::ErrorKind::InvalidData, "no bytes received"));
+            }
+
+            let resp_code_byte = [0; 1];
+            // data must be only 1-byte - this cannot panic
+            resp_code_byte.copy_from_slice(&src);
+            let response_code =
+                ResponseCode::from(u8::from_be_bytes(resp_code_byte));
+            match response_code {
+                ResponseCode::EncodingError => {
+                    // invalid encoding
+                    let response = RPCResponse::Error("Invalid Encoding".into());
+                    return Ok(Async::Ready(response));
+                }
+                ResponseCode::Success
+                | ResponseCode::InvalidRequest
+                | ResponseCode::ServerError => {
+                    // need to read another packet
+                    self.inner = RPCRequestResponseInner::Read(
+                        read_one(socket, max_size),
+                        response_code,
+                    )
+                }
+                ResponseCode::Unknown => {
+                    // unknown response code
+                    let response = RPCResponse::Error(format!(
+                        "Unknown response code: {}",
+                        (response_code as u8)
+                    ));
+                    return Ok(Async::Ready(response));
+
+
+
+
+
+    }
+
+}
+
+    
+
+
+
+/// SSZ Input stream
+pub struct SSZInboundSink<TSocket> {
+    inner: 
+    protocol: ProtocolId
+
+impl<TSocket> for SSZInputStream<TSocket>
+where
+    TSocket: AsyncRead + AsyncWrite
+{
+
+    /// Set up the initial input stream object.
+    pub fn new(incomming: TSocket, protocol: ProtocolId, max_size: usize) -> Self {
+
+        // this type of stream should only apply to ssz protocols
+        debug_assert!(protocol.encoding.as_str() == "ssz");
+
+        let mut uvi_codec = UviBytes::default();
+        uvi_codec.set_max_len(max_size);
+
+        let inner = Framed::new(incomming, uvi_codec).from_err()
+            .with(|response| {
+              self.encode(response) 
+            })
+        .and_then(|bytes| {
+            self.decode(request)
+        }).into_future();
+
+        //TODO: add timeout
+
+        SSZInputStream {
+            inner,
+            protocol
+        }
+    }
+
+    /// Decodes an SSZ-encoded RPCRequest.
+    fn decode(&self, request: RPCRequest) {
+
+        match self.protocol.message_name.as_str() {
+            "hello" => match protocol.version.as_str() {
+                "1.0.0" =>  Ok(RPCRequest::Hello(HelloMessage::from_ssz_bytes(&packet)?)),
+                _ => Err(RPCError::InvalidProtocol("Unknown HELLO version")),
+            },
+            "goodbye" => match protocol.version.as_str() {
+                "1.0.0" =>  Ok(RPCRequest::Goodbye(Goodbye::from_ssz_bytes(&packet)?)),
+                _ => Err(RPCError::InvalidProtocol(
+                    "Unknown GOODBYE version.as_str()",
+                )),
+            },
+            "beacon_block_roots" => match protocol.version.as_str() {
+                "1.0.0" => Ok(RPCRequest::BeaconBlockRoots(
+                        BeaconBlockRootsRequest::from_ssz_bytes(&packet)?,
+                    )),
+                _ => Err(RPCError::InvalidProtocol(
+                    "Unknown BEACON_BLOCK_ROOTS version.",
+                )),
+            },
+            "beacon_block_headers" => match protocol.version.as_str() {
+                "1.0.0" => Ok(RPCRequest::BeaconBlockHeaders(
+                        BeaconBlockHeadersRequest::from_ssz_bytes(&packet)?,
+                    )),
+                _ => Err(RPCError::InvalidProtocol(
+                    "Unknown BEACON_BLOCK_HEADERS version.",
+                )),
+            },
+            "beacon_block_bodies" => match protocol.version.as_str() {
+                "1.0.0" =>  Ok(RPCRequest::BeaconBlockBodies(
+                        BeaconBlockBodiesRequest::from_ssz_bytes(&packet)?,
+                    )),
+                _ => Err(RPCError::InvalidProtocol(
+                    "Unknown BEACON_BLOCK_BODIES version.",
+                )),
+            },
+            "beacon_chain_state" => match protocol.version.as_str() {
+                "1.0.0" =>  Ok(RPCRequest::BeaconChainState(
+                        BeaconChainStateRequest::from_ssz_bytes(&packet)?,
+                    )),
+                _ => Err(RPCError::InvalidProtocol(
+                    "Unknown BEACON_CHAIN_STATE version.",
+                )),
+            },
+        }
+    }
+
+    fn encode(&self, response: RPCResponse) {
+
+        // TODO: Add error code
+
+        match response {
+            RPCResponse::Hello(res) => res.as_ssz_bytes(),
+            RPCResponse::Goodbye => unreachable!(),
+            RPCResponse::BeaconBlockRoots(res) => res.as_ssz_bytes(),
+            RPCResponse::BeaconBlockHeaders(res) => res.headers, // already raw bytes
+            RPCResponse::BeaconBlockBodies(res) => res.block_bodies, // already raw bytes
+            RPCResponse::BeaconChainState(res) => res.as_ssz_bytes(),
+        }
+    }
+
+}
+
+type SSZInboundOutput = stream::AndThen<sink::With<stream::FromErr<Framed<TSocket, UviBytes<Vec<u8>>>, RPCError>,
+    RPCResponse,
+    fn(RPCResponse) -> Result<Vec<u8>, RPCError>, 
+    Result<Vec<u8>, RPCError>,
+    >,
+    fn(BytesMut) -> Result<RPCRequest, RPCError>,
+    Result<RPCRequest, RPCError>
+    >;
+
+impl<TSocket> Sink for SSZInputStreamSink<TSocket> {
+
+    type SinkItem = RPCResponse;
+    type SinkError = RPCError;
+
+    fn start_send(
+    &mut self,
+    item: Self::SinkItem
+) -> Result<AsyncSink<Self::SinkItem>, Self::SinkError> {
+        self.inner.start_send(item)
+    }
+
+    fn poll_complete(&mut self) -> Result<Async<()>, Self::SinkError> {
+        self.inner.poll_complete()
+    }
+}
+
+/* Outbound specific stream */
+
+// Implement our own decoder to handle the response byte
+
+struct SSZOutboundCodec
+
+
+
+pub struct SSZOutboundStreamSink<TSocket> {
+    inner: 
+    protocol: ProtocolId
+
+impl<TSocket> for SSZOutboundStreamSink<TSocket>
+where
+    TSocket: AsyncRead + AsyncWrite
+{
+
+    /// Set up the initial outbound stream object.
+    pub fn new(socket: TSocket, protocol: ProtocolId, max_size: usize) -> Self {
+
+        // this type of stream should only apply to ssz protocols
+        debug_assert!(protocol.encoding.as_str() == "ssz");
+
+        let mut uvi_codec = UviBytes::default();
+        uvi_codec.set_max_len(max_size);
+
+        let inner = Framed::new(socket, uvi_codec).from_err()
+            .with(|request| {
+              self.encode(request) 
+            })
+        .and_then(|bytes| {
+            self.decode(response)
+        });
+
+        SSZOutboundStream {
+            inner,
+            protocol
+        }
+    }
+
+
+
+
+
+    /// Decodes a response that was received on the same stream as a request. The response type should
+    /// therefore match the request protocol type.
+    pub fn decode(&self, response: Vec<u8>,
+        protocol: ProtocolId,
+        response_code: ResponseCode,
+    ) -> Result<Self, RPCError> {
+        match response_code {
+            ResponseCode::EncodingError => Ok(RPCResponse::Error("Encoding error".into())),
+            ResponseCode::InvalidRequest => {
+                let response = match protocol.encoding.as_str() {
+                    "ssz" => ErrorResponse::from_ssz_bytes(&packet)?,
+                    _ => return Err(RPCError::InvalidProtocol("Unknown Encoding")),
+                };
+                Ok(RPCResponse::Error(format!(
+                    "Invalid Request: {}",
+                    response.error_message
+                )))
+            }
+            ResponseCode::ServerError => {
+                let response = match protocol.encoding.as_str() {
+                    "ssz" => ErrorResponse::from_ssz_bytes(&packet)?,
+                    _ => return Err(RPCError::InvalidProtocol("Unknown Encoding")),
+                };
+                Ok(RPCResponse::Error(format!(
+                    "Remote Server Error: {}",
+                    response.error_message
+                )))
+            }
+            ResponseCode::Success => match protocol.message_name.as_str() {
+                "hello" => match protocol.version.as_str() {
+                    "1.0.0" => match protocol.encoding.as_str() {
+                        "ssz" => Ok(RPCResponse::Hello(HelloMessage::from_ssz_bytes(&packet)?)),
+                        _ => Err(RPCError::InvalidProtocol("Unknown HELLO encoding")),
+                    },
+                    _ => Err(RPCError::InvalidProtocol("Unknown HELLO version.")),
+                },
+                "goodbye" => Err(RPCError::Custom(
+                    "GOODBYE should not have a response".into(),
+                )),
+                "beacon_block_roots" => match protocol.version.as_str() {
+                    "1.0.0" => match protocol.encoding.as_str() {
+                        "ssz" => Ok(RPCResponse::BeaconBlockRoots(
+                            BeaconBlockRootsResponse::from_ssz_bytes(&packet)?,
+                        )),
+                        _ => Err(RPCError::InvalidProtocol(
+                            "Unknown BEACON_BLOCK_ROOTS encoding",
+                        )),
+                    },
+                    _ => Err(RPCError::InvalidProtocol(
+                        "Unknown BEACON_BLOCK_ROOTS version.",
+                    )),
+                },
+                "beacon_block_headers" => match protocol.version.as_str() {
+                    "1.0.0" => match protocol.encoding.as_str() {
+                        "ssz" => Ok(RPCResponse::BeaconBlockHeaders(
+                            BeaconBlockHeadersResponse { headers: packet },
+                        )),
+                        _ => Err(RPCError::InvalidProtocol(
+                            "Unknown BEACON_BLOCK_HEADERS encoding",
+                        )),
+                    },
+                    _ => Err(RPCError::InvalidProtocol(
+                        "Unknown BEACON_BLOCK_HEADERS version.",
+                    )),
+                },
+                "beacon_block_bodies" => match protocol.version.as_str() {
+                    "1.0.0" => match protocol.encoding.as_str() {
+                        "ssz" => Ok(RPCResponse::BeaconBlockBodies(BeaconBlockBodiesResponse {
+                            block_bodies: packet,
+                        })),
+                        _ => Err(RPCError::InvalidProtocol(
+                            "Unknown BEACON_BLOCK_BODIES encoding",
+                        )),
+                    },
+                    _ => Err(RPCError::InvalidProtocol(
+                        "Unknown BEACON_BLOCK_BODIES version.",
+                    )),
+                },
+                "beacon_chain_state" => match protocol.version.as_str() {
+                    "1.0.0" => match protocol.encoding.as_str() {
+                        "ssz" => Ok(RPCResponse::BeaconChainState(
+                            BeaconChainStateResponse::from_ssz_bytes(&packet)?,
+                        )),
+                        _ => Err(RPCError::InvalidProtocol(
+                            "Unknown BEACON_CHAIN_STATE encoding",
+                        )),
+                    },
+                    _ => Err(RPCError::InvalidProtocol(
+                        "Unknown BEACON_CHAIN_STATE version.",
+                    )),
+                },
+            },
+        }
+    }
+
+    fn encode(&self, response: RPCResponse) {
+
+        match response {
+            RPCResponse::Hello(res) => res.as_ssz_bytes(),
+            RPCResponse::Goodbye => unreachable!(),
+            RPCResponse::BeaconBlockRoots(res) => res.as_ssz_bytes(),
+            RPCResponse::BeaconBlockHeaders(res) => res.headers, // already raw bytes
+            RPCResponse::BeaconBlockBodies(res) => res.block_bodies, // already raw bytes
+            RPCResponse::BeaconChainState(res) => res.as_ssz_bytes(),
+        }
+    }
+
+}
+
+type SSZOutboundStream = stream::AndThen<sink::With<stream::FromErr<Framed<TSocket, UviBytes<Vec<u8>>>, RPCError>,
+    RPCResponse,
+    fn(RPCResponse) -> Result<Vec<u8>, RPCError>, 
+    Result<Vec<u8>, RPCError>,
+    >,
+    fn(BytesMut) -> Result<RPCRequest, RPCError>,
+    Result<RPCRequest, RPCError>
+    >;
+
+
+impl<TSocket> Stream for SSZInputStreamSink<TSocket> {
+
+    type Item = SSZInboundOutput;
+    type Error = RPCError;
+
+    fn poll(&mut self) -> Result<Async<Option<Self::Item>>, Self::Error>  {
+        self.inner.poll()
+    }
+}
+
+impl<TSocket> Sink for SSZInputStreamSink<TSocket> {
+
+    type SinkItem = RPCResponse;
+    type SinkError = RPCError;
+
+    fn start_send(
+    &mut self,
+    item: Self::SinkItem
+) -> Result<AsyncSink<Self::SinkItem>, Self::SinkError> {
+        self.inner.start_send(item)
+    }
+
+    fn poll_complete(&mut self) -> Result<Async<()>, Self::SinkError> {
+        self.inner.poll_complete()
+    }
+}
+
+
+
+
+
+
+

beacon_node/eth2-libp2p/src/rpc/protocol.rs

@@ -212,81 +212,6 @@ impl RPCRequest {
             RPCRequest::BeaconChainState(req) => req.as_ssz_bytes(),
         }
     }
-
-    // This function can be extended to provide further logic for supporting various protocol versions/encoding
-    /// Decodes a request received from our peer.
-    pub fn decode(packet: Vec<u8>, protocol: ProtocolId) -> Result<Self, RPCError> {
-        match protocol.message_name.as_str() {
-            "hello" => match protocol.version.as_str() {
-                "1.0.0" => match protocol.encoding.as_str() {
-                    "ssz" => Ok(RPCRequest::Hello(HelloMessage::from_ssz_bytes(&packet)?)),
-                    _ => Err(RPCError::InvalidProtocol("Unknown HELLO encoding")),
-                },
-                _ => Err(RPCError::InvalidProtocol("Unknown HELLO version")),
-            },
-            "goodbye" => match protocol.version.as_str() {
-                "1.0.0" => match protocol.encoding.as_str() {
-                    "ssz" => Ok(RPCRequest::Goodbye(Goodbye::from_ssz_bytes(&packet)?)),
-                    _ => Err(RPCError::InvalidProtocol("Unknown GOODBYE encoding")),
-                },
-                _ => Err(RPCError::InvalidProtocol(
-                    "Unknown GOODBYE version.as_str()",
-                )),
-            },
-            "beacon_block_roots" => match protocol.version.as_str() {
-                "1.0.0" => match protocol.encoding.as_str() {
-                    "ssz" => Ok(RPCRequest::BeaconBlockRoots(
-                        BeaconBlockRootsRequest::from_ssz_bytes(&packet)?,
-                    )),
-                    _ => Err(RPCError::InvalidProtocol(
-                        "Unknown BEACON_BLOCK_ROOTS encoding",
-                    )),
-                },
-                _ => Err(RPCError::InvalidProtocol(
-                    "Unknown BEACON_BLOCK_ROOTS version.",
-                )),
-            },
-            "beacon_block_headers" => match protocol.version.as_str() {
-                "1.0.0" => match protocol.encoding.as_str() {
-                    "ssz" => Ok(RPCRequest::BeaconBlockHeaders(
-                        BeaconBlockHeadersRequest::from_ssz_bytes(&packet)?,
-                    )),
-                    _ => Err(RPCError::InvalidProtocol(
-                        "Unknown BEACON_BLOCK_HEADERS encoding",
-                    )),
-                },
-                _ => Err(RPCError::InvalidProtocol(
-                    "Unknown BEACON_BLOCK_HEADERS version.",
-                )),
-            },
-            "beacon_block_bodies" => match protocol.version.as_str() {
-                "1.0.0" => match protocol.encoding.as_str() {
-                    "ssz" => Ok(RPCRequest::BeaconBlockBodies(
-                        BeaconBlockBodiesRequest::from_ssz_bytes(&packet)?,
-                    )),
-                    _ => Err(RPCError::InvalidProtocol(
-                        "Unknown BEACON_BLOCK_BODIES encoding",
-                    )),
-                },
-                _ => Err(RPCError::InvalidProtocol(
-                    "Unknown BEACON_BLOCK_BODIES version.",
-                )),
-            },
-            "beacon_chain_state" => match protocol.version.as_str() {
-                "1.0.0" => match protocol.encoding.as_str() {
-                    "ssz" => Ok(RPCRequest::BeaconChainState(
-                        BeaconChainStateRequest::from_ssz_bytes(&packet)?,
-                    )),
-                    _ => Err(RPCError::InvalidProtocol(
-                        "Unknown BEACON_CHAIN_STATE encoding",
-                    )),
-                },
-                _ => Err(RPCError::InvalidProtocol(
-                    "Unknown BEACON_CHAIN_STATE version.",
-                )),
-            },
-        }
-    }
 }
 
 /* Response Type */
@@ -305,163 +230,61 @@ pub enum RPCResponse {
     BeaconBlockBodies(BeaconBlockBodiesResponse),
     /// A response to a get BEACON_CHAIN_STATE request.
     BeaconChainState(BeaconChainStateResponse),
-    /// The Error returned from the peer during a request.
-    Error(String),
 }
 
-pub enum ResponseCode {
+pub enum RPCErrorResponse {
-    Success = 0,
+    Success(RPCResponse),
-    EncodingError = 1,
+    EncodingError,
-    InvalidRequest = 2,
+    InvalidRequest(ErrorMessage),
-    ServerError = 3,
+    ServerError(ErrorMessage),
-    Unknown = 255,
+    Unknown(ErrorMessage),
 }
 
-impl From<u8> for ResponseCode {
+impl RPCErrorResponse {
-    fn from(val: u8) -> ResponseCode {
+    /// If a response has no payload, returns the variant corresponding to the code.
-        match val {
+    pub fn internal_data(response_code: u8) -> Option<RPCErrorResponse> {
-            0 => ResponseCode::Success,
+        match response_code {
-            1 => ResponseCode::EncodingError,
+            // EncodingError
-            2 => ResponseCode::InvalidRequest,
+            1 => Some(RPCErrorResponse::EncodingError),
-            3 => ResponseCode::ServerError,
+            // All others require further data
-            _ => ResponseCode::Unknown,
+            _ => None,
         }
     }
-}
 
-impl Into<u8> for ResponseCode {
+    pub fn as_u8(&self) -> u8 {
-    fn into(self) -> u8 {
+        match self {
-        self as u8
+            RPCErrorResponse::Success(_) => 0,
+            RPCErrorResponse::EncodingError => 1,
+            RPCErrorResponse::InvalidRequest(_) => 2,
+            RPCErrorResponse::ServerError(_) => 3,
+            RPCErrorResponse::Unknown(_) => 255,
+        }
     }
+
+    /// 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 RPCErrorResponse from a response code and an ErrorMessage
+    pub fn from_error(response_code:u8, err: ErrorMessage) -> Self { 
+        match response_code {
+            2 => RPCErrorResponse::InvalidRequest(err),
+            3 => RPCErrorResponse::ServerError(err),
+            _ => RPCErrorResponse::Unknown(err),
+        }
 }
 
 #[derive(Encode, Decode)]
-struct ErrorResponse {
+struct ErrorMessage {
-    error_message: String,
+    /// The UTF-8 encoded Error message string.
+    error_message: Vec<u8>,
 }
 
-impl RPCResponse {
+// todo: SSZ-Encode
-    /// Decodes a response that was received on the same stream as a request. The response type should
+impl RPCResponse {}
-    /// therefore match the request protocol type.
-    pub fn decode(
-        packet: Vec<u8>,
-        protocol: ProtocolId,
-        response_code: ResponseCode,
-    ) -> Result<Self, RPCError> {
-        match response_code {
-            ResponseCode::EncodingError => Ok(RPCResponse::Error("Encoding error".into())),
-            ResponseCode::InvalidRequest => {
-                let response = match protocol.encoding.as_str() {
-                    "ssz" => ErrorResponse::from_ssz_bytes(&packet)?,
-                    _ => return Err(RPCError::InvalidProtocol("Unknown Encoding")),
-                };
-                Ok(RPCResponse::Error(format!(
-                    "Invalid Request: {}",
-                    response.error_message
-                )))
-            }
-            ResponseCode::ServerError => {
-                let response = match protocol.encoding.as_str() {
-                    "ssz" => ErrorResponse::from_ssz_bytes(&packet)?,
-                    _ => return Err(RPCError::InvalidProtocol("Unknown Encoding")),
-                };
-                Ok(RPCResponse::Error(format!(
-                    "Remote Server Error: {}",
-                    response.error_message
-                )))
-            }
-            ResponseCode::Success => match protocol.message_name.as_str() {
-                "hello" => match protocol.version.as_str() {
-                    "1.0.0" => match protocol.encoding.as_str() {
-                        "ssz" => Ok(RPCResponse::Hello(HelloMessage::from_ssz_bytes(&packet)?)),
-                        _ => Err(RPCError::InvalidProtocol("Unknown HELLO encoding")),
-                    },
-                    _ => Err(RPCError::InvalidProtocol("Unknown HELLO version.")),
-                },
-                "goodbye" => Err(RPCError::Custom(
-                    "GOODBYE should not have a response".into(),
-                )),
-                "beacon_block_roots" => match protocol.version.as_str() {
-                    "1.0.0" => match protocol.encoding.as_str() {
-                        "ssz" => Ok(RPCResponse::BeaconBlockRoots(
-                            BeaconBlockRootsResponse::from_ssz_bytes(&packet)?,
-                        )),
-                        _ => Err(RPCError::InvalidProtocol(
-                            "Unknown BEACON_BLOCK_ROOTS encoding",
-                        )),
-                    },
-                    _ => Err(RPCError::InvalidProtocol(
-                        "Unknown BEACON_BLOCK_ROOTS version.",
-                    )),
-                },
-                "beacon_block_headers" => match protocol.version.as_str() {
-                    "1.0.0" => match protocol.encoding.as_str() {
-                        "ssz" => Ok(RPCResponse::BeaconBlockHeaders(
-                            BeaconBlockHeadersResponse { headers: packet },
-                        )),
-                        _ => Err(RPCError::InvalidProtocol(
-                            "Unknown BEACON_BLOCK_HEADERS encoding",
-                        )),
-                    },
-                    _ => Err(RPCError::InvalidProtocol(
-                        "Unknown BEACON_BLOCK_HEADERS version.",
-                    )),
-                },
-                "beacon_block_bodies" => match protocol.version.as_str() {
-                    "1.0.0" => match protocol.encoding.as_str() {
-                        "ssz" => Ok(RPCResponse::BeaconBlockBodies(BeaconBlockBodiesResponse {
-                            block_bodies: packet,
-                        })),
-                        _ => Err(RPCError::InvalidProtocol(
-                            "Unknown BEACON_BLOCK_BODIES encoding",
-                        )),
-                    },
-                    _ => Err(RPCError::InvalidProtocol(
-                        "Unknown BEACON_BLOCK_BODIES version.",
-                    )),
-                },
-                "beacon_chain_state" => match protocol.version.as_str() {
-                    "1.0.0" => match protocol.encoding.as_str() {
-                        "ssz" => Ok(RPCResponse::BeaconChainState(
-                            BeaconChainStateResponse::from_ssz_bytes(&packet)?,
-                        )),
-                        _ => Err(RPCError::InvalidProtocol(
-                            "Unknown BEACON_CHAIN_STATE encoding",
-                        )),
-                    },
-                    _ => Err(RPCError::InvalidProtocol(
-                        "Unknown BEACON_CHAIN_STATE version.",
-                    )),
-                },
-            },
-        }
-    }
-
-    /// Encodes the Response object based on the negotiated protocol.
-    pub fn encode(&self, protocol: ProtocolId) -> Result<Vec<u8>, RPCError> {
-        // Match on the encoding and in the future, the version
-        match protocol.encoding.as_str() {
-            "ssz" => Ok(self.ssz_encode()),
-            _ => {
-                return Err(RPCError::Custom(format!(
-                    "Unknown Encoding: {}",
-                    protocol.encoding
-                )))
-            }
-        }
-    }
-
-    fn ssz_encode(&self) -> Vec<u8> {
-        match self {
-            RPCResponse::Hello(res) => res.as_ssz_bytes(),
-            RPCResponse::Goodbye => unreachable!(),
-            RPCResponse::BeaconBlockRoots(res) => res.as_ssz_bytes(),
-            RPCResponse::BeaconBlockHeaders(res) => res.headers, // already raw bytes
-            RPCResponse::BeaconBlockBodies(res) => res.block_bodies, // already raw bytes
-            RPCResponse::BeaconChainState(res) => res.as_ssz_bytes(),
-        }
-    }
-}
 
 /* Outbound upgrades */
 

codec.rs

@@ -0,0 +1,461 @@
+
+///! This handles the various supported encoding mechanism for the Eth 2.0 RPC. 
+///!
+///! Currently supported encodings are:
+///! - ssz - Varint length-prefixed SSZ-encoding.
+
+
+pub trait InnerCodec: Encoder + Decoder {
+    type Error;
+
+    pub fn decode_error(&mut self, &mut BytesMut) -> Result<Option<Self::Error>, <Self as Decoder>::Error>;
+}
+
+pub struct BaseInboundCodec<TCodec: InnerCodec> {
+    /// Inner codec for handling various encodings
+    inner: TCodec,
+}
+
+impl<TCodec> Encoder for BaseInboundCodec<TCodec>
+where 
+     TCodec: Encoder,
+     <TCodec as Encoder>::Item = RPCResponse,
+{
+    type Item = RPCResponse;
+    type Error = <TCodec as Encoder>::Error;
+
+    fn encode(
+        &mut self,
+        item: Self::Item,
+        dst: &mut BytesMut
+    ) -> Result<(), Self::Error> {
+        dst.clear();
+        dst.reserve(1);
+        dst.put_u8(item.as_u8);
+        return self.inner.encode();
+    }
+}
+
+
+impl<TCodec> Decoder for BaseInboundCodec<TCodec>
+where 
+   TCodec: Decoder,
+   <TCodec as Decoder>::Item: RPCrequest,
+   <TCodec as Decoder>::Error: From<RPCError>,
+{
+
+    type Item = RPCRequest;
+    type Error = <TCodec as Decoder>::Error;
+
+    fn decode(
+        &mut self,
+        src: &mut BytesMut
+    ) -> Result<Option<Self::Item>, Self::Error> {
+        self.inner.decode(src)
+    }
+}
+
+pub struct BaseOutboundCodec<TCodec> 
+where
+    TCodec: InnerCodec,
+    <TCodec as Decoder>::Item = RPCResponse,
+    <TCodec as InnerCodec>::ErrorItem = ErrorMessage,
+    {
+    /// Inner codec for handling various encodings
+    inner: TCodec,
+    /// Optimisation for decoding. True if the response code has been read and we are awaiting a
+    /// response.
+    response_code: Option<u8>,
+}
+
+impl<TCodec> Encoder for BaseOutboundCodec<TCodec>
+where 
+     TCodec: Encoder
+{
+    type Item = RPCRequest;
+    type Error = <TCodec as Encoder>::Error;
+
+    fn encode(
+        &mut self,
+        item: Self::Item,
+        dst: &mut BytesMut
+    ) -> Result<(), Self::Error> {
+        self.inner.encode(item, dst)
+    }
+}
+
+
+impl<TCodec> Decoder for BaseOutboundCodec<TCodec>
+where 
+   TCodec: InnerCodec,
+   <TCodec as Decoder>::Error: From<RPCError>,
+{
+
+    type Item = RPCResponse;
+    type Error = <TCodec as Decoder>::Error;
+
+    fn decode(
+        &mut self,
+        src: &mut BytesMut
+    ) -> Result<Option<Self::Item>, Self::Error> {
+
+
+        let response_code = {
+            if let Some(resp_code) =  self.response_code {
+                resp_code;
+            }
+            else {
+                if src.is_empty() {
+                    return Err(io::Error::new(io::ErrorKind::InvalidData, "no bytes received"));
+                }
+                let resp_byte = src.split_to(1);
+                let resp_code_byte = [0; 1];
+                resp_code_byte.copy_from_slice(&resp_byte);
+
+                let resp_code = u8::from_be_bytes(resp_code_byte);
+
+                if let Some(response) = RPCErrorResponse::internal_data(resp_code) {
+                    self.response_code = None;
+                    return response;
+                }
+                resp_code 
+            }
+        };
+
+        if RPCErrorResponse::is_response(response_code) {
+            // decode an actual response
+            return self.inner.decode(src).map(|r| r.map(|resp| RPCErrorResponse::Success(resp)));
+        }
+        else {
+            // decode an error
+            return self.inner.decode_error(src).map(|r| r.map(|resp| RPCErrorResponse::from_error(response_code, resp)));
+        }
+
+    }
+
+}
+
+
+
+/// SSZ Input stream
+pub struct SSZInboundSink<TSocket> {
+    inner: 
+    protocol: ProtocolId
+
+impl<TSocket> for SSZInputStream<TSocket>
+where
+    TSocket: AsyncRead + AsyncWrite
+{
+
+    /// Set up the initial input stream object.
+    pub fn new(incomming: TSocket, protocol: ProtocolId, max_size: usize) -> Self {
+
+        // this type of stream should only apply to ssz protocols
+        debug_assert!(protocol.encoding.as_str() == "ssz");
+
+        let mut uvi_codec = UviBytes::default();
+        uvi_codec.set_max_len(max_size);
+
+        let inner = Framed::new(incomming, uvi_codec).from_err()
+            .with(|response| {
+              self.encode(response) 
+            })
+        .and_then(|bytes| {
+            self.decode(request)
+        }).into_future();
+
+        //TODO: add timeout
+
+        SSZInputStream {
+            inner,
+            protocol
+        }
+    }
+
+    /// Decodes an SSZ-encoded RPCRequest.
+    fn decode(&self, request: RPCRequest) {
+
+        match self.protocol.message_name.as_str() {
+            "hello" => match protocol.version.as_str() {
+                "1.0.0" =>  Ok(RPCRequest::Hello(HelloMessage::from_ssz_bytes(&packet)?)),
+                _ => Err(RPCError::InvalidProtocol("Unknown HELLO version")),
+            },
+            "goodbye" => match protocol.version.as_str() {
+                "1.0.0" =>  Ok(RPCRequest::Goodbye(Goodbye::from_ssz_bytes(&packet)?)),
+                _ => Err(RPCError::InvalidProtocol(
+                    "Unknown GOODBYE version.as_str()",
+                )),
+            },
+            "beacon_block_roots" => match protocol.version.as_str() {
+                "1.0.0" => Ok(RPCRequest::BeaconBlockRoots(
+                        BeaconBlockRootsRequest::from_ssz_bytes(&packet)?,
+                    )),
+                _ => Err(RPCError::InvalidProtocol(
+                    "Unknown BEACON_BLOCK_ROOTS version.",
+                )),
+            },
+            "beacon_block_headers" => match protocol.version.as_str() {
+                "1.0.0" => Ok(RPCRequest::BeaconBlockHeaders(
+                        BeaconBlockHeadersRequest::from_ssz_bytes(&packet)?,
+                    )),
+                _ => Err(RPCError::InvalidProtocol(
+                    "Unknown BEACON_BLOCK_HEADERS version.",
+                )),
+            },
+            "beacon_block_bodies" => match protocol.version.as_str() {
+                "1.0.0" =>  Ok(RPCRequest::BeaconBlockBodies(
+                        BeaconBlockBodiesRequest::from_ssz_bytes(&packet)?,
+                    )),
+                _ => Err(RPCError::InvalidProtocol(
+                    "Unknown BEACON_BLOCK_BODIES version.",
+                )),
+            },
+            "beacon_chain_state" => match protocol.version.as_str() {
+                "1.0.0" =>  Ok(RPCRequest::BeaconChainState(
+                        BeaconChainStateRequest::from_ssz_bytes(&packet)?,
+                    )),
+                _ => Err(RPCError::InvalidProtocol(
+                    "Unknown BEACON_CHAIN_STATE version.",
+                )),
+            },
+        }
+    }
+
+    fn encode(&self, response: RPCResponse) {
+
+        // TODO: Add error code
+
+        match response {
+            RPCResponse::Hello(res) => res.as_ssz_bytes(),
+            RPCResponse::Goodbye => unreachable!(),
+            RPCResponse::BeaconBlockRoots(res) => res.as_ssz_bytes(),
+            RPCResponse::BeaconBlockHeaders(res) => res.headers, // already raw bytes
+            RPCResponse::BeaconBlockBodies(res) => res.block_bodies, // already raw bytes
+            RPCResponse::BeaconChainState(res) => res.as_ssz_bytes(),
+        }
+    }
+
+}
+
+type SSZInboundOutput = stream::AndThen<sink::With<stream::FromErr<Framed<TSocket, UviBytes<Vec<u8>>>, RPCError>,
+    RPCResponse,
+    fn(RPCResponse) -> Result<Vec<u8>, RPCError>, 
+    Result<Vec<u8>, RPCError>,
+    >,
+    fn(BytesMut) -> Result<RPCRequest, RPCError>,
+    Result<RPCRequest, RPCError>
+    >;
+
+impl<TSocket> Sink for SSZInputStreamSink<TSocket> {
+
+    type SinkItem = RPCResponse;
+    type SinkError = RPCError;
+
+    fn start_send(
+    &mut self,
+    item: Self::SinkItem
+) -> Result<AsyncSink<Self::SinkItem>, Self::SinkError> {
+        self.inner.start_send(item)
+    }
+
+    fn poll_complete(&mut self) -> Result<Async<()>, Self::SinkError> {
+        self.inner.poll_complete()
+    }
+}
+
+/* Outbound specific stream */
+
+// Implement our own decoder to handle the response byte
+
+struct SSZOutboundCodec
+
+
+
+pub struct SSZOutboundStreamSink<TSocket> {
+    inner: 
+    protocol: ProtocolId
+
+impl<TSocket> for SSZOutboundStreamSink<TSocket>
+where
+    TSocket: AsyncRead + AsyncWrite
+{
+
+    /// Set up the initial outbound stream object.
+    pub fn new(socket: TSocket, protocol: ProtocolId, max_size: usize) -> Self {
+
+        // this type of stream should only apply to ssz protocols
+        debug_assert!(protocol.encoding.as_str() == "ssz");
+
+        let mut uvi_codec = UviBytes::default();
+        uvi_codec.set_max_len(max_size);
+
+        let inner = Framed::new(socket, uvi_codec).from_err()
+            .with(|request| {
+              self.encode(request) 
+            })
+        .and_then(|bytes| {
+            self.decode(response)
+        });
+
+        SSZOutboundStream {
+            inner,
+            protocol
+        }
+    }
+
+
+
+
+
+    /// Decodes a response that was received on the same stream as a request. The response type should
+    /// therefore match the request protocol type.
+    pub fn decode(&self, response: Vec<u8>,
+        protocol: ProtocolId,
+        response_code: ResponseCode,
+    ) -> Result<Self, RPCError> {
+        match response_code {
+            ResponseCode::EncodingError => Ok(RPCResponse::Error("Encoding error".into())),
+            ResponseCode::InvalidRequest => {
+                let response = match protocol.encoding.as_str() {
+                    "ssz" => ErrorResponse::from_ssz_bytes(&packet)?,
+                    _ => return Err(RPCError::InvalidProtocol("Unknown Encoding")),
+                };
+                Ok(RPCResponse::Error(format!(
+                    "Invalid Request: {}",
+                    response.error_message
+                )))
+            }
+            ResponseCode::ServerError => {
+                let response = match protocol.encoding.as_str() {
+                    "ssz" => ErrorResponse::from_ssz_bytes(&packet)?,
+                    _ => return Err(RPCError::InvalidProtocol("Unknown Encoding")),
+                };
+                Ok(RPCResponse::Error(format!(
+                    "Remote Server Error: {}",
+                    response.error_message
+                )))
+            }
+            ResponseCode::Success => match protocol.message_name.as_str() {
+                "hello" => match protocol.version.as_str() {
+                    "1.0.0" => match protocol.encoding.as_str() {
+                        "ssz" => Ok(RPCResponse::Hello(HelloMessage::from_ssz_bytes(&packet)?)),
+                        _ => Err(RPCError::InvalidProtocol("Unknown HELLO encoding")),
+                    },
+                    _ => Err(RPCError::InvalidProtocol("Unknown HELLO version.")),
+                },
+                "goodbye" => Err(RPCError::Custom(
+                    "GOODBYE should not have a response".into(),
+                )),
+                "beacon_block_roots" => match protocol.version.as_str() {
+                    "1.0.0" => match protocol.encoding.as_str() {
+                        "ssz" => Ok(RPCResponse::BeaconBlockRoots(
+                            BeaconBlockRootsResponse::from_ssz_bytes(&packet)?,
+                        )),
+                        _ => Err(RPCError::InvalidProtocol(
+                            "Unknown BEACON_BLOCK_ROOTS encoding",
+                        )),
+                    },
+                    _ => Err(RPCError::InvalidProtocol(
+                        "Unknown BEACON_BLOCK_ROOTS version.",
+                    )),
+                },
+                "beacon_block_headers" => match protocol.version.as_str() {
+                    "1.0.0" => match protocol.encoding.as_str() {
+                        "ssz" => Ok(RPCResponse::BeaconBlockHeaders(
+                            BeaconBlockHeadersResponse { headers: packet },
+                        )),
+                        _ => Err(RPCError::InvalidProtocol(
+                            "Unknown BEACON_BLOCK_HEADERS encoding",
+                        )),
+                    },
+                    _ => Err(RPCError::InvalidProtocol(
+                        "Unknown BEACON_BLOCK_HEADERS version.",
+                    )),
+                },
+                "beacon_block_bodies" => match protocol.version.as_str() {
+                    "1.0.0" => match protocol.encoding.as_str() {
+                        "ssz" => Ok(RPCResponse::BeaconBlockBodies(BeaconBlockBodiesResponse {
+                            block_bodies: packet,
+                        })),
+                        _ => Err(RPCError::InvalidProtocol(
+                            "Unknown BEACON_BLOCK_BODIES encoding",
+                        )),
+                    },
+                    _ => Err(RPCError::InvalidProtocol(
+                        "Unknown BEACON_BLOCK_BODIES version.",
+                    )),
+                },
+                "beacon_chain_state" => match protocol.version.as_str() {
+                    "1.0.0" => match protocol.encoding.as_str() {
+                        "ssz" => Ok(RPCResponse::BeaconChainState(
+                            BeaconChainStateResponse::from_ssz_bytes(&packet)?,
+                        )),
+                        _ => Err(RPCError::InvalidProtocol(
+                            "Unknown BEACON_CHAIN_STATE encoding",
+                        )),
+                    },
+                    _ => Err(RPCError::InvalidProtocol(
+                        "Unknown BEACON_CHAIN_STATE version.",
+                    )),
+                },
+            },
+        }
+    }
+
+    fn encode(&self, response: RPCResponse) {
+
+        match response {
+            RPCResponse::Hello(res) => res.as_ssz_bytes(),
+            RPCResponse::Goodbye => unreachable!(),
+            RPCResponse::BeaconBlockRoots(res) => res.as_ssz_bytes(),
+            RPCResponse::BeaconBlockHeaders(res) => res.headers, // already raw bytes
+            RPCResponse::BeaconBlockBodies(res) => res.block_bodies, // already raw bytes
+            RPCResponse::BeaconChainState(res) => res.as_ssz_bytes(),
+        }
+    }
+
+}
+
+type SSZOutboundStream = stream::AndThen<sink::With<stream::FromErr<Framed<TSocket, UviBytes<Vec<u8>>>, RPCError>,
+    RPCResponse,
+    fn(RPCResponse) -> Result<Vec<u8>, RPCError>, 
+    Result<Vec<u8>, RPCError>,
+    >,
+    fn(BytesMut) -> Result<RPCRequest, RPCError>,
+    Result<RPCRequest, RPCError>
+    >;
+
+
+impl<TSocket> Stream for SSZInputStreamSink<TSocket> {
+
+    type Item = SSZInboundOutput;
+    type Error = RPCError;
+
+    fn poll(&mut self) -> Result<Async<Option<Self::Item>>, Self::Error>  {
+        self.inner.poll()
+    }
+}
+
+impl<TSocket> Sink for SSZInputStreamSink<TSocket> {
+
+    type SinkItem = RPCResponse;
+    type SinkError = RPCError;
+
+    fn start_send(
+    &mut self,
+    item: Self::SinkItem
+) -> Result<AsyncSink<Self::SinkItem>, Self::SinkError> {
+        self.inner.start_send(item)
+    }
+
+    fn poll_complete(&mut self) -> Result<Async<()>, Self::SinkError> {
+        self.inner.poll_complete()
+    }
+}
+
+
+
+
+
+
+