From a4cfe682726a035b8f4dce4d395074f98205e8f6 Mon Sep 17 00:00:00 2001
From: Age Manning <Age@AgeManning.com>
Date: Fri, 22 Mar 2019 16:46:52 +1100
Subject: [PATCH] Thread beacon node RPC server

---
 beacon_node/client/src/lib.rs  | 23 +++++++++++-----------
 beacon_node/rpc/Cargo.toml     |  2 ++
 beacon_node/rpc/src/lib.rs     | 35 +++++++++++++++++++++++++---------
 validator_client/src/config.rs |  2 +-
 4 files changed, 41 insertions(+), 21 deletions(-)

diff --git a/beacon_node/client/src/lib.rs b/beacon_node/client/src/lib.rs
index d8da18cae..a033da87b 100644
--- a/beacon_node/client/src/lib.rs
+++ b/beacon_node/client/src/lib.rs
@@ -24,12 +24,8 @@ pub struct Client<T: ClientTypes> {
     beacon_chain: Arc<BeaconChain<T::DB, T::SlotClock, T::ForkChoice>>,
     /// Reference to the network service.
     pub network: Arc<NetworkService>,
-    /// Future to stop and begin shutdown of the Client.
-    //TODO: Decide best way to handle shutdown
-    pub exit: exit_future::Exit,
-    /// The sending future to call to terminate the Client.
-    //TODO: Decide best way to handle shutdown
-    pub exit_signal: Signal,
+    /// Signal to terminate the RPC server.
+    pub rpc_exit_signal: Option<Signal>,
     /// The clients logger.
     log: slog::Logger,
     /// Marker to pin the beacon chain generics.
@@ -43,8 +39,6 @@ impl<TClientType: ClientTypes> Client<TClientType> {
         log: slog::Logger,
         executor: &TaskExecutor,
     ) -> error::Result<Self> {
-        let (exit_signal, exit) = exit_future::signal();
-
         // generate a beacon chain
         let beacon_chain = TClientType::initialise_beacon_chain(&config);
 
@@ -59,16 +53,23 @@ impl<TClientType: ClientTypes> Client<TClientType> {
             network_logger,
         )?;
 
+        let mut rpc_exit_signal = None;
         // spawn the RPC server
         if config.rpc_conf.enabled {
-            rpc::start_server(&config.rpc_conf, beacon_chain.clone(), &log);
+            rpc_exit_signal = Some(rpc::start_server(
+                &config.rpc_conf,
+                executor,
+                beacon_chain.clone(),
+                &log,
+            ));
         }
 
+        println!("Here");
+
         Ok(Client {
             config,
             beacon_chain,
-            exit,
-            exit_signal,
+            rpc_exit_signal,
             log,
             network,
             phantom: PhantomData,
diff --git a/beacon_node/rpc/Cargo.toml b/beacon_node/rpc/Cargo.toml
index acb68972c..d405982db 100644
--- a/beacon_node/rpc/Cargo.toml
+++ b/beacon_node/rpc/Cargo.toml
@@ -21,3 +21,5 @@ futures = "0.1.23"
 slog = "^2.2.3"
 slog-term = "^2.4.0"
 slog-async = "^2.3.0"
+tokio = "0.1.17"
+exit-future = "0.1.4"
diff --git a/beacon_node/rpc/src/lib.rs b/beacon_node/rpc/src/lib.rs
index 4565abf7a..02e34781c 100644
--- a/beacon_node/rpc/src/lib.rs
+++ b/beacon_node/rpc/src/lib.rs
@@ -9,24 +9,28 @@ use self::beacon_chain::BeaconChain;
 use self::beacon_node::BeaconNodeServiceInstance;
 use self::validator::ValidatorServiceInstance;
 pub use config::Config as RPCConfig;
+use futures::{future, Future};
 use grpcio::{Environment, Server, ServerBuilder};
 use protos::services_grpc::{
     create_beacon_block_service, create_beacon_node_service, create_validator_service,
 };
+use slog::{info, o, warn};
 use std::sync::Arc;
-
-use slog::{info, o};
+use tokio::runtime::TaskExecutor;
 
 pub fn start_server(
     config: &RPCConfig,
+    executor: &TaskExecutor,
     beacon_chain: Arc<BeaconChain>,
     log: &slog::Logger,
-) -> Server {
+) -> exit_future::Signal {
     let log = log.new(o!("Service"=>"RPC"));
     let env = Arc::new(Environment::new(1));
 
-    // build the individual rpc services
+    // build a channel to kill the rpc server
+    let (rpc_exit_signal, rpc_exit) = exit_future::signal();
 
+    // build the individual rpc services
     let beacon_node_service = {
         let instance = BeaconNodeServiceInstance {
             chain: beacon_chain.clone(),
@@ -50,9 +54,22 @@ pub fn start_server(
         .bind(config.listen_address.to_string(), config.port)
         .build()
         .unwrap();
-    server.start();
-    for &(ref host, port) in server.bind_addrs() {
-        info!(log, "gRPC listening on {}:{}", host, port);
-    }
-    server
+
+    let spawn_rpc = {
+        server.start();
+        for &(ref host, port) in server.bind_addrs() {
+            info!(log, "gRPC listening on {}:{}", host, port);
+        }
+        rpc_exit.and_then(move |_| {
+            info!(log, "RPC Server shutting down");
+            server
+                .shutdown()
+                .wait()
+                .map(|_| ())
+                .map_err(|e| warn!(log, "RPC server failed to shutdown: {:?}", e))?;
+            Ok(())
+        })
+    };
+    executor.spawn(spawn_rpc);
+    rpc_exit_signal
 }
diff --git a/validator_client/src/config.rs b/validator_client/src/config.rs
index 68405ed2f..49e0a506f 100644
--- a/validator_client/src/config.rs
+++ b/validator_client/src/config.rs
@@ -21,7 +21,7 @@ impl ClientConfig {
         };
         fs::create_dir_all(&data_dir)
             .unwrap_or_else(|_| panic!("Unable to create {:?}", &data_dir));
-        let server = "localhost:50051".to_string();
+        let server = "localhost:5051".to_string();
         let spec = ChainSpec::foundation();
         Self {
             data_dir,