|
|
@@ -16,343 +16,156 @@
|
|
|
* along with this program. If not, see <https://www.gnu.org/licenses/>.
|
|
|
*/
|
|
|
|
|
|
-use std::{collections::HashSet, fs, path::PathBuf};
|
|
|
+use std::collections::{HashMap, HashSet};
|
|
|
|
|
|
-use async_std::{stream::StreamExt, sync::Arc};
|
|
|
+use async_std::{
|
|
|
+ fs::File,
|
|
|
+ stream::StreamExt,
|
|
|
+ sync::{Arc, RwLock},
|
|
|
+};
|
|
|
use async_trait::async_trait;
|
|
|
-use darkfi_serial::serialize;
|
|
|
-use log::{debug, error, info, warn};
|
|
|
+use log::{error, info};
|
|
|
use serde_json::{json, Value};
|
|
|
-use structopt_toml::{serde::Deserialize, structopt::StructOpt, StructOptToml};
|
|
|
+use smol::Executor;
|
|
|
+use structopt_toml::{structopt::StructOpt, StructOptToml};
|
|
|
use url::Url;
|
|
|
|
|
|
use darkfi::{
|
|
|
async_daemonize, cli_desc,
|
|
|
- dht::{waiting_for_response, Dht, DhtPtr},
|
|
|
- net,
|
|
|
+ geode::Geode,
|
|
|
+ net::{self, settings::SettingsOpt, P2p, P2pPtr},
|
|
|
rpc::{
|
|
|
- jsonrpc::{
|
|
|
- ErrorCode,
|
|
|
- ErrorCode::{InvalidParams, MethodNotFound},
|
|
|
- JsonError, JsonRequest, JsonResponse, JsonResult,
|
|
|
- },
|
|
|
+ jsonrpc::{ErrorCode, JsonError, JsonRequest, JsonResponse, JsonResult},
|
|
|
server::{listen_and_serve, RequestHandler},
|
|
|
},
|
|
|
util::path::expand_path,
|
|
|
Result,
|
|
|
};
|
|
|
|
|
|
-mod error;
|
|
|
-use error::{server_error, RpcError};
|
|
|
+/// P2P protocols
|
|
|
+mod proto;
|
|
|
+use proto::{FudFilePut, ProtocolFud};
|
|
|
|
|
|
const CONFIG_FILE: &str = "fud_config.toml";
|
|
|
const CONFIG_FILE_CONTENTS: &str = include_str!("../fud_config.toml");
|
|
|
|
|
|
-#[derive(Clone, Debug, Deserialize, StructOpt, StructOptToml)]
|
|
|
+#[derive(Clone, Debug, serde::Deserialize, StructOpt, StructOptToml)]
|
|
|
#[serde(default)]
|
|
|
#[structopt(name = "fud", about = cli_desc!())]
|
|
|
struct Args {
|
|
|
- #[structopt(short, long)]
|
|
|
- /// Configuration file to use
|
|
|
- config: Option<String>,
|
|
|
-
|
|
|
- #[structopt(long, default_value = "~/.config/darkfi/fud")]
|
|
|
- /// Path to the contents directory
|
|
|
- folder: String,
|
|
|
+ #[structopt(short, parse(from_occurrences))]
|
|
|
+ /// Increase verbosity (-vvv supported)
|
|
|
+ verbose: u8,
|
|
|
|
|
|
#[structopt(long, default_value = "tcp://127.0.0.1:13336")]
|
|
|
/// JSON-RPC listen URL
|
|
|
rpc_listen: Url,
|
|
|
|
|
|
- #[structopt(long)]
|
|
|
- /// P2P accept addresses (repeatable flag)
|
|
|
- p2p_accept: Vec<Url>,
|
|
|
-
|
|
|
- #[structopt(long)]
|
|
|
- /// P2P external addresses (repeatable flag)
|
|
|
- p2p_external: Vec<Url>,
|
|
|
-
|
|
|
- #[structopt(long, default_value = "8")]
|
|
|
- /// Connection slots
|
|
|
- slots: usize,
|
|
|
-
|
|
|
- #[structopt(long)]
|
|
|
- /// Connect to seed (repeatable flag)
|
|
|
- seeds: Vec<Url>,
|
|
|
-
|
|
|
- #[structopt(long)]
|
|
|
- /// Connect to peer (repeatable flag)
|
|
|
- peers: Vec<Url>,
|
|
|
-
|
|
|
- #[structopt(long)]
|
|
|
- /// Prefered transports for outbound connections (repeatable flag)
|
|
|
- transports: Vec<String>,
|
|
|
-
|
|
|
- #[structopt(long)]
|
|
|
- /// Enable localnet hosts
|
|
|
- localnet: bool,
|
|
|
+ #[structopt(short, long)]
|
|
|
+ /// Configuration file to use
|
|
|
+ config: Option<String>,
|
|
|
|
|
|
#[structopt(long)]
|
|
|
- /// Enable channel log
|
|
|
- channel_log: bool,
|
|
|
-
|
|
|
- #[structopt(short, long)]
|
|
|
- /// Set log file to ouput into
|
|
|
+ /// Set log file path to output daemon logs into
|
|
|
log: Option<String>,
|
|
|
|
|
|
- #[structopt(short, parse(from_occurrences))]
|
|
|
- /// Increase verbosity (-vvv supported)
|
|
|
- verbose: u8,
|
|
|
+ #[structopt(long, default_value = "~/.local/share/fud")]
|
|
|
+ /// Base directory for filesystem storage
|
|
|
+ base_dir: String,
|
|
|
+
|
|
|
+ #[structopt(flatten)]
|
|
|
+ /// Network settings
|
|
|
+ net: SettingsOpt,
|
|
|
}
|
|
|
|
|
|
-/// Struct representing the daemon.
|
|
|
pub struct Fud {
|
|
|
- /// Daemon dht state
|
|
|
- dht: DhtPtr,
|
|
|
-
|
|
|
- /// Path to the contents directory
|
|
|
- folder: PathBuf,
|
|
|
+ /// Routing table for file metadata
|
|
|
+ metadata_router: Arc<RwLock<HashMap<blake3::Hash, HashSet<Url>>>>,
|
|
|
+ /// Routing table for file chunks
|
|
|
+ chunks_router: Arc<RwLock<HashMap<blake3::Hash, HashSet<Url>>>>,
|
|
|
+ /// Pointer to the P2P network instance
|
|
|
+ p2p: P2pPtr,
|
|
|
+ /// The Geode instance
|
|
|
+ geode: Geode,
|
|
|
}
|
|
|
|
|
|
-impl Fud {
|
|
|
- pub async fn new(dht: DhtPtr, folder: PathBuf) -> Result<Self> {
|
|
|
- Ok(Self { dht, folder })
|
|
|
- }
|
|
|
-
|
|
|
- /// Initialize fud dht state by reading the contents folder and generating
|
|
|
- /// the corresponding dht records.
|
|
|
- async fn init(&self) -> Result<()> {
|
|
|
- info!("Initializing fud dht state for folder: {:?}", self.folder);
|
|
|
-
|
|
|
- if !self.folder.exists() {
|
|
|
- fs::create_dir_all(&self.folder)?;
|
|
|
+#[async_trait]
|
|
|
+impl RequestHandler for Fud {
|
|
|
+ async fn handle_request(&self, req: JsonRequest) -> JsonResult {
|
|
|
+ if !req.params.is_array() {
|
|
|
+ return JsonError::new(ErrorCode::InvalidParams, None, req.id).into()
|
|
|
}
|
|
|
|
|
|
- let entries = fs::read_dir(&self.folder).unwrap();
|
|
|
- {
|
|
|
- let mut lock = self.dht.write().await;
|
|
|
-
|
|
|
- // Sync lookup map with network
|
|
|
- if let Err(e) = lock.sync_lookup_map().await {
|
|
|
- error!("Failed to sync lookup map: {}", e);
|
|
|
- }
|
|
|
-
|
|
|
- for entry in entries {
|
|
|
- let e = entry.unwrap();
|
|
|
- let name = String::from(e.file_name().to_str().unwrap());
|
|
|
- info!("Entry: {}", name);
|
|
|
- let key_hash = blake3::hash(&serialize(&name));
|
|
|
- let value: Vec<u8> = std::fs::read(e.path()).unwrap();
|
|
|
- if let Err(e) = lock.insert(key_hash, value).await {
|
|
|
- error!("Failed to insert key: {}", e);
|
|
|
- }
|
|
|
- }
|
|
|
- }
|
|
|
+ let params = req.params.as_array().unwrap();
|
|
|
|
|
|
- Ok(())
|
|
|
- }
|
|
|
+ match req.method.as_str() {
|
|
|
+ Some("put") => return self.put(req.id, params).await,
|
|
|
+ Some("get") => return self.get(req.id, params).await,
|
|
|
|
|
|
- /// Signaling fud network that node goes offline.
|
|
|
- async fn disconnect(&self) -> Result<()> {
|
|
|
- debug!("Peer disconnecting, signaling network");
|
|
|
-
|
|
|
- {
|
|
|
- let mut lock = self.dht.write().await;
|
|
|
- let records = lock.map.clone();
|
|
|
- for key in records.keys() {
|
|
|
- let result = lock.remove(*key).await;
|
|
|
- match result {
|
|
|
- Ok(option) => match option {
|
|
|
- Some(k) => {
|
|
|
- debug!("Hash key removed: {}", k);
|
|
|
- }
|
|
|
- None => {
|
|
|
- warn!("Did not find key: {}", key);
|
|
|
- }
|
|
|
- },
|
|
|
- Err(e) => {
|
|
|
- error!("Failed to remove key: {}", e);
|
|
|
- }
|
|
|
- }
|
|
|
- }
|
|
|
+ Some("ping") => return self.pong(req.id, params).await,
|
|
|
+ Some("dnet_switch") => return self.dnet_switch(req.id, params).await,
|
|
|
+ Some("dnet_info") => return self.dnet_info(req.id, params).await,
|
|
|
+ Some(_) | None => return JsonError::new(ErrorCode::MethodNotFound, None, req.id).into(),
|
|
|
}
|
|
|
-
|
|
|
- Ok(())
|
|
|
}
|
|
|
+}
|
|
|
|
|
|
+impl Fud {
|
|
|
// RPCAPI:
|
|
|
- // Returns all folder contents, with file changes.
|
|
|
- // --> {"jsonrpc": "2.0", "method": "list", "params": [], "id": 1}
|
|
|
- // <-- {"jsonrpc": "2.0", "result": "[[files],[new],[deleted]", "id": 1}
|
|
|
- pub async fn list(&self, id: Value, _params: &[Value]) -> JsonResult {
|
|
|
- let mut content = HashSet::new();
|
|
|
- let mut new = HashSet::new();
|
|
|
- let mut deleted = HashSet::new();
|
|
|
-
|
|
|
- let entries = fs::read_dir(&self.folder).unwrap();
|
|
|
- let records = self.dht.read().await.map.clone();
|
|
|
- let mut entries_hashes = HashSet::new();
|
|
|
-
|
|
|
- // We iterate files for new records
|
|
|
- for entry in entries {
|
|
|
- let e = entry.unwrap();
|
|
|
- let name = String::from(e.file_name().to_str().unwrap());
|
|
|
- let key_hash = blake3::hash(&serialize(&name));
|
|
|
- entries_hashes.insert(key_hash);
|
|
|
-
|
|
|
- if records.contains_key(&key_hash) {
|
|
|
- content.insert(name.clone());
|
|
|
- } else {
|
|
|
- new.insert(name);
|
|
|
- }
|
|
|
- }
|
|
|
-
|
|
|
- // We check records for removed files
|
|
|
- for key in records.keys() {
|
|
|
- if entries_hashes.contains(key) {
|
|
|
- continue
|
|
|
- }
|
|
|
- deleted.insert(key.to_string());
|
|
|
+ // Put a file onto the network. Takes a local filesystem path as a parameter.
|
|
|
+ // Returns the fil hashe that serves as a pointer to the uploaded file.
|
|
|
+ //
|
|
|
+ // --> {"jsonrpc": "2.0", "method": "put", "params": ["/foo.txt"], "id": 42}
|
|
|
+ // <-- {"jsonrpc": "2.0", "result: "df4...3db7", "id": 42}
|
|
|
+ async fn put(&self, id: Value, params: &[Value]) -> JsonResult {
|
|
|
+ if params.len() != 1 || !params[0].is_string() {
|
|
|
+ return JsonError::new(ErrorCode::InvalidParams, None, id).into()
|
|
|
}
|
|
|
|
|
|
- JsonResponse::new(json!((content, new, deleted)), id).into()
|
|
|
- }
|
|
|
+ let path = match expand_path(params[0].as_str().unwrap()) {
|
|
|
+ Ok(v) => v,
|
|
|
+ Err(_) => return JsonError::new(ErrorCode::InvalidParams, None, id).into(),
|
|
|
+ };
|
|
|
|
|
|
- // RPCAPI:
|
|
|
- // Iterate contents folder and dht for potential changes.
|
|
|
- // --> {"jsonrpc": "2.0", "method": "sync", "params": [], "id": 1}
|
|
|
- // <-- {"jsonrpc": "2.0", "result": "true", "id": 1}
|
|
|
- pub async fn sync(&self, id: Value, _params: &[Value]) -> JsonResult {
|
|
|
- info!("Sync process started");
|
|
|
-
|
|
|
- let entries = fs::read_dir(&self.folder).unwrap();
|
|
|
- {
|
|
|
- let mut lock = self.dht.write().await;
|
|
|
- let records = lock.map.clone();
|
|
|
- let mut entries_hashes = HashSet::new();
|
|
|
-
|
|
|
- // We iterate files for new records
|
|
|
- for entry in entries {
|
|
|
- let e = entry.unwrap();
|
|
|
- let name = String::from(e.file_name().to_str().unwrap());
|
|
|
- info!("Entry: {}", name);
|
|
|
- let key_hash = blake3::hash(&serialize(&name));
|
|
|
- entries_hashes.insert(key_hash);
|
|
|
-
|
|
|
- if records.contains_key(&key_hash) {
|
|
|
- continue
|
|
|
- }
|
|
|
-
|
|
|
- let value: Vec<u8> = std::fs::read(e.path()).unwrap();
|
|
|
- if let Err(e) = lock.insert(key_hash, value).await {
|
|
|
- error!("Failed to insert key: {}", e);
|
|
|
- return server_error(RpcError::KeyInsertFail, id)
|
|
|
- }
|
|
|
+ // A valid path was passed. Let's see if we can read it, and if so,
|
|
|
+ // add it to Geode.
|
|
|
+ let fd = match File::open(&path).await {
|
|
|
+ Ok(v) => v,
|
|
|
+ Err(e) => {
|
|
|
+ error!("Failed to open {:?}: {}", path, e);
|
|
|
+ return JsonError::new(ErrorCode::InvalidParams, None, id).into()
|
|
|
}
|
|
|
+ };
|
|
|
|
|
|
- // We check records for removed files
|
|
|
- let records = lock.map.clone();
|
|
|
- for key in records.keys() {
|
|
|
- if entries_hashes.contains(key) {
|
|
|
- continue
|
|
|
- }
|
|
|
-
|
|
|
- let result = lock.remove(*key).await;
|
|
|
- match result {
|
|
|
- Ok(option) => match option {
|
|
|
- Some(k) => {
|
|
|
- debug!("Hash key removed: {}", k);
|
|
|
- }
|
|
|
- None => {
|
|
|
- warn!("Did not find key: {}", key);
|
|
|
- }
|
|
|
- },
|
|
|
- Err(e) => {
|
|
|
- error!("Failed to remove key: {}", e);
|
|
|
- return server_error(RpcError::KeyRemoveFail, id)
|
|
|
- }
|
|
|
- }
|
|
|
+ let (file_hash, chunk_hashes) = match self.geode.insert(fd).await {
|
|
|
+ Ok(v) => v,
|
|
|
+ Err(e) => {
|
|
|
+ error!("Failed inserting file {:?} to geode: {}", path, e);
|
|
|
+ // FIXME: Custom error here
|
|
|
+ return JsonError::new(ErrorCode::InvalidParams, None, id).into()
|
|
|
}
|
|
|
- }
|
|
|
+ };
|
|
|
|
|
|
- JsonResponse::new(json!(true), id).into()
|
|
|
+ // TODO: Broadcast file_hash and chunk_hashes as FudFilePut {};
|
|
|
+ let fud_file = FudFilePut { file_hash, chunk_hashes };
|
|
|
+ self.p2p.broadcast(&fud_file).await;
|
|
|
+
|
|
|
+ JsonResponse::new(json!(file_hash.to_hex().as_str()), id).into()
|
|
|
}
|
|
|
|
|
|
// RPCAPI:
|
|
|
- // Checks if provided key exists and retrieve it from the local map or queries the network.
|
|
|
- // Returns key or not found message.
|
|
|
- // --> {"jsonrpc": "2.0", "method": "get", "params": ["name"], "id": 1}
|
|
|
- // <-- {"jsonrpc": "2.0", "result": "path", "id": 1}
|
|
|
- async fn get(&self, id: Value, params: &[Value]) -> JsonResult {
|
|
|
- if params.len() != 1 || !params[0].is_string() {
|
|
|
- return JsonError::new(InvalidParams, None, id).into()
|
|
|
- }
|
|
|
-
|
|
|
- let key = params[0].as_str().unwrap().to_string();
|
|
|
- let key_hash = blake3::hash(&serialize(&key));
|
|
|
-
|
|
|
- // We execute this sequence to prevent lock races between threads
|
|
|
- // Verify key exists
|
|
|
- let exists = self.dht.read().await.contains_key(key_hash);
|
|
|
- if exists.is_none() {
|
|
|
- info!("Did not find key: {}", key);
|
|
|
- return server_error(RpcError::UnknownKey, id)
|
|
|
- }
|
|
|
-
|
|
|
- // Check if key is local or should query network
|
|
|
- let path = self.folder.join(key.clone());
|
|
|
- let local = exists.unwrap();
|
|
|
- if local {
|
|
|
- match self.dht.read().await.get(key_hash) {
|
|
|
- Some(_) => return JsonResponse::new(json!(path), id).into(),
|
|
|
- None => {
|
|
|
- info!("Did not find key: {}", key);
|
|
|
- return server_error(RpcError::UnknownKey, id)
|
|
|
- }
|
|
|
- }
|
|
|
- }
|
|
|
-
|
|
|
- info!("Key doesn't exist locally, querring network...");
|
|
|
- if let Err(e) = self.dht.read().await.request_key(key_hash).await {
|
|
|
- error!("Failed to query key: {}", e);
|
|
|
- return server_error(RpcError::QueryFailed, id)
|
|
|
- }
|
|
|
-
|
|
|
- info!("Waiting response...");
|
|
|
- match waiting_for_response(self.dht.clone()).await {
|
|
|
- Ok(response) => {
|
|
|
- match response {
|
|
|
- Some(resp) => {
|
|
|
- info!("Key found!");
|
|
|
- // Optionally, we insert the key to our local map
|
|
|
- if let Err(e) =
|
|
|
- self.dht.write().await.insert(resp.key, resp.value.clone()).await
|
|
|
- {
|
|
|
- error!("Failed to insert key: {}", e);
|
|
|
- return server_error(RpcError::KeyInsertFail, id)
|
|
|
- }
|
|
|
-
|
|
|
- if let Err(e) = std::fs::write(path.clone(), resp.value) {
|
|
|
- error!("Failed to generate file for key: {}", e);
|
|
|
- return server_error(RpcError::FileGenerationFail, id)
|
|
|
- }
|
|
|
- JsonResponse::new(json!(path), id).into()
|
|
|
- }
|
|
|
- None => {
|
|
|
- info!("Did not find key: {}", key);
|
|
|
- server_error(RpcError::UnknownKey, id)
|
|
|
- }
|
|
|
- }
|
|
|
- }
|
|
|
- Err(e) => {
|
|
|
- error!("Error while waiting network response: {}", e);
|
|
|
- server_error(RpcError::WaitingNetworkError, id)
|
|
|
- }
|
|
|
- }
|
|
|
+ // Fetch a file from the network. Takes a file hash as parameter.
|
|
|
+ // Returns the path to the local file containing the metadata.
|
|
|
+ //
|
|
|
+ // --> {"jsonrpc": "2.0", "method": "get", "params": ["1211...abfd"], "id": 42}
|
|
|
+ // <-- {"jsonrpc": "2.0", "result: "~/.local/share/fud/files/1211...abfd", "id": 42}
|
|
|
+ async fn get(&self, id: Value, _params: &[Value]) -> JsonResult {
|
|
|
+ JsonResponse::new(json!([]), id).into()
|
|
|
}
|
|
|
|
|
|
// RPCAPI:
|
|
|
// Replies to a ping method.
|
|
|
+ //
|
|
|
// --> {"jsonrpc": "2.0", "method": "ping", "params": [], "id": 42}
|
|
|
// <-- {"jsonrpc": "2.0", "result": "pong", "id": 42}
|
|
|
async fn pong(&self, id: Value, _params: &[Value]) -> JsonResult {
|
|
|
@@ -372,9 +185,9 @@ impl Fud {
|
|
|
}
|
|
|
|
|
|
if params[0].as_bool().unwrap() {
|
|
|
- self.dht.read().await.p2p.dnet_enable().await;
|
|
|
+ self.p2p.dnet_enable().await;
|
|
|
} else {
|
|
|
- self.dht.read().await.p2p.dnet_disable().await;
|
|
|
+ self.p2p.dnet_disable().await;
|
|
|
}
|
|
|
|
|
|
JsonResponse::new(json!(true), id).into()
|
|
|
@@ -385,82 +198,52 @@ impl Fud {
|
|
|
// --> {"jsonrpc": "2.0", "method": "dnet_info", "params": [], "id": 42}
|
|
|
// <-- {"jsonrpc": "2.0", result": {"nodeID": [], "nodeinfo": [], "id": 42}
|
|
|
async fn dnet_info(&self, id: Value, _params: &[Value]) -> JsonResult {
|
|
|
- let dnet_info = self.dht.read().await.p2p.dnet_info().await;
|
|
|
- JsonResponse::new(net::P2p::map_dnet_info(dnet_info), id).into()
|
|
|
+ let dnet_info = self.p2p.dnet_info().await;
|
|
|
+ JsonResponse::new(P2p::map_dnet_info(dnet_info), id).into()
|
|
|
}
|
|
|
}
|
|
|
|
|
|
-#[async_trait]
|
|
|
-impl RequestHandler for Fud {
|
|
|
- async fn handle_request(&self, req: JsonRequest) -> JsonResult {
|
|
|
- if !req.params.is_array() {
|
|
|
- return JsonError::new(InvalidParams, None, req.id).into()
|
|
|
- }
|
|
|
+async_daemonize!(realmain);
|
|
|
+async fn realmain(args: Args, ex: Arc<Executor<'_>>) -> Result<()> {
|
|
|
+ // The working directory for this daemon and geode.
|
|
|
+ let basedir = expand_path(&args.base_dir)?;
|
|
|
|
|
|
- let params = req.params.as_array().unwrap();
|
|
|
+ // Hashmaps used for routing
|
|
|
+ let metadata_router = Arc::new(RwLock::new(HashMap::new()));
|
|
|
+ let chunks_router = Arc::new(RwLock::new(HashMap::new()));
|
|
|
|
|
|
- match req.method.as_str() {
|
|
|
- Some("list") => return self.list(req.id, params).await,
|
|
|
- Some("sync") => return self.sync(req.id, params).await,
|
|
|
- Some("get") => return self.get(req.id, params).await,
|
|
|
- Some("ping") => return self.pong(req.id, params).await,
|
|
|
- Some("dnet_switch") => return self.dnet_switch(req.id, params).await,
|
|
|
- Some("dnet_info") => return self.dnet_info(req.id, params).await,
|
|
|
- Some(_) | None => return JsonError::new(MethodNotFound, None, req.id).into(),
|
|
|
- }
|
|
|
- }
|
|
|
-}
|
|
|
+ info!("Instantiating Geode instance");
|
|
|
+ let geode = Geode::new(&basedir.into()).await?;
|
|
|
|
|
|
-async_daemonize!(realmain);
|
|
|
-async fn realmain(args: Args, ex: Arc<smol::Executor<'_>>) -> Result<()> {
|
|
|
- // Signal handling for graceful termination.
|
|
|
- let (signals_handler, signals_task) = SignalHandler::new()?;
|
|
|
+ info!("Instantiating P2P network");
|
|
|
+ let p2p = P2p::new(args.net.into()).await;
|
|
|
|
|
|
- // P2P network
|
|
|
- let network_settings = net::Settings {
|
|
|
- inbound_addrs: args.p2p_accept,
|
|
|
- outbound_connections: args.slots,
|
|
|
- external_addrs: args.p2p_external,
|
|
|
- peers: args.peers.clone(),
|
|
|
- seeds: args.seeds.clone(),
|
|
|
- allowed_transports: args.transports,
|
|
|
- localnet: args.localnet,
|
|
|
- ..Default::default()
|
|
|
- };
|
|
|
-
|
|
|
- let p2p = net::P2p::new(network_settings).await;
|
|
|
-
|
|
|
- // Initialize daemon dht
|
|
|
- let dht = Dht::new(None, p2p.clone(), signals_handler.term_rx.clone(), ex.clone()).await?;
|
|
|
-
|
|
|
- // Initialize daemon
|
|
|
- let folder = expand_path(&args.folder)?;
|
|
|
- let fud = Fud::new(dht.clone(), folder).await?;
|
|
|
- let fud = Arc::new(fud);
|
|
|
-
|
|
|
- // JSON-RPC server
|
|
|
- info!("Starting JSON-RPC server");
|
|
|
- let _ex = ex.clone();
|
|
|
- ex.spawn(listen_and_serve(args.rpc_listen, fud.clone(), _ex)).detach();
|
|
|
+ // Daemon instantiation
|
|
|
+ let fud = Arc::new(Fud { metadata_router, chunks_router, p2p: p2p.clone(), geode });
|
|
|
+ let _fud = fud.clone();
|
|
|
|
|
|
- info!("Starting sync P2P network");
|
|
|
- p2p.clone().start(ex.clone()).await?;
|
|
|
+ info!("Starting JSON-RPC server on {}", args.rpc_listen);
|
|
|
let _ex = ex.clone();
|
|
|
- let _p2p = p2p.clone();
|
|
|
- ex.spawn(async move {
|
|
|
- if let Err(e) = _p2p.run(_ex).await {
|
|
|
- error!("Failed starting P2P network: {}", e);
|
|
|
- }
|
|
|
- })
|
|
|
- .detach();
|
|
|
-
|
|
|
- fud.init().await?;
|
|
|
+ ex.spawn(listen_and_serve(args.rpc_listen, fud, _ex)).detach();
|
|
|
+
|
|
|
+ info!("Starting P2P protocols");
|
|
|
+ let registry = p2p.protocol_registry();
|
|
|
+ registry
|
|
|
+ .register(net::SESSION_ALL, move |channel, p2p| {
|
|
|
+ let _fud = _fud.clone();
|
|
|
+ async move { ProtocolFud::init(_fud, channel, p2p).await.unwrap() }
|
|
|
+ })
|
|
|
+ .await;
|
|
|
+ p2p.clone().start(ex.clone()).await?;
|
|
|
|
|
|
- // Wait for termination signal
|
|
|
+ // Signal handling for graceful termination.
|
|
|
+ let (signals_handler, signals_task) = SignalHandler::new()?;
|
|
|
signals_handler.wait_termination(signals_task).await?;
|
|
|
info!("Caught termination signal, cleaning up and exiting...");
|
|
|
|
|
|
- fud.disconnect().await?;
|
|
|
+ info!("Stopping P2P network");
|
|
|
+ p2p.stop().await;
|
|
|
|
|
|
+ info!("Bye!");
|
|
|
Ok(())
|
|
|
}
|
parazyd