diff --git a/crates/dht/src/dht.rs b/crates/dht/src/dht.rs index e6610aa..183b80a 100644 --- a/crates/dht/src/dht.rs +++ b/crates/dht/src/dht.rs @@ -145,7 +145,7 @@ struct RecursiveRequest { callbacks: C, } -struct RequestPeersStream { +pub struct RequestPeersStream { rx: tokio::sync::mpsc::UnboundedReceiver, cancel_join_handle: tokio::task::JoinHandle<()>, } @@ -959,10 +959,7 @@ impl DhtState { Ok(state) } - pub fn get_peers( - self: &Arc, - info_hash: Id20, - ) -> anyhow::Result + Unpin> { + pub fn get_peers(self: &Arc, info_hash: Id20) -> anyhow::Result { Ok(RequestPeersStream::new(self.clone(), info_hash)) } diff --git a/crates/dht/src/lib.rs b/crates/dht/src/lib.rs index 5a28d07..30bb171 100644 --- a/crates/dht/src/lib.rs +++ b/crates/dht/src/lib.rs @@ -8,7 +8,7 @@ use std::sync::Arc; use std::time::Duration; pub use crate::dht::DhtStats; -pub use crate::dht::{DhtConfig, DhtState}; +pub use crate::dht::{DhtConfig, DhtState, RequestPeersStream}; pub use librqbit_core::id20::Id20; pub use persistence::{PersistentDht, PersistentDhtConfig}; diff --git a/crates/librqbit/src/session.rs b/crates/librqbit/src/session.rs index 14cb4ed..4304da2 100644 --- a/crates/librqbit/src/session.rs +++ b/crates/librqbit/src/session.rs @@ -12,7 +12,7 @@ use std::{ use anyhow::{bail, Context}; use bencode::{bencode_serialize_to_writer, BencodeDeserializer}; use buffers::ByteString; -use dht::{Dht, DhtBuilder, Id20, PersistentDht, PersistentDhtConfig}; +use dht::{Dht, DhtBuilder, Id20, PersistentDht, PersistentDhtConfig, RequestPeersStream}; use librqbit_core::{ magnet::Magnet, peer_id::generate_peer_id, @@ -21,7 +21,6 @@ use librqbit_core::{ use parking_lot::RwLock; use reqwest::Url; use serde::{Deserialize, Deserializer, Serialize, Serializer}; -use tokio_stream::StreamExt; use tracing::{debug, error, error_span, info, warn}; use crate::{ @@ -451,7 +450,7 @@ impl Session { pub async fn add_torrent( &self, - add: impl Into>, + add: AddTorrent<'_>, opts: Option, ) -> anyhow::Result { // Magnet links are different in that we first need to discover the metadata. @@ -502,7 +501,11 @@ impl Session { ( info_hash, info, - if opts.paused { None } else { Some(dht_rx) }, + if opts.paused || opts.list_only { + None + } else { + Some(dht_rx) + }, trackers, initial_peers, ) @@ -527,7 +530,7 @@ impl Session { }; let dht_rx = match self.dht.as_ref() { - Some(dht) if !opts.paused => { + Some(dht) if !opts.paused && !opts.list_only => { debug!("reading peers for {:?} from DHT", torrent.info_hash); Some(dht.get_peers(torrent.info_hash)?) } @@ -578,7 +581,7 @@ impl Session { &self, info_hash: Id20, info: TorrentMetaV1Info, - dht_peer_rx: Option + Unpin + Send + Sync + 'static>, + dht_peer_rx: Option, initial_peers: Vec, trackers: Vec, opts: AddTorrentOptions, diff --git a/crates/librqbit/src/torrent_state/mod.rs b/crates/librqbit/src/torrent_state/mod.rs index 1e9e72c..28035c4 100644 --- a/crates/librqbit/src/torrent_state/mod.rs +++ b/crates/librqbit/src/torrent_state/mod.rs @@ -15,6 +15,7 @@ use std::time::Duration; use anyhow::bail; use anyhow::Context; use buffers::ByteString; +use dht::RequestPeersStream; use librqbit_core::id20::Id20; use librqbit_core::lengths::Lengths; use librqbit_core::peer_id::generate_peer_id; @@ -165,7 +166,7 @@ impl ManagedTorrent { pub fn start( self: &Arc, initial_peers: Vec, - peer_rx: Option + Unpin + Send + Sync + 'static>, + peer_rx: Option, start_paused: bool, ) -> anyhow::Result<()> { let mut g = self.locked.write(); @@ -195,7 +196,7 @@ impl ManagedTorrent { fn spawn_peer_adder( live: &Arc, initial_peers: Vec, - peer_rx: Option + Unpin + Send + Sync + 'static>, + peer_rx: Option, ) { let span = live.meta().span.clone(); let live = Arc::downgrade(live);