Port websockets, timer and notifier to stable futures (#1035)

* Fix lcli

* Port timer to stable futures

* Fix timer

* Port websocket_server to stable futures

* Port notifier to stable futures

* Add TODOS

* Port remote_beacon_node to stable futures

beacon_node/timer/src/lib.rs

@@ -3,20 +3,22 @@
 //! This service allows task execution on the beacon node for various functionality.
 
 use beacon_chain::{BeaconChain, BeaconChainTypes};
+use futures::stream::StreamExt;
 use futures::{future, prelude::*};
-use slog::error;
 use slot_clock::SlotClock;
 use std::sync::Arc;
-use std::time::{Duration, Instant};
-use tokio::runtime::TaskExecutor;
-use tokio::timer::Interval;
+use std::time::Duration;
+use tokio::runtime::Handle;
+use tokio::time::{interval_at, Instant};
 
 /// Spawns a timer service which periodically executes tasks for the beacon chain
+/// TODO: We might not need a `Handle` to the runtime since this function should be
+/// called from the context of a runtime and we can simply spawn using task::spawn.
+/// Check for issues without the Handle.
 pub fn spawn<T: BeaconChainTypes>(
-    executor: &TaskExecutor,
+    handle: &Handle,
     beacon_chain: Arc<BeaconChain<T>>,
     milliseconds_per_slot: u64,
-    log: slog::Logger,
 ) -> Result<tokio::sync::oneshot::Sender<()>, &'static str> {
     let (exit_signal, exit) = tokio::sync::oneshot::channel();
 
@@ -26,25 +28,15 @@ pub fn spawn<T: BeaconChainTypes>(
             .duration_to_next_slot()
             .ok_or_else(|| "slot_notifier unable to determine time to next slot")?;
 
-    let timer_future = Interval::new(start_instant, Duration::from_millis(milliseconds_per_slot))
-        .map_err(move |e| {
-            error!(
-                log,
-                "Beacon chain timer failed";
-                "error" => format!("{:?}", e)
-            )
-        })
+    // Warning: `interval_at` panics if `milliseconds_per_slot` = 0.
+    let timer_future = interval_at(start_instant, Duration::from_millis(milliseconds_per_slot))
         .for_each(move |_| {
             beacon_chain.per_slot_task();
-            future::ok(())
+            future::ready(())
         });
 
-    executor.spawn(
-        exit.map_err(|_| ())
-            .select(timer_future)
-            .map(|_| ())
-            .map_err(|_| ()),
-    );
+    let future = futures::future::select(timer_future, exit.map_err(|_| ()).map(|_| ()));
+    handle.spawn(future);
 
     Ok(exit_signal)
 }