feat(request-response): deprecate Config::set_connection_keep_alive

protocols/perf/src/bin/perf.rs

@@ -102,7 +102,7 @@ async fn main() -> Result<()> {
 }
 
 async fn server(server_address: SocketAddr) -> Result<()> {
-    let mut swarm = swarm::<libp2p_perf::server::Behaviour>().await?;
+    let mut swarm = swarm::<libp2p_perf::Behaviour<libp2p_perf::server::Behaviour>>().await?;
 
     swarm.listen_on(
         Multiaddr::empty()
@@ -135,9 +135,12 @@ async fn server(server_address: SocketAddr) -> Result<()> {
                     info!("Established connection to {:?} via {:?}", peer_id, endpoint);
                 }
                 SwarmEvent::ConnectionClosed { .. } => {}
-                SwarmEvent::Behaviour(()) => {
+                SwarmEvent::Behaviour(libp2p_perf::BehaviourEvent::Inner(())) => {
                     info!("Finished run",)
                 }
+                SwarmEvent::Behaviour(libp2p_perf::BehaviourEvent::KeepAlive(never)) => {
+                    void::unreachable(never)
+                }
                 e => panic!("{e:?}"),
             }
         }
@@ -298,7 +301,7 @@ async fn requests_per_second(server_address: Multiaddr) -> Result<()> {
     let to_receive = 1;
 
     for _ in 0..num {
-        swarm.behaviour_mut().perf(
+        swarm.behaviour_mut().inner.perf(
             server_peer_id,
             RunParams {
                 to_send,
@@ -312,10 +315,12 @@ async fn requests_per_second(server_address: Multiaddr) -> Result<()> {
 
     loop {
         match swarm.next().await.unwrap() {
-            SwarmEvent::Behaviour(libp2p_perf::client::Event {
-                id: _,
-                result: Ok(_),
-            }) => {
+            SwarmEvent::Behaviour(libp2p_perf::BehaviourEvent::Inner(
+                libp2p_perf::client::Event {
+                    id: _,
+                    result: Ok(_),
+                },
+            )) => {
                 finished += 1;
 
                 if finished == num {
@@ -425,7 +430,7 @@ async fn swarm<B: NetworkBehaviour + Default>() -> Result<Swarm<B>> {
 }
 
 async fn connect(
-    swarm: &mut Swarm<libp2p_perf::client::Behaviour>,
+    swarm: &mut Swarm<libp2p_perf::Behaviour<libp2p_perf::client::Behaviour>>,
     server_address: Multiaddr,
 ) -> Result<(PeerId, Duration)> {
     let start = Instant::now();
@@ -450,17 +455,17 @@ async fn connect(
 }
 
 async fn perf(
-    swarm: &mut Swarm<libp2p_perf::client::Behaviour>,
+    swarm: &mut Swarm<libp2p_perf::Behaviour<libp2p_perf::client::Behaviour>>,
     server_peer_id: PeerId,
     params: RunParams,
 ) -> Result<RunDuration> {
-    swarm.behaviour_mut().perf(server_peer_id, params)?;
+    swarm.behaviour_mut().inner.perf(server_peer_id, params)?;
 
     let duration = match swarm.next().await.unwrap() {
-        SwarmEvent::Behaviour(libp2p_perf::client::Event {
+        SwarmEvent::Behaviour(libp2p_perf::BehaviourEvent::Inner(libp2p_perf::client::Event {
             id: _,
             result: Ok(duration),
-        }) => duration,
+        })) => duration,
         e => panic!("{e:?}"),
     };
 

protocols/perf/src/client.rs

@@ -59,7 +59,6 @@ pub struct Behaviour {
 impl Default for Behaviour {
     fn default() -> Self {
         let mut req_resp_config = request_response::Config::default();
-        req_resp_config.set_connection_keep_alive(Duration::from_secs(60 * 5));
         req_resp_config.set_request_timeout(Duration::from_secs(60 * 5));
         Self {
             connected: Default::default(),

protocols/perf/src/lib.rs

@@ -27,14 +27,39 @@
 use std::fmt::Display;
 
 use instant::Duration;
-use libp2p_swarm::StreamProtocol;
+use libp2p_swarm::{keep_alive, StreamProtocol};
 
 pub mod client;
 mod protocol;
 pub mod server;
 
 pub const PROTOCOL_NAME: StreamProtocol = StreamProtocol::new("/perf/1.0.0");
 
+#[derive(libp2p_swarm::NetworkBehaviour, Default)]
+#[behaviour(prelude = "libp2p_swarm::derive_prelude")]
+pub struct Behaviour<T> {
+    pub inner: T,
+    keep_alive: keep_alive::Behaviour,
+}
+
+impl Behaviour<client::Behaviour> {
+    pub fn client() -> Self {
+        Self {
+            inner: client::Behaviour::default(),
+            keep_alive: keep_alive::Behaviour,
+        }
+    }
+}
+
+impl Behaviour<server::Behaviour> {
+    pub fn server() -> Self {
+        Self {
+            inner: server::Behaviour::default(),
+            keep_alive: keep_alive::Behaviour,
+        }
+    }
+}
+
 /// Parameters for a single run, i.e. one stream, sending and receiving data.
 ///
 /// Property names are from the perspective of the actor. E.g. `to_send` is the amount of data to

protocols/perf/src/server.rs

@@ -38,7 +38,6 @@ pub struct Behaviour {
 impl Default for Behaviour {
     fn default() -> Self {
         let mut req_resp_config = request_response::Config::default();
-        req_resp_config.set_connection_keep_alive(Duration::from_secs(60 * 5));
         req_resp_config.set_request_timeout(Duration::from_secs(60 * 5));
 
         Self {

protocols/request-response/src/handler.rs

@@ -27,7 +27,6 @@ use crate::handler::protocol::{RequestProtocol, ResponseProtocol};
 use crate::{RequestId, EMPTY_QUEUE_SHRINK_THRESHOLD};
 
 use futures::{channel::oneshot, future::BoxFuture, prelude::*, stream::FuturesUnordered};
-use instant::Instant;
 use libp2p_swarm::handler::{
     ConnectionEvent, DialUpgradeError, FullyNegotiatedInbound, FullyNegotiatedOutbound,
     ListenUpgradeError,
@@ -57,14 +56,9 @@ where
     inbound_protocols: SmallVec<[TCodec::Protocol; 2]>,
     /// The request/response message codec.
     codec: TCodec,
-    /// The keep-alive timeout of idle connections. A connection is considered
-    /// idle if there are no outbound substreams.
-    keep_alive_timeout: Duration,
     /// The timeout for inbound and outbound substreams (i.e. request
     /// and response processing).
     substream_timeout: Duration,
-    /// The current connection keep-alive.
-    keep_alive: KeepAlive,
     /// Queue of events to emit in `poll()`.
     pending_events: VecDeque<Event<TCodec>>,
     /// Outbound upgrades waiting to be emitted as an `OutboundSubstreamRequest`.
@@ -92,15 +86,12 @@ where
     pub(super) fn new(
         inbound_protocols: SmallVec<[TCodec::Protocol; 2]>,
         codec: TCodec,
-        keep_alive_timeout: Duration,
         substream_timeout: Duration,
         inbound_request_id: Arc<AtomicU64>,
     ) -> Self {
         Self {
             inbound_protocols,
             codec,
-            keep_alive: KeepAlive::Yes,
-            keep_alive_timeout,
             substream_timeout,
             outbound: VecDeque::new(),
             inbound: FuturesUnordered::new(),
@@ -280,12 +271,19 @@ where
     }
 
     fn on_behaviour_event(&mut self, request: Self::FromBehaviour) {
-        self.keep_alive = KeepAlive::Yes;
         self.outbound.push_back(request);
     }
 
     fn connection_keep_alive(&self) -> KeepAlive {
-        self.keep_alive
+        if !self.outbound.is_empty() {
+            return KeepAlive::Yes;
+        }
+
+        if !self.inbound.is_empty() {
+            return KeepAlive::Yes;
+        }
+
+        KeepAlive::No
     }
 
     fn poll(
@@ -306,7 +304,6 @@ where
             match result {
                 Ok(((id, rq), rs_sender)) => {
                     // We received an inbound request.
-                    self.keep_alive = KeepAlive::Yes;
                     return Poll::Ready(ConnectionHandlerEvent::NotifyBehaviour(Event::Request {
                         request_id: id,
                         request: rq,
@@ -336,14 +333,6 @@ where
             self.outbound.shrink_to_fit();
         }
 
-        if self.inbound.is_empty() && self.keep_alive.is_yes() {
-            // No new inbound or outbound requests. However, we may just have
-            // started the latest inbound or outbound upgrade(s), so make sure
-            // the keep-alive timeout is preceded by the substream timeout.
-            let until = Instant::now() + self.substream_timeout + self.keep_alive_timeout;
-            self.keep_alive = KeepAlive::Until(until);
-        }
-
         Poll::Pending
     }
 

protocols/request-response/src/lib.rs

@@ -84,8 +84,8 @@ use libp2p_identity::PeerId;
 use libp2p_swarm::{
     behaviour::{AddressChange, ConnectionClosed, ConnectionEstablished, DialFailure, FromSwarm},
     dial_opts::DialOpts,
-    ConnectionDenied, ConnectionId, NetworkBehaviour, NotifyHandler, PollParameters, THandler,
-    THandlerInEvent, THandlerOutEvent, ToSwarm,
+    ConnectionDenied, ConnectionHandler, ConnectionId, NetworkBehaviour, NotifyHandler,
+    PollParameters, THandler, THandlerInEvent, THandlerOutEvent, ToSwarm,
 };
 use smallvec::SmallVec;
 use std::{
@@ -284,25 +284,17 @@ impl fmt::Display for RequestId {
 #[derive(Debug, Clone)]
 pub struct Config {
     request_timeout: Duration,
-    connection_keep_alive: Duration,
 }
 
 impl Default for Config {
     fn default() -> Self {
         Self {
-            connection_keep_alive: Duration::from_secs(10),
             request_timeout: Duration::from_secs(10),
         }
     }
 }
 
 impl Config {
-    /// Sets the keep-alive timeout of idle connections.
-    pub fn set_connection_keep_alive(&mut self, v: Duration) -> &mut Self {
-        self.connection_keep_alive = v;
-        self
-    }
-
     /// Sets the timeout for inbound and outbound requests.
     pub fn set_request_timeout(&mut self, v: Duration) -> &mut Self {
         self.request_timeout = v;
@@ -713,17 +705,24 @@ where
     fn handle_established_inbound_connection(
         &mut self,
         _: ConnectionId,
-        _: PeerId,
+        peer: PeerId,
         _: &Multiaddr,
         _: &Multiaddr,
     ) -> Result<THandler<Self>, ConnectionDenied> {
-        Ok(Handler::new(
+        let mut handler = Handler::new(
             self.inbound_protocols.clone(),
             self.codec.clone(),
-            self.config.connection_keep_alive,
             self.config.request_timeout,
             self.next_inbound_id.clone(),
-        ))
+        );
+
+        if let Some(pending_requests) = self.pending_outbound_requests.remove(&peer) {
+            for request in pending_requests {
+                handler.on_behaviour_event(request);
+            }
+        }
+
+        Ok(handler)
     }
 
     fn handle_pending_outbound_connection(
@@ -752,17 +751,24 @@ where
     fn handle_established_outbound_connection(
         &mut self,
         _: ConnectionId,
-        _: PeerId,
+        peer: PeerId,
         _: &Multiaddr,
         _: Endpoint,
     ) -> Result<THandler<Self>, ConnectionDenied> {
-        Ok(Handler::new(
+        let mut handler = Handler::new(
             self.inbound_protocols.clone(),
             self.codec.clone(),
-            self.config.connection_keep_alive,
             self.config.request_timeout,
             self.next_inbound_id.clone(),
-        ))
+        );
+
+        if let Some(pending_requests) = self.pending_outbound_requests.remove(&peer) {
+            for request in pending_requests {
+                handler.on_behaviour_event(request);
+            }
+        }
+
+        Ok(handler)
     }
 
     fn on_swarm_event(&mut self, event: FromSwarm<Self::ConnectionHandler>) {