chore: Replace tokio-stream StreamMap with pollable-map StreamMap (#306)

CHANGELOG.md

@@ -21,6 +21,7 @@
 - refactor: Remove redundant static lifetime. [PR 301](https://github.com/dariusc93/rust-ipfs/pull/301)
 - refactor: Remove optional error from `UnixfsStatus`.
 - refactor: Remove sled and redb datastore and dependency. [PR 304](https://github.com/dariusc93/rust-ipfs/pull/304)
+- chore: Replace tokio-stream `StreamMap` with pollable-map `StreamMap`. [PR 306](https://github.com/dariusc93/rust-ipfs/pull/306)
 
 # 0.11.21
 - chore: Put libp2p-webrtc-websys behind feature.

Cargo.toml

@@ -56,6 +56,7 @@ multihash-derive = "0.9.0"
 p256 = { version = "0.13.2", default-features = false, features = ["ecdsa", "std", "pem"] }
 parking_lot = "0.12.3"
 pem = { version = "3.0.4" }
+pollable-map = "0.1.0-alpha"
 quick-protobuf = { version = "0.8.1" }
 quick-protobuf-codec = "0.3.1"
 rand = "0.8.5"
@@ -118,6 +119,7 @@ multihash-derive.workspace = true
 p256.workspace = true
 parking_lot.workspace = true
 pem.workspace = true
+pollable-map.workspace = true
 quick-protobuf-codec.workspace = true
 quick-protobuf.workspace = true
 rand.workspace = true

src/p2p/bitswap.rs

@@ -23,7 +23,7 @@ use libp2p::{
     },
     Multiaddr, PeerId,
 };
-use tokio_stream::StreamMap;
+use pollable_map::stream::StreamMap;
 
 mod bitswap_pb {
     pub use super::pb::bitswap_pb::Message;
@@ -280,12 +280,7 @@ impl Behaviour {
         match cids.is_empty() {
             false => {
                 for cid in cids {
-                    let Some(session) = self
-                        .want_session
-                        .iter_mut()
-                        .find(|(session_cid, _)| *session_cid == cid)
-                        .map(|(_, session)| session)
-                    else {
+                    let Some(session) = self.want_session.get_mut(&cid) else {
                         continue;
                     };
                     for peer_id in &peers {
@@ -416,12 +411,7 @@ impl NetworkBehaviour for Behaviour {
                 self.have_session.insert(*cid, have_session);
             }
 
-            let Some(session) = self
-                .have_session
-                .iter_mut()
-                .find(|(session_cid, _)| session_cid == cid)
-                .map(|(_, session)| session)
-            else {
+            let Some(session) = self.have_session.get_mut(cid) else {
                 if !*cancel {
                     tracing::warn!(block = %cid, %peer_id, %connection_id, "have session does not exist. Skipping request");
                 }
@@ -444,12 +434,7 @@ impl NetworkBehaviour for Behaviour {
         }
 
         for (cid, response) in responses {
-            let Some(session) = self
-                .want_session
-                .iter_mut()
-                .find(|(session_cid, _)| *session_cid == cid)
-                .map(|(_, session)| session)
-            else {
+            let Some(session) = self.want_session.get_mut(&cid) else {
                 tracing::warn!(block = %cid, %peer_id, %connection_id, "want session does not exist. Skipping response");
                 continue;
             };