ping: Conform to the spec & exclude from connection keep-alive

src/protocol/libp2p/ping/config.rs

@@ -18,6 +18,7 @@
 // FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
 // DEALINGS IN THE SOFTWARE.
 
+use std::time::Duration;
 use crate::{
     codec::ProtocolCodec, protocol::libp2p::ping::PingEvent, types::protocol::ProtocolName,
     DEFAULT_CHANNEL_SIZE,
@@ -36,6 +37,8 @@ const PING_PAYLOAD_SIZE: usize = 32;
 /// Maximum PING failures.
 const MAX_FAILURES: usize = 3;
 
+pub const PING_INTERVAL: Duration = Duration::from_secs(15);
+
 /// Ping configuration.
 pub struct Config {
     /// Protocol name.
@@ -49,6 +52,8 @@ pub struct Config {
 
     /// TX channel for sending events to the user protocol.
     pub(crate) tx_event: Sender<PingEvent>,
+
+    pub(crate) ping_interval: Duration,
 }
 
 impl Config {
@@ -61,6 +66,7 @@ impl Config {
         (
             Self {
                 tx_event,
+                ping_interval: PING_INTERVAL,
                 max_failures: MAX_FAILURES,
                 protocol: ProtocolName::from(PROTOCOL_NAME),
                 codec: ProtocolCodec::Identity(PING_PAYLOAD_SIZE),
@@ -80,6 +86,7 @@ pub struct ConfigBuilder {
 
     /// Maximum failures before the peer is considered unreachable.
     max_failures: usize,
+    ping_interval: Duration,
 }
 
 impl Default for ConfigBuilder {
@@ -92,6 +99,7 @@ impl ConfigBuilder {
     /// Create new default [`Config`] which can be modified by the user.
     pub fn new() -> Self {
         Self {
+            ping_interval: PING_INTERVAL,
             max_failures: MAX_FAILURES,
             protocol: ProtocolName::from(PROTOCOL_NAME),
             codec: ProtocolCodec::Identity(PING_PAYLOAD_SIZE),
@@ -104,13 +112,19 @@ impl ConfigBuilder {
         self
     }
 
+    pub fn with_ping_interval(mut self, ping_interval: Duration) -> Self {
+        self.ping_interval = ping_interval;
+        self
+    }
+
     /// Build [`Config`].
     pub fn build(self) -> (Config, Box<dyn Stream<Item = PingEvent> + Send + Unpin>) {
         let (tx_event, rx_event) = channel(DEFAULT_CHANNEL_SIZE);
 
         (
             Config {
                 tx_event,
+                ping_interval: self.ping_interval,
                 max_failures: self.max_failures,
                 protocol: self.protocol,
                 codec: self.codec,

src/protocol/libp2p/ping/mod.rs

@@ -77,13 +77,16 @@ pub(crate) struct Ping {
 
     /// Pending inbound substreams.
     pending_inbound: FuturesUnordered<BoxFuture<'static, crate::Result<()>>>,
+
+    ping_interval: Duration,
 }
 
 impl Ping {
     /// Create new [`Ping`] protocol.
     pub fn new(service: TransportService, config: Config) -> Self {
         Self {
             service,
+            ping_interval: config.ping_interval,
             tx: config.tx_event,
             peers: HashSet::new(),
             pending_outbound: FuturesUnordered::new(),
@@ -96,7 +99,6 @@ impl Ping {
     fn on_connection_established(&mut self, peer: PeerId) -> crate::Result<()> {
         tracing::trace!(target: LOG_TARGET, ?peer, "connection established");
 
-        self.service.open_substream(peer)?;
         self.peers.insert(peer);
 
         Ok(())
@@ -166,12 +168,13 @@ impl Ping {
     /// Start [`Ping`] event loop.
     pub async fn run(mut self) {
         tracing::debug!(target: LOG_TARGET, "starting ping event loop");
+        let mut interval = tokio::time::interval(self.ping_interval);
 
         loop {
             tokio::select! {
                 event = self.service.next() => match event {
                     Some(TransportEvent::ConnectionEstablished { peer, .. }) => {
-                        let _ = self.on_connection_established(peer);
+                        self.on_connection_established(peer);
                     }
                     Some(TransportEvent::ConnectionClosed { peer }) => {
                         self.on_connection_closed(peer);
@@ -192,6 +195,19 @@ impl Ping {
                     Some(_) => {}
                     None => return,
                 },
+                _ = interval.tick() => {
+                    for peer in &self.peers {
+                        tracing::trace!(target: LOG_TARGET, ?peer, "sending ping");
+                        if let Err(error) = self.service.open_substream(*peer) {
+                            tracing::debug!(
+                                target: LOG_TARGET,
+                                ?peer,
+                                ?error,
+                                "failed to open substream for ping"
+                            );
+                        }
+                    }
+                }
                 _event = self.pending_inbound.next(), if !self.pending_inbound.is_empty() => {}
                 event = self.pending_outbound.next(), if !self.pending_outbound.is_empty() => {
                     match event {