init commit

Author: RRRadicalEdward

Diff for: .gitignore

@@ -0,0 +1 @@
+/target

Diff for: Cargo.toml

@@ -0,0 +1,21 @@
+[package]
+name = "p2p"
+version = "0.1.0"
+edition = "2021"
+
+[dependencies]
+anyhow = "1"
+clap = { version = "3.0", default-features = false, features = ["std"] }
+futures = { version = "0.3", default-features = false, features = ["std"] }
+log = { version = "0.4", default-features = false, features = ["std"] }
+serde = { version = "1", default-features = false, features = ["derive"] }
+serde_json = { version = "1", default-features = false, features = ["std"] }
+simple_logger = { version = "1.16", default-features = false, features = ["colors"] }
+tokio = { version = "1.15", default-features = false, features = ["io-util", "fs", "signal", "sync", "rt", "rt-multi-thread", "net", "time"] }
+tokio-serde = { version = "0.8", default-features = false, features = ["json"] }
+tokio-util = { version = "0.6", default-features = false, features = ["codec"] }
+uuid = { version = "0.8", default-features = false, features = ["v4", "serde"] }
+rustydht-lib = { path = "../../RustPart2/rustydht-lib" }
+tower = { version = "0.4.12" }
+sha-1 = "0.10.0"
+portpicker = "0.1.1"

Diff for: TODO.md

