Add a Ping behaviour instead of PeriodicPing and PingListen (#815)

* Add a Ping behaviour instead of PeriodicPing and PingListen

* Fix tests

protocols/ping/src/lib.rs

@@ -81,12 +81,99 @@
 //! ```
 //!
 
-pub use self::dial_layer::PeriodicPing;
-pub use self::listen_layer::PingListen;
-
 pub mod dial_handler;
 pub mod listen_handler;
 pub mod protocol;
 
-mod dial_layer;
-mod listen_layer;
+use futures::prelude::*;
+use libp2p_core::either::EitherOutput;
+use libp2p_core::swarm::{ConnectedPoint, NetworkBehaviour, NetworkBehaviourAction, PollParameters};
+use libp2p_core::{protocols_handler::ProtocolsHandler, protocols_handler::ProtocolsHandlerSelect, PeerId};
+use std::{marker::PhantomData, time::Duration};
+use tokio_io::{AsyncRead, AsyncWrite};
+
+/// Network behaviour that handles receiving pings sent by other nodes and periodically pings the
+/// nodes we connect to.
+pub struct Ping<TSubstream> {
+    /// Marker to pin the generics.
+    marker: PhantomData<TSubstream>,
+    /// Queue of events to report to the user.
+    events: Vec<PingEvent>,
+}
+
+/// Event generated by the `Ping` behaviour.
+pub enum PingEvent {
+    /// We have successfully pinged a peer we are connected to.
+    PingSuccess {
+        /// Id of the peer that we pinged.
+        peer: PeerId,
+        /// Time elapsed between when we sent the ping and when we received the response.
+        time: Duration,
+    }
+}
+
+impl<TSubstream> Ping<TSubstream> {
+    /// Creates a `Ping`.
+    pub fn new() -> Self {
+        Ping {
+            marker: PhantomData,
+            events: Vec::new(),
+        }
+    }
+}
+
+impl<TSubstream> Default for Ping<TSubstream> {
+    #[inline]
+    fn default() -> Self {
+        Ping::new()
+    }
+}
+
+impl<TSubstream, TTopology> NetworkBehaviour<TTopology> for Ping<TSubstream>
+where
+    TSubstream: AsyncRead + AsyncWrite,
+{
+    type ProtocolsHandler = ProtocolsHandlerSelect<listen_handler::PingListenHandler<TSubstream>, dial_handler::PeriodicPingHandler<TSubstream>>;
+    type OutEvent = PingEvent;
+
+    fn new_handler(&mut self) -> Self::ProtocolsHandler {
+        listen_handler::PingListenHandler::new()
+            .select(dial_handler::PeriodicPingHandler::new())
+    }
+
+    fn inject_connected(&mut self, _: PeerId, _: ConnectedPoint) {}
+
+    fn inject_disconnected(&mut self, _: &PeerId, _: ConnectedPoint) {}
+
+    fn inject_node_event(
+        &mut self,
+        source: PeerId,
+        event: <Self::ProtocolsHandler as ProtocolsHandler>::OutEvent,
+    ) {
+        match event {
+            EitherOutput::Second(dial_handler::OutEvent::PingSuccess(time)) => {
+                self.events.push(PingEvent::PingSuccess {
+                    peer: source,
+                    time,
+                })
+            },
+            _ => ()
+        }
+    }
+
+    fn poll(
+        &mut self,
+        _: &mut PollParameters<TTopology>,
+    ) -> Async<
+        NetworkBehaviourAction<
+            <Self::ProtocolsHandler as ProtocolsHandler>::InEvent,
+            Self::OutEvent,
+        >,
+    > {
+        if !self.events.is_empty() {
+            return Async::Ready(NetworkBehaviourAction::GenerateEvent(self.events.remove(0)));
+        }
+
+        Async::NotReady
+    }
+}