Add RPC protocol to lh network behaviour.

beacon_node/libp2p/src/behaviour.rs

@@ -1,3 +1,4 @@
+use crate::rpc::{RPCMethod, RPCRequest, RPCResponse, Rpc, RpcEvent};
 use futures::prelude::*;
 use libp2p::{
     core::swarm::{NetworkBehaviourAction, NetworkBehaviourEventProcess},
@@ -15,7 +16,7 @@ pub struct Behaviour<TSubstream: AsyncRead + AsyncWrite> {
     gossipsub: Gossipsub<TSubstream>,
     // TODO: Add Kademlia for peer discovery
     /// The events generated by this behaviour to be consumed in the swarm poll.
-    // We use gossipsub events for now, generalise later.
+    serenity_rpc: Rpc<TSubstream>,
     #[behaviour(ignore)]
     events: Vec<BehaviourEvent>,
 }
@@ -37,10 +38,34 @@ impl<TSubstream: AsyncRead + AsyncWrite> NetworkBehaviourEventProcess<GossipsubE
     }
 }
 
+impl<TSubstream: AsyncRead + AsyncWrite> NetworkBehaviourEventProcess<RpcEvent>
+    for Behaviour<TSubstream>
+{
+    fn inject_event(&mut self, event: RpcEvent) {
+        match event {
+            RpcEvent::Request {
+                id,
+                method_id,
+                body,
+            } => self.events.push(BehaviourEvent::RPCRequest {
+                id,
+                method: RPCMethod::from(method_id),
+                body,
+            }),
+            RpcEvent::Response {
+                id,
+                method_id,
+                result,
+            } => self.events.push(BehaviourEvent::RPCResponse { id, result }),
+        }
+    }
+}
+
 impl<TSubstream: AsyncRead + AsyncWrite> Behaviour<TSubstream> {
     pub fn new(local_peer_id: PeerId, gs_config: GossipsubConfig) -> Self {
         Behaviour {
             gossipsub: Gossipsub::new(local_peer_id, gs_config),
+            serenity_rpc: Rpc::new(),
             events: Vec::new(),
         }
     }
@@ -70,6 +95,15 @@ impl<TSubstream: AsyncRead + AsyncWrite> Behaviour<TSubstream> {
 
 /// The types of events than can be obtained from polling the behaviour.
 pub enum BehaviourEvent {
+    RPCRequest {
+        id: u64,
+        method: RPCMethod,
+        body: RPCRequest,
+    },
+    RPCResponse {
+        id: u64,
+        result: RPCResponse,
+    },
     // TODO: This is a stub at the moment
     Message(String),
 }

beacon_node/libp2p/src/rpc/mod.rs

@@ -1,9 +1,9 @@
-mod handler;
-mod methods;
 /// RPC Protocol over libp2p.
 ///
 /// This is purpose built for Ethereum 2.0 serenity and the protocol listens on
 /// `/eth/serenity/rpc/1.0.0`
+mod handler;
+mod methods;
 mod protocol;
 
 use futures::prelude::*;
@@ -12,8 +12,8 @@ use libp2p::core::swarm::{
     ConnectedPoint, NetworkBehaviour, NetworkBehaviourAction, PollParameters,
 };
 use libp2p::{Multiaddr, PeerId};
-use methods::RPCRequest;
-use protocol::{RPCProtocol, RpcEvent};
+pub use methods::{RPCMethod, RPCRequest, RPCResponse};
+pub use protocol::{RPCProtocol, RpcEvent};
 use std::marker::PhantomData;
 use tokio::io::{AsyncRead, AsyncWrite};
 

beacon_node/libp2p/src/rpc/protocol.rs

@@ -158,6 +158,7 @@ impl Encodable for RpcEvent {
     }
 }
 
+#[derive(Debug)]
 pub enum DecodeError {
     ReadError(upgrade::ReadOneError),
     SSZDecodeError(ssz::DecodeError),