mtorrent 0.3.9

Fast and lightweight BitTorrent client in Rust
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56

use super::connections::PeerReporter;
use local_async_utils::prelude::*;
use mtorrent_core::pwp::PeerOrigin;
use mtorrent_dht as dht;
use mtorrent_utils::info_stopwatch;
use std::time::Duration;
use tokio::{select, sync::mpsc, time};

/// Start dht search of `info_hash` and relay results into `peer_sink`.
/// Restart search every minute, exit if the search terminates by itself.
/// When `peer_sink` no longer has a receiver, this will eventually stop,
/// but not immmediately after the channel has been closed.
pub async fn run_dht_search(
    info_hash: [u8; 20],
    dht_cmds: dht::CommandSink,
    peer_reporter: PeerReporter,
    pwp_listener_port: u16,
) {
    const SEARCH_RESTART_INTERVAL: Duration = sec!(120);

    let _sw = info_stopwatch!("DHT search operation");

    async fn do_search(
        info_hash: &[u8; 20],
        dht_cmds: &dht::CommandSink,
        peer_reporter: &PeerReporter,
        pwp_listener_port: u16,
    ) {
        let (sender, mut result_receiver) = mpsc::channel(1024);
        if let Err(e) = dht_cmds
            .send(dht::Command::FindPeers {
                info_hash: *info_hash,
                callback: sender,
                local_peer_port: pwp_listener_port,
            })
            .await
        {
            log::error!("Failed to start DHT search: {e}");
            return;
        }
        while let Some(peer_addr) = result_receiver.recv().await {
            log::debug!("DHT discovered peer: {peer_addr}");
            if !peer_reporter.report_discovered(peer_addr, PeerOrigin::Dht).await {
                return;
            }
        }
    }

    loop {
        select! {
            biased;
            _ = do_search(&info_hash, &dht_cmds, &peer_reporter, pwp_listener_port) => break,
            _ = time::sleep(SEARCH_RESTART_INTERVAL) => (),
        }
    }
}