Add network channel into message handler

beacon_node/network/src/message_handler.rs

@@ -1,6 +1,7 @@
 use crate::beacon_chain::BeaconChain;
 use crate::error;
 use crate::messages::NodeMessage;
+use crate::service::NetworkMessage;
 use crossbeam_channel::{unbounded as channel, Sender};
 use futures::future;
 use futures::prelude::*;
@@ -22,6 +23,8 @@ pub struct MessageHandler {
     chain: Arc<BeaconChain>,
     /// The syncing framework.
     sync: SimpleSync,
+    /// The network channel to relay messages to the Network service.
+    network_send: crossbeam_channel::Sender<NetworkMessage>,
     /// A mapping of peers we have sent a HELLO rpc request to.
     hello_requests: HashMap<PeerId, Instant>,
     /// The `MessageHandler` logger.
@@ -45,6 +48,7 @@ impl MessageHandler {
     /// Initializes and runs the MessageHandler.
     pub fn new(
         beacon_chain: Arc<BeaconChain>,
+        network_send: crossbeam_channel::Sender<NetworkMessage>,
         executor: &tokio::runtime::TaskExecutor,
         log: slog::Logger,
     ) -> error::Result<Sender<HandlerMessage>> {
@@ -62,6 +66,7 @@ impl MessageHandler {
         let mut handler = MessageHandler {
             chain: beacon_chain.clone(),
             sync,
+            network_send,
             hello_requests: HashMap::new(),
             log: log.clone(),
         };
@@ -81,12 +86,19 @@ impl MessageHandler {
 
     fn handle_message(&mut self, message: HandlerMessage) {
         match message {
-            HandlerMessage::PeerDialed(peer_id) => self.send_hello(peer_id),
+            HandlerMessage::PeerDialed(peer_id) => {
+                // register RPC request
+                self.hello_requests.insert(peer_id.clone(), Instant::now());
+                self.send_hello(peer_id);
+            }
             //TODO: Handle all messages
             _ => {}
         }
     }
 
     /// Sends a HELLO RPC request to a newly connected peer.
-    fn send_hello(&self, peer_id: PeerId) {}
+    fn send_hello(&self, peer_id: PeerId) {
+        // send the hello request to the network
+        //sync.hello()
+    }
 }

beacon_node/network/src/messages.rs

@@ -2,7 +2,10 @@ use libp2p::PeerId;
 use libp2p::{HelloMessage, RPCEvent};
 use types::{Hash256, Slot};
 
+//TODO: This module can be entirely replaced in the RPC rewrite
+
 /// Messages between nodes across the network.
+//TODO: Remove this in the RPC rewrite
 #[derive(Debug, Clone)]
 pub enum NodeMessage {
     RPC(RPCEvent),
@@ -10,11 +13,3 @@ pub enum NodeMessage {
     // TODO: only for testing - remove
     Message(String),
 }
-
-/// Types of messages that the network service can receive.
-#[derive(Debug, Clone)]
-pub enum NetworkMessage {
-    /// Send a message to libp2p service.
-    //TODO: Define typing for messages across the wire
-    Send(PeerId, NodeMessage),
-}

beacon_node/network/src/service.rs

@@ -1,7 +1,7 @@
 use crate::beacon_chain::BeaconChain;
 use crate::error;
 use crate::message_handler::{HandlerMessage, MessageHandler};
-use crate::messages::{NetworkMessage, NodeMessage};
+use crate::messages::NodeMessage;
 use crate::NetworkConfig;
 use crossbeam_channel::{unbounded as channel, Sender, TryRecvError};
 use futures::prelude::*;
@@ -29,18 +29,29 @@ impl Service {
         executor: &TaskExecutor,
         log: slog::Logger,
     ) -> error::Result<(Arc<Self>, Sender<NetworkMessage>)> {
+        // build the network channel
+        let (network_send, network_recv) = channel::<NetworkMessage>();
         // launch message handler thread
         let message_handler_log = log.new(o!("Service" => "MessageHandler"));
-        let message_handler_send =
+        let message_handler_send = MessageHandler::new(
-            MessageHandler::new(beacon_chain, executor, message_handler_log)?;
+            beacon_chain,
+            network_send.clone(),
+            executor,
+            message_handler_log,
+        )?;
 
         // launch libp2p service
         let libp2p_log = log.new(o!("Service" => "Libp2p"));
         let libp2p_service = LibP2PService::new(config.clone(), libp2p_log)?;
 
         // TODO: Spawn thread to handle libp2p messages and pass to message handler thread.
-        let (network_send, libp2p_exit) =
+        let libp2p_exit = spawn_service(
-            spawn_service(libp2p_service, message_handler_send, executor, log)?;
+            libp2p_service,
+            network_recv,
+            message_handler_send,
+            executor,
+            log,
+        )?;
         let network = Service {
             libp2p_exit,
             network_send: network_send.clone(),
@@ -59,15 +70,12 @@ impl Service {
 
 fn spawn_service(
     libp2p_service: LibP2PService,
+    network_recv: crossbeam_channel::Receiver<NetworkMessage>,
     message_handler_send: crossbeam_channel::Sender<HandlerMessage>,
     executor: &TaskExecutor,
     log: slog::Logger,
-) -> error::Result<(
+) -> error::Result<oneshot::Sender<()>> {
-    crossbeam_channel::Sender<NetworkMessage>,
-    oneshot::Sender<()>,
-)> {
     let (network_exit, exit_rx) = oneshot::channel();
-    let (network_send, network_recv) = channel::<NetworkMessage>();
 
     // spawn on the current executor
     executor.spawn(
@@ -85,7 +93,7 @@ fn spawn_service(
         }),
     );
 
-    Ok((network_send, network_exit))
+    Ok(network_exit)
 }
 
 fn network_service(
@@ -148,3 +156,11 @@ fn network_service(
         Ok(Async::NotReady)
     })
 }
+
+/// Types of messages that the network service can receive.
+#[derive(Debug, Clone)]
+pub enum NetworkMessage {
+    /// Send a message to libp2p service.
+    //TODO: Define typing for messages across the wire
+    Send(PeerId, NodeMessage),
+}