Upgrade futures-bounded to 0.3.0

Author: teor2345

Cargo.lock

@@ -121,7 +121,7 @@ libp2p-yamux = { version = "0.47.0", path = "muxers/yamux" }
 asynchronous-codec = { version = "0.7.0" }
 env_logger = "0.11"
 futures = "0.3.30"
-futures-bounded = { version = "0.2.4" }
+futures-bounded = { version = "0.3.0", features = ["futures-timer"] }
 futures-rustls = { version = "0.26.0", default-features = false }
 getrandom = "0.2"
 if-watch = "3.2.1"

Cargo.toml

@@ -6,7 +6,7 @@ use std::{
 };
 
 use futures::channel::oneshot;
-use futures_bounded::StreamSet;
+use futures_bounded::{Delay, StreamSet};
 use libp2p_core::upgrade::{DeniedUpgrade, ReadyUpgrade};
 use libp2p_swarm::{
     handler::{ConnectionEvent, FullyNegotiatedInbound, ListenUpgradeError},
@@ -22,7 +22,7 @@ pub struct Handler {
 impl Handler {
     pub(crate) fn new() -> Self {
         Self {
-            inbound: StreamSet::new(Duration::from_secs(5), 2),
+            inbound: StreamSet::new(|| Delay::futures_timer(Duration::from_secs(5)), 2),
         }
     }
 }

protocols/autonat/src/v2/client/handler/dial_back.rs

@@ -8,7 +8,7 @@ use std::{
 };
 
 use futures::{channel::oneshot, AsyncWrite};
-use futures_bounded::FuturesMap;
+use futures_bounded::{Delay, FuturesMap};
 use libp2p_core::{
     upgrade::{DeniedUpgrade, ReadyUpgrade},
     Multiaddr,
@@ -91,7 +91,7 @@ impl Handler {
     pub(crate) fn new() -> Self {
         Self {
             queued_events: VecDeque::new(),
-            outbound: FuturesMap::new(Duration::from_secs(10), 10),
+            outbound: FuturesMap::new(|| Delay::futures_timer(Duration::from_secs(10)), 10),
             queued_streams: VecDeque::default(),
         }
     }

protocols/autonat/src/v2/client/handler/dial_request.rs

@@ -6,7 +6,7 @@ use std::{
 };
 
 use futures::{AsyncRead, AsyncWrite};
-use futures_bounded::FuturesSet;
+use futures_bounded::{Delay, FuturesSet};
 use libp2p_core::upgrade::{DeniedUpgrade, ReadyUpgrade};
 use libp2p_swarm::{
     handler::{ConnectionEvent, DialUpgradeError, FullyNegotiatedOutbound},
@@ -33,7 +33,7 @@ impl Handler {
         Self {
             pending_nonce: Some(cmd),
             requested_substream_nonce: None,
-            outbound: FuturesSet::new(Duration::from_secs(10), 5),
+            outbound: FuturesSet::new(|| Delay::futures_timer(Duration::from_secs(10)), 5),
         }
     }
 }

protocols/autonat/src/v2/server/handler/dial_back.rs

@@ -10,7 +10,7 @@ use futures::{
     channel::{mpsc, oneshot},
     AsyncRead, AsyncWrite, SinkExt, StreamExt,
 };
-use futures_bounded::FuturesSet;
+use futures_bounded::{Delay, FuturesSet};
 use libp2p_core::{
     upgrade::{DeniedUpgrade, ReadyUpgrade},
     Multiaddr,
@@ -64,7 +64,7 @@ where
             observed_multiaddr,
             dial_back_cmd_sender,
             dial_back_cmd_receiver,
-            inbound: FuturesSet::new(Duration::from_secs(10), 10),
+            inbound: FuturesSet::new(|| Delay::futures_timer(Duration::from_secs(10)), 10),
             rng,
         }
     }

protocols/autonat/src/v2/server/handler/dial_request.rs

@@ -29,6 +29,7 @@ use std::{
 
 use either::Either;
 use futures::future;
+use futures_bounded::Delay;
 use libp2p_core::{
     multiaddr::Multiaddr,
     upgrade::{DeniedUpgrade, ReadyUpgrade},
@@ -87,8 +88,14 @@ impl Handler {
         Self {
             endpoint,
             queued_events: Default::default(),
-            inbound_stream: futures_bounded::FuturesSet::new(Duration::from_secs(10), 1),
-            outbound_stream: futures_bounded::FuturesSet::new(Duration::from_secs(10), 1),
+            inbound_stream: futures_bounded::FuturesSet::new(
+                || Delay::futures_timer(Duration::from_secs(10)),
+                1,
+            ),
+            outbound_stream: futures_bounded::FuturesSet::new(
+                || Delay::futures_timer(Duration::from_secs(10)),
+                1,
+            ),
             holepunch_candidates,
             attempts: 0,
         }

protocols/dcutr/src/handler/relayed.rs

@@ -139,7 +139,7 @@ impl Handler {
             remote_peer_id,
             events: SmallVec::new(),
             active_streams: futures_bounded::FuturesSet::new(
-                STREAM_TIMEOUT,
+                || futures_bounded::Delay::futures_timer(STREAM_TIMEOUT),
                 MAX_CONCURRENT_STREAMS_PER_CONNECTION,
             ),
             trigger_next_identify: Delay::new(Duration::ZERO),

protocols/identify/src/handler.rs

@@ -28,6 +28,7 @@ use std::{
 
 use either::Either;
 use futures::{channel::oneshot, prelude::*, stream::SelectAll};
+use futures_bounded::Delay;
 use libp2p_core::{upgrade, ConnectedPoint};
 use libp2p_identity::PeerId;
 use libp2p_swarm::{
@@ -463,7 +464,7 @@ impl Handler {
             next_connec_unique_id: UniqueConnecId(0),
             inbound_substreams: Default::default(),
             outbound_substreams: futures_bounded::FuturesTupleSet::new(
-                substreams_timeout,
+                move || Delay::futures_timer(substreams_timeout),
                 MAX_NUM_STREAMS,
             ),
             pending_streams: Default::default(),

protocols/kad/src/handler.rs

@@ -24,6 +24,7 @@ use std::{
 };
 
 use futures::FutureExt;
+use futures_bounded::Delay;
 use libp2p_core::upgrade::{DeniedUpgrade, ReadyUpgrade};
 use libp2p_swarm::{
     handler::{
@@ -49,7 +50,7 @@ impl Handler {
     pub fn new() -> Self {
         Self {
             inbound: futures_bounded::FuturesSet::new(
-                crate::RUN_TIMEOUT,
+                || Delay::futures_timer(crate::RUN_TIMEOUT),
                 crate::MAX_PARALLEL_RUNS_PER_CONNECTION,
             ),
         }