leyoda/rqbit
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Initial peers now passed to read_metainfo_from_peer_receiver

Browse source
This commit is contained in:
Igor Katson 2023-12-01 09:44:51 +00:00
parent f337ab1837
commit 5ee9390bb1
No known key found for this signature in database
GPG key ID: B4EC22B66D61A3F5
2 changed files with 11 additions and 8 deletions
Download patch file Download diff file Expand all files Collapse all files

18
crates/librqbit/src/dht_utils.rs
View file

@ -26,15 +26,12 @@ pub enum ReadMetainfoResult<Rx> {
pub async fn read_metainfo_from_peer_receiver<A: Stream<Item = SocketAddr> + Unpin>( pub async fn read_metainfo_from_peer_receiver<A: Stream<Item = SocketAddr> + Unpin>(
peer_id: Id20, peer_id: Id20,
info_hash: Id20, info_hash: Id20,
mut addrs: A, initial_addrs: Vec<SocketAddr>,
addrs_stream: A,
peer_connection_options: Option<PeerConnectionOptions>, peer_connection_options: Option<PeerConnectionOptions>,
) -> ReadMetainfoResult<A> { ) -> ReadMetainfoResult<A> {
let mut seen = HashSet::<SocketAddr>::new(); let mut seen = HashSet::<SocketAddr>::new();
let first_addr = match addrs.next().await { let mut addrs = addrs_stream;
Some(addr) => addr,
None => return ReadMetainfoResult::ChannelClosed { seen },
};
seen.insert(first_addr);
let semaphore = tokio::sync::Semaphore::new(128); let semaphore = tokio::sync::Semaphore::new(128);
@ -57,7 +54,11 @@ pub async fn read_metainfo_from_peer_receiver<A: Stream<Item = SocketAddr> + Unp
}; };
let mut unordered = FuturesUnordered::new(); let mut unordered = FuturesUnordered::new();
unordered.push(read_info_guarded(first_addr));
for a in initial_addrs {
seen.insert(a);
unordered.push(read_info_guarded(a));
}
loop { loop {
tokio::select! { tokio::select! {
@ -109,7 +110,8 @@ mod tests {
let dht = DhtBuilder::new().await.unwrap(); let dht = DhtBuilder::new().await.unwrap();
let peer_rx = dht.get_peers(info_hash).unwrap(); let peer_rx = dht.get_peers(info_hash).unwrap();
let peer_id = generate_peer_id(); let peer_id = generate_peer_id();
match read_metainfo_from_peer_receiver(peer_id, info_hash, peer_rx, None).await { match read_metainfo_from_peer_receiver(peer_id, info_hash, Vec::new(), peer_rx, None).await
{
ReadMetainfoResult::Found { info, .. } => dbg!(info), ReadMetainfoResult::Found { info, .. } => dbg!(info),
ReadMetainfoResult::ChannelClosed { .. } => todo!("should not have happened"), ReadMetainfoResult::ChannelClosed { .. } => todo!("should not have happened"),
}; };

1
crates/librqbit/src/session.rs
View file

@ -504,6 +504,7 @@ impl Session {
let (info, dht_rx, initial_peers) = match read_metainfo_from_peer_receiver( let (info, dht_rx, initial_peers) = match read_metainfo_from_peer_receiver(
self.peer_id, self.peer_id,
info_hash, info_hash,
opts.initial_peers.clone().unwrap_or_default(),
dht_rx, dht_rx,
Some(self.merge_peer_opts(opts.peer_opts)), Some(self.merge_peer_opts(opts.peer_opts)),
) )