diff --git a/crates/librqbit/src/torrent_state/live/mod.rs b/crates/librqbit/src/torrent_state/live/mod.rs index 65e7918..f58357c 100644 --- a/crates/librqbit/src/torrent_state/live/mod.rs +++ b/crates/librqbit/src/torrent_state/live/mod.rs @@ -66,6 +66,7 @@ use librqbit_core::{ torrent_metainfo::TorrentMetaV1Info, }; use parking_lot::{RwLock, RwLockReadGuard, RwLockWriteGuard}; +use peer::OutgoingAddressType; use peer_binary_protocol::{ extended::{ handshake::ExtendedHandshake, ut_metadata::UtMetadata, ut_pex::UtPex, ExtendedMessage, @@ -785,7 +786,11 @@ impl TorrentStateLive { pub(crate) fn reconnect_all_not_needed_peers(&self) { for mut pe in self.peers.states.iter_mut() { if pe.state.not_needed_to_queued(&self.peer_stats()) { - let retry_addr = pe.value().outgoing_address.unwrap_or_else(|| *pe.key()); + let retry_addr = match pe.value().outgoing_address { + peer::OutgoingAddressType::Default => *pe.key(), + peer::OutgoingAddressType::None => continue, + peer::OutgoingAddressType::Known(socket_addr) => socket_addr, + }; if self.peer_queue_tx.send(retry_addr).is_err() { return; } @@ -920,18 +925,16 @@ impl<'a> PeerConnectionHandler for &'a PeerHandler { if let Some(peer_pex_msg_id) = hs.ut_pex() { trace!("peer supports pex at {peer_pex_msg_id}"); } - if let Some(port) = hs.p { - // Lets update outgoing Socket address for incoming connection - if self.incoming { - if let Ok(port) = hs.port() { - let peer_ip = hs.ip_addr().unwrap_or(self.addr.ip()); - let outgoing_addr = SocketAddr::new(peer_ip, port); - self.state - .peers - .with_peer_mut(self.addr, "update outgoing addr", |peer| { - peer.outgoing_address = Some(outgoing_addr) - }); - } + // Lets update outgoing Socket address for incoming connection + if self.incoming { + if let Some(port) = hs.port() { + let peer_ip = hs.ip_addr().unwrap_or(self.addr.ip()); + let outgoing_addr = SocketAddr::new(peer_ip, port); + self.state + .peers + .with_peer_mut(self.addr, "update outgoing addr", |peer| { + peer.outgoing_address = OutgoingAddressType::Known(outgoing_addr) + }); } } Ok(()) diff --git a/crates/librqbit/src/torrent_state/live/peer/mod.rs b/crates/librqbit/src/torrent_state/live/peer/mod.rs index d932f55..ac2521b 100644 --- a/crates/librqbit/src/torrent_state/live/peer/mod.rs +++ b/crates/librqbit/src/torrent_state/live/peer/mod.rs @@ -17,11 +17,19 @@ pub(crate) type InflightRequest = ChunkInfo; pub(crate) type PeerRx = UnboundedReceiver; pub(crate) type PeerTx = UnboundedSender; +#[derive(Debug, Default)] +pub(crate) enum OutgoingAddressType { + #[default] + Default, + None, + Known(SocketAddr), +} + #[derive(Debug, Default)] pub(crate) struct Peer { pub state: PeerStateNoMut, pub stats: stats::atomic::PeerStats, - pub outgoing_address: Option, + pub outgoing_address: OutgoingAddressType, } impl Peer { @@ -37,7 +45,7 @@ impl Peer { Self { state, stats: Default::default(), - outgoing_address: None, + outgoing_address: OutgoingAddressType::None, } } } diff --git a/crates/peer_binary_protocol/src/extended/handshake.rs b/crates/peer_binary_protocol/src/extended/handshake.rs index 65af794..66a004f 100644 --- a/crates/peer_binary_protocol/src/extended/handshake.rs +++ b/crates/peer_binary_protocol/src/extended/handshake.rs @@ -71,14 +71,14 @@ where } pub fn ip_addr(&self) -> Option { - if let Some(b) = self.ipv4 { + if let Some(ref b) = self.ipv4 { let b = b.as_slice(); if b.len() == 4 { let ip_bytes: &[u8; 4] = b[0..4].try_into().unwrap(); // Safe to unwrap as we check slice length return Some(IpAddr::from(*ip_bytes)); } } - if let Some(b) = self.ipv6 { + if let Some(ref b) = self.ipv6 { let b = b.as_slice(); if b.len() == 16 { let ip_bytes: &[u8; 16] = b[0..16].try_into().unwrap(); // Safe to unwrap as we check slice length