@@ -0,0 +1,25 @@
+🔥 High priority:
+
+[ ✔ ] Create something like manifest file that contains shards ids, their hashes, pieces size
+ 
+[ 🚧 ] Research DHT(distributed hash table)
+
+ 1. [ ✔ ] Implement/add DHT.  
+ 2. [ 🚧 ] Use DHT logic inside sessions.
+     - Download logic.
+     - Upload logic.
+ 3. [ ] Own BHT bootstrapping node - postponed. use https://github.com/bittorrent/bootstrap-dht for now. 
+
+🦀 Medium priority:
+
+1. Improve manifest file to better support DHT(see https://en.wikipedia.org/wiki/Torrent_file)
+   1. [ ✔ ] Add nodes address in manifest files.
+   2. [ ✔ ] Add the target file size in manifest files.
+   3. [ ✔ ] Add pieces size in manifest files.
+   4. [ ✔ ] Add infohash of a manifest file to uniquely identify a manifest file amount other nodes.
+
+2. Download logic(blocked by DHT research/implementing)   
+
+🐮 Low priority:
+
+1. [ ] Better naming: A "peer" is a client/server listening on a TCP port that implements the BitTorrent protocol. A "node" is a client/server listening on a UDP port implementing the distributed hash table protocol.

Diff for: rustfmt.toml

@@ -0,0 +1,3 @@
+reorder_imports = true
+imports_granularity = "Crate"
+max_width = 120

Diff for: src/config.rs

@@ -0,0 +1,108 @@
+use clap::{App, Arg, ArgMatches, Values};
+use std::{
+    net::SocketAddr,
+    path::{Path, PathBuf},
+};
+
+const ARG_NODE_ADDRESS: &str = "address";
+const ARG_UPLOAD: &str = "upload";
+const ARG_DOWNLOAD: &str = "download";
+pub const MANIFEST_FILE_EXT: &str = "manifest";
+
+#[derive(Clone)]
+pub struct Config {
+    pub node_address: SocketAddr,
+    pub upload_files: Option<Vec<PathBuf>>,
+    pub manifest_files: Option<Vec<PathBuf>>,
+}
+
+impl Config {
+    pub fn new() -> Self {
+        Self::default()
+    }
+
+    fn config() -> ArgMatches {
+        let file_path_validation = |path: &str| -> Result<(), String> {
+            if !Path::new(path).exists() {
+                Err(format!("{path} file doesn't exists"))
+            } else {
+                Ok(())
+            }
+        };
+
+        let manifest_path_validator = |path: &str| -> Result<(), String> {
+            match file_path_validation(path).map(|_| {
+                if Path::new(path).extension() != Some(MANIFEST_FILE_EXT.as_ref()) {
+                    Err(format!("incorrect file extension, expect: .{MANIFEST_FILE_EXT}"))
+                } else {
+                    Ok(())
+                }
+            }) {
+                Ok(Ok(())) => Ok(()),
+                Ok(Err(err)) => Err(err),
+                Err(err) => Err(err),
+            }
+        };
+        let app = App::new("p2p application")
+            .arg(
+                Arg::new(ARG_NODE_ADDRESS)
+                    .short('a')
+                    .long(ARG_NODE_ADDRESS)
+                    .help("address node to be bound")
+                    .takes_value(true)
+                    .validator(|value| {
+                        value
+                            .parse::<SocketAddr>()
+                            .map(|_| ())
+                            .map_err(|err| format!("incorrect socket address: {}", err))
+                    })
+                    .required(true),
+            )
+            .arg(
+                Arg::new(ARG_UPLOAD)
+                    .short('u')
+                    .long(ARG_UPLOAD)
+                    .help("Specify file path to file to share")
+                    .takes_value(true)
+                    .multiple_values(true)
+                    .validator(file_path_validation),
+            )
+            .arg(
+                Arg::new(ARG_DOWNLOAD)
+                    .short('d')
+                    .long(ARG_DOWNLOAD)
+                    .help("Specify manifest path to correspond file to be download")
+                    .takes_value(true)
+                    .multiple_values(true)
+                    .validator(manifest_path_validator),
+            );
+
+        app.get_matches()
+    }
+}
+
+impl Default for Config {
+    fn default() -> Self {
+        let config = Self::config();
+
+        let node_address = config
+            .value_of(ARG_NODE_ADDRESS)
+            .map(|value| value.parse::<SocketAddr>().expect("node address is expected to valid"))
+            .expect("node address is required");
+
+        let values_to_vec_path_buf = |values: Option<Values>| -> Option<Vec<PathBuf>> {
+            Some(
+                values
+                    .into_iter()
+                    .flat_map(|value| value.into_iter().map(PathBuf::from))
+                    .collect(),
+            )
+        };
+
+        Self {
+            node_address,
+            upload_files: values_to_vec_path_buf(config.values_of(ARG_UPLOAD)),
+            manifest_files: values_to_vec_path_buf(config.values_of(ARG_DOWNLOAD)),
+        }
+    }
+}

Diff for: src/dht_node.rs

@@ -0,0 +1,86 @@
+use anyhow::Context;
+use log::{debug, trace};
+use rustydht_lib::{
+    common::{ipv4_addr_src::IPV4Consensus, Id, Node},
+    dht::{
+        dht_event::DHTEvent,
+        operations::{announce_peer, find_node, get_peers, GetPeersResult},
+        DHTBuilder, DHTSettingsBuilder, DHT,
+    },
+    shutdown::{create_shutdown, ShutdownSender},
+};
+use std::{
+    net::{Ipv4Addr, SocketAddrV4},
+    time::Duration,
+};
+use tokio::sync::mpsc::Receiver;
+
+pub struct DhtNode {
+    dht: DHT,
+    timeout_duration: Duration,
+    shutdown_notifier: ShutdownSender,
+}
+
+impl DhtNode {
+    pub fn new(timeout_duration: Duration) -> anyhow::Result<Self> {
+        let (sender, receiver) = create_shutdown();
+
+        let port = portpicker::pick_unused_port().expect("no free port in the system :(");
+        let dht = DHTBuilder::new()
+            .listen_addr(SocketAddrV4::new(Ipv4Addr::UNSPECIFIED, port))
+            .settings(
+                DHTSettingsBuilder::new()
+                    .read_only(false)
+                    .routers(vec!["localhost:6882".to_string()])
+                    .build(),
+            )
+            .ip_source(Box::new(IPV4Consensus::new(0, 10)))
+            .build(receiver)?;
+
+        debug!("running DHT peer on 0.0.0.0:{port}");
+
+        Ok(Self {
+            dht,
+            timeout_duration,
+            shutdown_notifier: sender,
+        })
+    }
+
+    pub async fn run(&self) -> anyhow::Result<()> {
+        self.dht.run_event_loop().await.map_err(|err| err.into())
+    }
+
+    pub async fn announce_peer<H: AsRef<[u8]>>(&self, info_hash: H, port: u16) -> anyhow::Result<Vec<Node>> {
+        debug!("announce infohash: {:02x?}", info_hash.as_ref());
+        let id = Id::from_bytes(info_hash.as_ref())?;
+        announce_peer(&self.dht, id, Some(port), self.timeout_duration)
+            .await
+            .with_context(|| format!("announce peer failed for infohash {:02x?}", info_hash.as_ref()))
+    }
+
+    pub async fn find_nodes<H: AsRef<[u8]>>(&self, info_hash: H) -> anyhow::Result<Vec<Node>> {
+        debug!("find nodes for infohash {:02x?}", info_hash.as_ref());
+        let id = Id::from_bytes(info_hash.as_ref())?;
+        find_node(&self.dht, id, self.timeout_duration)
+            .await
+            .with_context(|| format!("find nodes failed for infohash {:02x?}", info_hash.as_ref()))
+    }
+
+    pub async fn get_peers<H: AsRef<[u8]>>(&self, info_hash: H) -> anyhow::Result<GetPeersResult> {
+        debug!("get peers for infohash {:02x?}", info_hash.as_ref());
+        let id = Id::from_bytes(info_hash.as_ref())?;
+
+        get_peers(&self.dht, id, self.timeout_duration)
+            .await
+            .with_context(|| format!("get peers failed for infohash {:02x?}", info_hash.as_ref()))
+    }
+
+    pub async fn stop(&mut self) {
+        trace!("stopping DHT node");
+        self.shutdown_notifier.shutdown().await
+    }
+
+    pub fn subscribe(&self) -> Receiver<DHTEvent> {
+        self.dht.subscribe()
+    }
+}