mtorrent 0.4.0

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
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105

use mtorrent_dht as dht;
use mtorrent_utils::{info_stopwatch, ip, upnp, worker};
use std::io;
use std::net::{Ipv4Addr, SocketAddr, SocketAddrV4};
use std::path::PathBuf;
use tokio::net::UdpSocket;
use tokio::{join, task};

/// Startup configuration for the DHT system.
#[derive(Debug, Clone)]
pub struct Config {
    /// Local UDP port to bind to.
    pub local_port: u16,
    /// Maximum number of outbound queries in flight.
    /// Uses an internal default if `None`.
    pub max_concurrent_queries: Option<usize>,
    /// Directory for storing data persistent across boots.
    pub config_dir: PathBuf,
    /// Create and maintain port mappings via UPnP.
    pub use_upnp: bool,
    /// Override default bootstrap nodes.
    pub bootstrap_nodes_override: Option<Vec<String>>,
}

/// Spawn a thread with a Tokio runtime running the DHT system, and return handle to it.
pub fn launch_dht_node_runtime(cfg: Config) -> io::Result<(worker::rt::Handle, dht::CommandSink)> {
    let (cmd_sender, cmd_server) = dht::setup_commands();

    let worker_handle = worker::with_local_runtime(worker::rt::Config {
        name: "dht".to_owned(),
        io_enabled: true,
        time_enabled: true,
        ..Default::default()
    })?;
    worker_handle.runtime_handle().spawn(async move {
        // spawn_local() must be called from the dht thread
        task::spawn_local(dht_main(
            cmd_server,
            cfg.local_port,
            cfg.config_dir,
            cfg.max_concurrent_queries,
            cfg.use_upnp,
            cfg.bootstrap_nodes_override,
        ));
    });

    Ok((worker_handle, cmd_sender))
}

async fn start_upnp(local_port: u16) -> io::Result<()> {
    let local_addr = ip::get_local_addr()?;

    // try create a port mapping with the same port number
    let port_opener = upnp::PortOpener::new(
        SocketAddr::new(local_addr.into(), local_port),
        upnp::PortMappingProtocol::UDP,
        Some(local_port),
    )
    .await
    .map_err(io::Error::other)?;

    log::info!("UPnP for DHT succeeded, public ip: {}", port_opener.external_ip());

    // start periodic renewal of port mapping. It will stop and remove the mapping
    // automatically once the DHT runtime shuts down
    task::spawn_local(async move {
        if let Err(e) = port_opener.run_continuous_renewal().await {
            log::error!("UPnP port renewal for DHT failed: {e}");
        }
    });
    Ok(())
}

async fn dht_main(
    cmd_server: dht::CommandSource,
    local_port: u16,
    config_dir: PathBuf,
    max_concurrent_queries: Option<usize>,
    use_upnp: bool,
    bootstrap_nodes: Option<Vec<String>>,
) {
    let _sw = info_stopwatch!("DHT");

    let socket = match UdpSocket::bind(SocketAddrV4::new(Ipv4Addr::UNSPECIFIED, local_port)).await {
        Err(e) => return log::error!("Failed to create a UDP socket for DHT: {e}"),
        Ok(socket) => socket,
    };

    if use_upnp && let Err(e) = start_upnp(local_port).await {
        log::error!("UPnP for DHT failed: {e}");
    }

    let (outgoing_msgs_sink, incoming_msgs_source, udp_runner) = dht::setup_udp(socket);

    let (client, server, queries_runner) =
        dht::setup_queries(outgoing_msgs_sink, incoming_msgs_source, max_concurrent_queries);

    let mut processor = dht::Processor::new(config_dir, client);

    if let Some(nodes) = bootstrap_nodes {
        processor.set_bootstrap_nodes(nodes);
    }

    join!(udp_runner.run(), queries_runner.run(), processor.run(server, cmd_server));
}