transport: Fix waking up on filtered events from poll_next

src/lib.rs

@@ -80,6 +80,7 @@ pub mod yamux;
 
 mod bandwidth;
 mod multistream_select;
+mod utils;
 
 #[cfg(test)]
 mod mock;

src/protocol/libp2p/kademlia/executor.rs

@@ -19,9 +19,8 @@
 // DEALINGS IN THE SOFTWARE.
 
 use crate::{
-    protocol::libp2p::kademlia::{futures_stream::FuturesStream, query::QueryId},
-    substream::Substream,
-    PeerId,
+    protocol::libp2p::kademlia::query::QueryId, substream::Substream,
+    utils::futures_stream::FuturesStream, PeerId,
 };
 
 use bytes::{Bytes, BytesMut};

src/protocol/libp2p/kademlia/mod.rs

@@ -71,7 +71,6 @@ const PARALLELISM_FACTOR: usize = 3;
 mod bucket;
 mod config;
 mod executor;
-mod futures_stream;
 mod handle;
 mod message;
 mod query;

src/protocol/libp2p/kademlia/store.rs

@@ -23,10 +23,10 @@
 use crate::{
     protocol::libp2p::kademlia::{
         config::{DEFAULT_PROVIDER_REFRESH_INTERVAL, DEFAULT_PROVIDER_TTL},
-        futures_stream::FuturesStream,
         record::{ContentProvider, Key, ProviderRecord, Record},
         types::Key as KademliaKey,
     },
+    utils::futures_stream::FuturesStream,
     PeerId,
 };
 

src/transport/tcp/mod.rs

@@ -34,6 +34,7 @@ use crate::{
         Transport, TransportBuilder, TransportEvent,
     },
     types::ConnectionId,
+    utils::futures_stream::FuturesStream,
 };
 
 use futures::{
@@ -111,12 +112,11 @@ pub(crate) struct TcpTransport {
     pending_inbound_connections: HashMap<ConnectionId, PendingInboundConnection>,
 
     /// Pending opening connections.
-    pending_connections: FuturesUnordered<
-        BoxFuture<'static, Result<NegotiatedConnection, (ConnectionId, DialError)>>,
-    >,
+    pending_connections:
+        FuturesStream<BoxFuture<'static, Result<NegotiatedConnection, (ConnectionId, DialError)>>>,
 
     /// Pending raw, unnegotiated connections.
-    pending_raw_connections: FuturesUnordered<BoxFuture<'static, RawConnectionResult>>,
+    pending_raw_connections: FuturesStream<BoxFuture<'static, RawConnectionResult>>,
 
     /// Opened raw connection, waiting for approval/rejection from `TransportManager`.
     opened_raw: HashMap<ConnectionId, (TcpStream, Multiaddr)>,
@@ -295,8 +295,8 @@ impl TransportBuilder for TcpTransport {
                 pending_open: HashMap::new(),
                 pending_dials: HashMap::new(),
                 pending_inbound_connections: HashMap::new(),
-                pending_connections: FuturesUnordered::new(),
-                pending_raw_connections: FuturesUnordered::new(),
+                pending_connections: FuturesStream::new(),
+                pending_raw_connections: FuturesStream::new(),
                 cancel_futures: HashMap::new(),
             },
             listen_addresses,

src/transport/websocket/mod.rs

@@ -33,6 +33,7 @@ use crate::{
         Transport, TransportBuilder, TransportEvent,
     },
     types::ConnectionId,
+    utils::futures_stream::FuturesStream,
     DialError, PeerId,
 };
 
@@ -115,12 +116,11 @@ pub(crate) struct WebSocketTransport {
     pending_inbound_connections: HashMap<ConnectionId, PendingInboundConnection>,
 
     /// Pending connections.
-    pending_connections: FuturesUnordered<
-        BoxFuture<'static, Result<NegotiatedConnection, (ConnectionId, DialError)>>,
-    >,
+    pending_connections:
+        FuturesStream<BoxFuture<'static, Result<NegotiatedConnection, (ConnectionId, DialError)>>>,
 
     /// Pending raw, unnegotiated connections.
-    pending_raw_connections: FuturesUnordered<BoxFuture<'static, RawConnectionResult>>,
+    pending_raw_connections: FuturesStream<BoxFuture<'static, RawConnectionResult>>,
 
     /// Opened raw connection, waiting for approval/rejection from `TransportManager`.
     opened_raw: HashMap<ConnectionId, (WebSocketStream<MaybeTlsStream<TcpStream>>, Multiaddr)>,
@@ -325,8 +325,8 @@ impl TransportBuilder for WebSocketTransport {
                 pending_open: HashMap::new(),
                 pending_dials: HashMap::new(),
                 pending_inbound_connections: HashMap::new(),
-                pending_connections: FuturesUnordered::new(),
-                pending_raw_connections: FuturesUnordered::new(),
+                pending_connections: FuturesStream::new(),
+                pending_raw_connections: FuturesStream::new(),
                 cancel_futures: HashMap::new(),
             },
             listen_addresses,

src/protocol/libp2p/kademlia/futures_stream.rs → src/utils/futures_stream.rs

@@ -44,7 +44,7 @@ impl<F> FuturesStream<F> {
         }
     }
 
-    /// Number of futeres in the stream.
+    /// Number of futures in the stream.
     #[cfg(test)]
     pub fn len(&self) -> usize {
         self.futures.len()

src/utils/mod.rs

@@ -0,0 +1,21 @@
+// Copyright 2024 litep2p developers
+//
+// Permission is hereby granted, free of charge, to any person obtaining a
+// copy of this software and associated documentation files (the "Software"),
+// to deal in the Software without restriction, including without limitation
+// the rights to use, copy, modify, merge, publish, distribute, sublicense,
+// and/or sell copies of the Software, and to permit persons to whom the
+// Software is furnished to do so, subject to the following conditions:
+//
+// The above copyright notice and this permission notice shall be included in
+// all copies or substantial portions of the Software.
+//
+// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS
+// OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
+// FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
+// DEALINGS IN THE SOFTWARE.
+
+pub mod futures_stream;