main.rs 34 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915
  1. /* This file is part of DarkFi (https://dark.fi)
  2. *
  3. * Copyright (C) 2020-2025 Dyne.org foundation
  4. *
  5. * This program is free software: you can redistribute it and/or modify
  6. * it under the terms of the GNU Affero General Public License as
  7. * published by the Free Software Foundation, either version 3 of the
  8. * License, or (at your option) any later version.
  9. *
  10. * This program is distributed in the hope that it will be useful,
  11. * but WITHOUT ANY WARRANTY; without even the implied warranty of
  12. * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
  13. * GNU Affero General Public License for more details.
  14. *
  15. * You should have received a copy of the GNU Affero General Public License
  16. * along with this program. If not, see <https://www.gnu.org/licenses/>.
  17. */
  18. use clap::{Parser, Subcommand};
  19. use smol::lock::RwLock;
  20. use std::{
  21. collections::HashMap,
  22. io::{stdout, Write},
  23. sync::Arc,
  24. };
  25. use termcolor::{ColorChoice, StandardStream, WriteColor};
  26. use tracing::error;
  27. use url::Url;
  28. use darkfi::{
  29. cli_desc,
  30. rpc::{
  31. client::RpcClient,
  32. jsonrpc::{ErrorCode, JsonError, JsonRequest, JsonResult},
  33. util::JsonValue,
  34. },
  35. system::{ExecutorPtr, Publisher, StoppableTask},
  36. util::logger::setup_logging,
  37. Error, Result,
  38. };
  39. use fud::{
  40. resource::{Resource, ResourceStatus, ResourceType},
  41. util::hash_to_string,
  42. };
  43. mod util;
  44. use crate::util::{
  45. format_bytes, format_duration, format_progress_bytes, optional_value, print_tree,
  46. status_to_colorspec, type_to_colorspec, TreeNode,
  47. };
  48. #[derive(Parser)]
  49. #[clap(name = "fu", about = cli_desc!(), version)]
  50. #[clap(arg_required_else_help(true))]
  51. struct Args {
  52. #[clap(short, action = clap::ArgAction::Count)]
  53. /// Increase verbosity (-vvv supported)
  54. verbose: u8,
  55. #[clap(short, long, default_value = "tcp://127.0.0.1:13336")]
  56. /// fud JSON-RPC endpoint
  57. endpoint: Url,
  58. #[clap(subcommand)]
  59. command: Subcmd,
  60. }
  61. #[derive(Subcommand)]
  62. enum Subcmd {
  63. /// Retrieve provided resource from the fud network
  64. Get {
  65. /// Resource hash
  66. hash: String,
  67. /// Download path (relative or absolute)
  68. path: Option<String>,
  69. /// Optional list of files you want to download (only used for directories)
  70. #[arg(short, long, num_args = 1..)]
  71. files: Option<Vec<String>>,
  72. },
  73. /// Put a file or directory onto the fud network
  74. Put {
  75. /// File path or directory path
  76. path: String,
  77. },
  78. /// List resources
  79. Ls {},
  80. /// Watch
  81. Watch {},
  82. /// Remove a resource from fud
  83. Rm {
  84. /// Resource hash
  85. hash: String,
  86. },
  87. /// Get the current node buckets
  88. Buckets {},
  89. /// Get the router state
  90. Seeders {},
  91. /// Verify local files
  92. Verify {
  93. /// File hashes
  94. files: Option<Vec<String>>,
  95. },
  96. /// Lookup seeders of a resource from the network
  97. Lookup {
  98. /// Resource hash
  99. hash: String,
  100. },
  101. }
  102. struct Fu {
  103. pub rpc_client: Arc<RpcClient>,
  104. pub endpoint: Url,
  105. }
  106. impl Fu {
  107. async fn get(
  108. &self,
  109. hash: String,
  110. path: Option<String>,
  111. files: Option<Vec<String>>,
  112. ex: ExecutorPtr,
  113. ) -> Result<()> {
  114. let publisher = Publisher::new();
  115. let subscription = Arc::new(publisher.clone().subscribe().await);
  116. let subscriber_task = StoppableTask::new();
  117. let hash_ = hash.clone();
  118. let publisher_ = publisher.clone();
  119. let rpc_client_ = self.rpc_client.clone();
  120. subscriber_task.clone().start(
  121. async move {
  122. let req = JsonRequest::new("subscribe", JsonValue::Array(vec![]));
  123. rpc_client_.subscribe(req, publisher).await
  124. },
  125. move |res| async move {
  126. match res {
  127. Ok(()) | Err(Error::DetachedTaskStopped) => { /* Do nothing */ }
  128. Err(e) => {
  129. error!("{e}");
  130. publisher_
  131. .notify(JsonResult::Error(JsonError::new(
  132. ErrorCode::InternalError,
  133. None,
  134. 0,
  135. )))
  136. .await;
  137. }
  138. }
  139. },
  140. Error::DetachedTaskStopped,
  141. ex.clone(),
  142. );
  143. let progress_bar_width = 20;
  144. let mut started = false;
  145. let mut tstdout = StandardStream::stdout(ColorChoice::Auto);
  146. let mut print_progress = |info: &HashMap<String, JsonValue>| {
  147. started = true;
  148. let rs: Resource = info.get("resource").unwrap().clone().into();
  149. print!("\x1B[2K\r"); // Clear current line
  150. // Progress bar
  151. let percent = if rs.target_bytes_downloaded > rs.target_bytes_size {
  152. 1.0
  153. } else if rs.target_bytes_size > 0 {
  154. rs.target_bytes_downloaded as f64 / rs.target_bytes_size as f64
  155. } else {
  156. 0.0
  157. };
  158. let completed = (percent * progress_bar_width as f64) as usize;
  159. let remaining = match progress_bar_width > completed {
  160. true => progress_bar_width - completed,
  161. false => 0,
  162. };
  163. let bar = "=".repeat(completed) + &" ".repeat(remaining);
  164. print!("[{bar}] {:.1}% | ", percent * 100.0);
  165. // Downloaded / Total (in bytes)
  166. if rs.target_bytes_size > 0 {
  167. if rs.target_bytes_downloaded == rs.target_bytes_size {
  168. print!("{} | ", format_bytes(rs.target_bytes_size));
  169. } else {
  170. print!(
  171. "{} | ",
  172. format_progress_bytes(rs.target_bytes_downloaded, rs.target_bytes_size)
  173. );
  174. }
  175. }
  176. // Download speed (in bytes/sec)
  177. if !rs.speeds.is_empty() && rs.target_chunks_downloaded < rs.target_chunks_count {
  178. print!("{}/s | ", format_bytes(*rs.speeds.last().unwrap() as u64));
  179. }
  180. // Downloaded / Total (in chunks)
  181. if rs.target_chunks_count > 0 {
  182. let s = if rs.target_chunks_count > 1 { "s" } else { "" };
  183. if rs.target_chunks_downloaded == rs.target_chunks_count {
  184. print!("{} chunk{s} | ", rs.target_chunks_count);
  185. } else {
  186. print!(
  187. "{}/{} chunk{s} | ",
  188. rs.target_chunks_downloaded, rs.target_chunks_count
  189. );
  190. }
  191. }
  192. // ETA
  193. if !rs.speeds.is_empty() && rs.target_chunks_downloaded < rs.target_chunks_count {
  194. print!("ETA: {} | ", format_duration(rs.get_eta()));
  195. }
  196. // Status
  197. let is_done = rs.target_chunks_downloaded == rs.target_chunks_count &&
  198. rs.status.as_str() == "incomplete";
  199. let status = if is_done { ResourceStatus::Seeding } else { rs.status };
  200. tstdout.set_color(&status_to_colorspec(&status)).unwrap();
  201. print!(
  202. "{}",
  203. if let ResourceStatus::Seeding = status { "done" } else { status.as_str() }
  204. );
  205. tstdout.reset().unwrap();
  206. stdout().flush().unwrap();
  207. };
  208. let req = JsonRequest::new(
  209. "get",
  210. JsonValue::Array(vec![
  211. JsonValue::String(hash_.clone()),
  212. JsonValue::String(path.unwrap_or_default()),
  213. match files {
  214. Some(files) => {
  215. JsonValue::Array(files.into_iter().map(JsonValue::String).collect())
  216. }
  217. None => JsonValue::Null,
  218. },
  219. ]),
  220. );
  221. // Create a RPC client to send the `get` request
  222. let rpc_client_getter = RpcClient::new(self.endpoint.clone(), ex.clone()).await?;
  223. let _ = rpc_client_getter.request(req).await?;
  224. loop {
  225. match subscription.receive().await {
  226. JsonResult::Notification(n) => {
  227. let params = n.params.get::<HashMap<String, JsonValue>>().unwrap();
  228. let info = params.get("info");
  229. if info.is_none() {
  230. continue
  231. }
  232. let info = info.unwrap().get::<HashMap<String, JsonValue>>().unwrap();
  233. let hash = match info.get("hash") {
  234. Some(hash_value) => hash_value.get::<String>().unwrap(),
  235. None => continue,
  236. };
  237. if *hash != hash_ {
  238. continue;
  239. }
  240. match params.get("event").unwrap().get::<String>().unwrap().as_str() {
  241. "download_started" |
  242. "metadata_download_completed" |
  243. "chunk_download_completed" |
  244. "resource_updated" => {
  245. print_progress(info);
  246. }
  247. "download_completed" => {
  248. let resource_json = info
  249. .get("resource")
  250. .unwrap()
  251. .get::<HashMap<String, JsonValue>>()
  252. .unwrap();
  253. let path = resource_json.get("path").unwrap().get::<String>().unwrap();
  254. print_progress(info);
  255. println!("\nDownload completed:\n{path}");
  256. return Ok(());
  257. }
  258. "metadata_not_found" => {
  259. println!();
  260. return Err(Error::Custom(format!("Could not find {hash}")));
  261. }
  262. "chunk_not_found" => {
  263. // A seeder does not have a chunk we are looking for,
  264. // we will try another seeder so there is nothing to do
  265. }
  266. "missing_chunks" => {
  267. // We tried all seeders and some chunks are still missing
  268. println!();
  269. return Err(Error::Custom("Missing chunks".to_string()));
  270. }
  271. "download_error" => {
  272. // An error that caused the download to be unsuccessful
  273. if started {
  274. println!();
  275. }
  276. return Err(Error::Custom(
  277. info.get("error").unwrap().get::<String>().unwrap().to_string(),
  278. ));
  279. }
  280. _ => {}
  281. }
  282. }
  283. JsonResult::Error(e) => {
  284. return Err(Error::UnexpectedJsonRpc(format!("Got error from JSON-RPC: {e:?}")))
  285. }
  286. x => {
  287. return Err(Error::UnexpectedJsonRpc(format!(
  288. "Got unexpected data from JSON-RPC: {x:?}"
  289. )))
  290. }
  291. }
  292. }
  293. }
  294. async fn put(&self, path: String, ex: ExecutorPtr) -> Result<()> {
  295. let publisher = Publisher::new();
  296. let subscription = Arc::new(publisher.clone().subscribe().await);
  297. let subscriber_task = StoppableTask::new();
  298. let publisher_ = publisher.clone();
  299. let rpc_client_ = self.rpc_client.clone();
  300. subscriber_task.clone().start(
  301. async move {
  302. let req = JsonRequest::new("subscribe", JsonValue::Array(vec![]));
  303. rpc_client_.subscribe(req, publisher).await
  304. },
  305. move |res| async move {
  306. match res {
  307. Ok(()) | Err(Error::DetachedTaskStopped) => { /* Do nothing */ }
  308. Err(e) => {
  309. error!("{e}");
  310. publisher_
  311. .notify(JsonResult::Error(JsonError::new(
  312. ErrorCode::InternalError,
  313. None,
  314. 0,
  315. )))
  316. .await;
  317. }
  318. }
  319. },
  320. Error::DetachedTaskStopped,
  321. ex.clone(),
  322. );
  323. let rpc_client_putter = RpcClient::new(self.endpoint.clone(), ex.clone()).await?;
  324. let req = JsonRequest::new("put", JsonValue::Array(vec![JsonValue::String(path)]));
  325. let rep = rpc_client_putter.request(req).await?;
  326. let path_str = rep.get::<String>().unwrap().clone();
  327. loop {
  328. match subscription.receive().await {
  329. JsonResult::Notification(n) => {
  330. let params = n.params.get::<HashMap<String, JsonValue>>().unwrap();
  331. let info =
  332. params.get("info").unwrap().get::<HashMap<String, JsonValue>>().unwrap();
  333. let path = match info.get("path") {
  334. Some(path) => path.get::<String>().unwrap(),
  335. None => continue,
  336. };
  337. if *path != path_str {
  338. continue;
  339. }
  340. match params.get("event").unwrap().get::<String>().unwrap().as_str() {
  341. "insert_completed" => {
  342. let id = info.get("hash").unwrap().get::<String>().unwrap().to_string();
  343. println!("{id}");
  344. break Ok(())
  345. }
  346. "insert_error" => {
  347. return Err(Error::Custom(
  348. info.get("error").unwrap().get::<String>().unwrap().to_string(),
  349. ));
  350. }
  351. _ => {}
  352. }
  353. }
  354. JsonResult::Error(e) => {
  355. return Err(Error::UnexpectedJsonRpc(format!("Got error from JSON-RPC: {e:?}")))
  356. }
  357. x => {
  358. return Err(Error::UnexpectedJsonRpc(format!(
  359. "Got unexpected data from JSON-RPC: {x:?}"
  360. )))
  361. }
  362. }
  363. }
  364. }
  365. async fn list_resources(&self) -> Result<()> {
  366. let req = JsonRequest::new("list_resources", JsonValue::Array(vec![]));
  367. let rep = self.rpc_client.request(req).await?;
  368. let resources_json: Vec<JsonValue> = rep.clone().try_into().unwrap();
  369. let resources: Vec<Resource> = resources_json.into_iter().map(|v| v.into()).collect();
  370. for resource in resources.iter() {
  371. let tree: Vec<TreeNode<&str>> = vec![
  372. TreeNode::kv("ID", hash_to_string(&resource.hash)),
  373. TreeNode::kvc(
  374. "Type",
  375. resource.rtype.as_str().to_string(),
  376. type_to_colorspec(&resource.rtype),
  377. ),
  378. TreeNode::kvc(
  379. "Status",
  380. resource.status.as_str().to_string(),
  381. status_to_colorspec(&resource.status),
  382. ),
  383. TreeNode::kv("Chunks", {
  384. if let ResourceType::Directory = resource.rtype {
  385. format!(
  386. "{}/{} ({}/{})",
  387. resource.total_chunks_downloaded,
  388. optional_value!(resource.total_chunks_count),
  389. resource.target_chunks_downloaded,
  390. optional_value!(resource.target_chunks_count)
  391. )
  392. } else {
  393. format!(
  394. "{}/{}",
  395. resource.total_chunks_downloaded,
  396. optional_value!(resource.total_chunks_count)
  397. )
  398. }
  399. }),
  400. TreeNode::kv("Bytes", {
  401. if let ResourceType::Directory = resource.rtype {
  402. format!(
  403. "{} ({})",
  404. optional_value!(resource.total_bytes_size, |x: u64| {
  405. format_progress_bytes(resource.total_bytes_downloaded, x)
  406. }),
  407. optional_value!(resource.target_bytes_size, |x: u64| {
  408. format_progress_bytes(resource.target_bytes_downloaded, x)
  409. })
  410. )
  411. } else {
  412. optional_value!(resource.total_bytes_size, |x: u64| format_progress_bytes(
  413. resource.total_bytes_downloaded,
  414. x
  415. ))
  416. }
  417. }),
  418. ];
  419. print_tree(&resource.path.to_string_lossy(), &tree);
  420. }
  421. Ok(())
  422. }
  423. async fn buckets(&self) -> Result<()> {
  424. let req = JsonRequest::new("list_buckets", JsonValue::Array(vec![]));
  425. let rep = self.rpc_client.request(req).await?;
  426. let buckets: Vec<JsonValue> = rep.try_into().unwrap();
  427. let mut empty = true;
  428. for (bucket_i, bucket) in buckets.into_iter().enumerate() {
  429. let nodes: Vec<JsonValue> = bucket.try_into().unwrap();
  430. if nodes.is_empty() {
  431. continue
  432. }
  433. empty = false;
  434. let tree: Vec<TreeNode<String>> = nodes
  435. .into_iter()
  436. .map(|n| {
  437. let node: Vec<JsonValue> = n.try_into().unwrap();
  438. let node_id: JsonValue = node[0].clone();
  439. let addresses: Vec<JsonValue> = node[1].clone().try_into().unwrap();
  440. let addresses_vec: Vec<String> = addresses
  441. .into_iter()
  442. .map(|addr| TryInto::<String>::try_into(addr).unwrap())
  443. .collect();
  444. let node_id_string: String = node_id.try_into().unwrap();
  445. TreeNode {
  446. key: node_id_string,
  447. value: None,
  448. color: None,
  449. children: addresses_vec
  450. .into_iter()
  451. .map(|addr| TreeNode::key(addr.clone()))
  452. .collect(),
  453. }
  454. })
  455. .collect();
  456. print_tree(format!("Bucket {bucket_i}").as_str(), &tree);
  457. }
  458. if empty {
  459. println!("All buckets are empty");
  460. }
  461. Ok(())
  462. }
  463. async fn seeders(&self) -> Result<()> {
  464. let req = JsonRequest::new("list_seeders", JsonValue::Array(vec![]));
  465. let rep = self.rpc_client.request(req).await?;
  466. let resources: HashMap<String, JsonValue> = rep["seeders"].clone().try_into().unwrap();
  467. if resources.is_empty() {
  468. println!("No known seeders");
  469. return Ok(())
  470. }
  471. for (hash, nodes) in resources {
  472. let nodes: Vec<JsonValue> = nodes.try_into().unwrap();
  473. let tree: Vec<TreeNode<String>> = nodes
  474. .into_iter()
  475. .map(|n| {
  476. let node: Vec<JsonValue> = n.try_into().unwrap();
  477. let node_id: JsonValue = node[0].clone();
  478. let addresses: Vec<JsonValue> = node[1].clone().try_into().unwrap();
  479. let addresses_vec: Vec<String> = addresses
  480. .into_iter()
  481. .map(|addr| TryInto::<String>::try_into(addr).unwrap())
  482. .collect();
  483. let node_id_string: String = node_id.try_into().unwrap();
  484. TreeNode {
  485. key: node_id_string,
  486. value: None,
  487. color: None,
  488. children: addresses_vec
  489. .into_iter()
  490. .map(|addr| TreeNode::key(addr.clone()))
  491. .collect(),
  492. }
  493. })
  494. .collect();
  495. print_tree(&hash, &tree);
  496. }
  497. Ok(())
  498. }
  499. async fn watch(&self, ex: ExecutorPtr) -> Result<()> {
  500. let req = JsonRequest::new("list_resources", JsonValue::Array(vec![]));
  501. let rep = self.rpc_client.request(req).await?;
  502. let resources_json: Vec<JsonValue> = rep.clone().try_into().unwrap();
  503. let resources: Arc<RwLock<Vec<Resource>>> = Arc::new(RwLock::new(vec![]));
  504. let publisher = Publisher::new();
  505. let subscription = Arc::new(publisher.clone().subscribe().await);
  506. let subscriber_task = StoppableTask::new();
  507. let publisher_ = publisher.clone();
  508. let rpc_client_ = self.rpc_client.clone();
  509. subscriber_task.clone().start(
  510. async move {
  511. let req = JsonRequest::new("subscribe", JsonValue::Array(vec![]));
  512. rpc_client_.subscribe(req, publisher).await
  513. },
  514. move |res| async move {
  515. match res {
  516. Ok(()) | Err(Error::DetachedTaskStopped) => { /* Do nothing */ }
  517. Err(e) => {
  518. error!("{e}");
  519. publisher_
  520. .notify(JsonResult::Error(JsonError::new(
  521. ErrorCode::InternalError,
  522. None,
  523. 0,
  524. )))
  525. .await;
  526. }
  527. }
  528. },
  529. Error::DetachedTaskStopped,
  530. ex,
  531. );
  532. let mut tstdout = StandardStream::stdout(ColorChoice::Auto);
  533. let mut update_resource = async |resource: &Resource| {
  534. let mut resources_write = resources.write().await;
  535. let i = match resources_write.iter().position(|r| r.hash == resource.hash) {
  536. Some(i) => {
  537. resources_write.remove(i);
  538. resources_write.insert(i, resource.clone());
  539. i
  540. }
  541. None => {
  542. resources_write.push(resource.clone());
  543. resources_write.len() - 1
  544. }
  545. };
  546. // Move the cursor to the i-th line and clear it
  547. print!("\x1b[{};1H\x1B[2K", i + 2);
  548. // Hash
  549. print!("\r{:>44} ", hash_to_string(&resource.hash));
  550. // Type
  551. tstdout.set_color(&type_to_colorspec(&resource.rtype)).unwrap();
  552. print!(
  553. "{:>4} ",
  554. match resource.rtype.as_str() {
  555. "unknown" => "?",
  556. "directory" => "dir",
  557. _ => resource.rtype.as_str(),
  558. }
  559. );
  560. tstdout.reset().unwrap();
  561. // Status
  562. tstdout.set_color(&status_to_colorspec(&resource.status)).unwrap();
  563. print!("{:>11} ", resource.status.as_str());
  564. tstdout.reset().unwrap();
  565. // Downloaded / Total (in bytes)
  566. match resource.total_bytes_size {
  567. 0 => {
  568. print!("{:>5.1} {:>16} ", 0.0, "?");
  569. }
  570. _ => {
  571. let percent = resource.total_bytes_downloaded as f64 /
  572. resource.total_bytes_size as f64 *
  573. 100.0;
  574. if resource.total_bytes_downloaded == resource.total_bytes_size {
  575. print!("{:>5.1} {:>16} ", percent, format_bytes(resource.total_bytes_size));
  576. } else {
  577. print!(
  578. "{:>5.1} {:>16} ",
  579. percent,
  580. format_progress_bytes(
  581. resource.total_bytes_downloaded,
  582. resource.total_bytes_size
  583. )
  584. );
  585. }
  586. }
  587. };
  588. // Downloaded / Total (in chunks)
  589. match resource.total_chunks_count {
  590. 0 => {
  591. print!("{:>9} ", format!("{}/?", resource.total_chunks_downloaded));
  592. }
  593. _ => {
  594. if resource.total_chunks_downloaded == resource.total_chunks_count {
  595. print!("{:>9} ", resource.total_chunks_count.to_string());
  596. } else {
  597. print!(
  598. "{:>9} ",
  599. format!(
  600. "{}/{}",
  601. resource.total_chunks_downloaded, resource.total_chunks_count
  602. )
  603. );
  604. }
  605. }
  606. };
  607. // Download speed (in bytes/sec)
  608. let speed_available = resource.total_bytes_downloaded < resource.total_bytes_size &&
  609. resource.status.as_str() == "downloading" &&
  610. !resource.speeds.is_empty();
  611. print!(
  612. "{:>12} ",
  613. match speed_available {
  614. false => "-".to_string(),
  615. true => format!("{}/s", format_bytes(*resource.speeds.last().unwrap() as u64)),
  616. }
  617. );
  618. // ETA
  619. let eta = resource.get_eta();
  620. print!(
  621. "{:>6}",
  622. match eta {
  623. 0 => "-".to_string(),
  624. _ => format_duration(eta),
  625. }
  626. );
  627. println!();
  628. // Move the cursor to end
  629. print!("\x1b[{};1H", resources_write.len() + 2);
  630. stdout().flush().unwrap();
  631. };
  632. let print_begin = async || {
  633. // Clear
  634. print!("\x1B[2J\x1B[1;1H");
  635. // Print column headers
  636. println!(
  637. "\x1b[4m{:>44} {:>4} {:>11} {:>5} {:>16} {:>9} {:>12} {:>6}\x1b[0m",
  638. "Hash", "Type", "Status", "%", "Bytes", "Chunks", "Speed", "ETA"
  639. );
  640. };
  641. print_begin().await;
  642. if resources_json.is_empty() {
  643. println!("No known resources");
  644. } else {
  645. for resource in resources_json.iter() {
  646. let rs: Resource = resource.clone().into();
  647. update_resource(&rs).await;
  648. }
  649. }
  650. loop {
  651. match subscription.receive().await {
  652. JsonResult::Notification(n) => {
  653. let params = n.params.get::<HashMap<String, JsonValue>>().unwrap();
  654. let info = params.get("info");
  655. if info.is_none() {
  656. continue
  657. }
  658. let info = info.unwrap().get::<HashMap<String, JsonValue>>().unwrap();
  659. match params.get("event").unwrap().get::<String>().unwrap().as_str() {
  660. "download_started" |
  661. "metadata_download_completed" |
  662. "chunk_download_completed" |
  663. "download_completed" |
  664. "missing_chunks" |
  665. "metadata_not_found" |
  666. "resource_updated" => {
  667. let resource: Resource = info.get("resource").unwrap().clone().into();
  668. update_resource(&resource).await;
  669. }
  670. "resource_removed" => {
  671. {
  672. let hash = info.get("hash").unwrap().get::<String>().unwrap();
  673. let mut resources_write = resources.write().await;
  674. let i = resources_write
  675. .iter()
  676. .position(|r| hash_to_string(&r.hash) == *hash);
  677. if let Some(i) = i {
  678. resources_write.remove(i);
  679. }
  680. }
  681. let r = resources.read().await.clone();
  682. print_begin().await;
  683. for resource in r.iter() {
  684. update_resource(resource).await;
  685. }
  686. }
  687. "download_error" => {
  688. // An error that caused the download to be unsuccessful
  689. }
  690. _ => {}
  691. }
  692. }
  693. JsonResult::Error(e) => {
  694. return Err(Error::UnexpectedJsonRpc(format!("Got error from JSON-RPC: {e:?}")))
  695. }
  696. x => {
  697. return Err(Error::UnexpectedJsonRpc(format!(
  698. "Got unexpected data from JSON-RPC: {x:?}"
  699. )))
  700. }
  701. }
  702. }
  703. }
  704. async fn remove(&self, hash: String) -> Result<()> {
  705. let req = JsonRequest::new("remove", JsonValue::Array(vec![JsonValue::String(hash)]));
  706. self.rpc_client.request(req).await?;
  707. Ok(())
  708. }
  709. async fn verify(&self, files: Option<Vec<String>>) -> Result<()> {
  710. let files = files.unwrap_or_default().into_iter().map(JsonValue::String).collect();
  711. let req = JsonRequest::new("verify", JsonValue::Array(files));
  712. self.rpc_client.request(req).await?;
  713. Ok(())
  714. }
  715. async fn lookup(&self, hash: String, ex: ExecutorPtr) -> Result<()> {
  716. let publisher = Publisher::new();
  717. let subscription = Arc::new(publisher.clone().subscribe().await);
  718. let subscriber_task = StoppableTask::new();
  719. let publisher_ = publisher.clone();
  720. let rpc_client_ = self.rpc_client.clone();
  721. subscriber_task.clone().start(
  722. async move {
  723. let req = JsonRequest::new("subscribe", JsonValue::Array(vec![]));
  724. rpc_client_.subscribe(req, publisher).await
  725. },
  726. move |res| async move {
  727. match res {
  728. Ok(()) | Err(Error::DetachedTaskStopped) => { /* Do nothing */ }
  729. Err(e) => {
  730. error!("{e}");
  731. publisher_
  732. .notify(JsonResult::Error(JsonError::new(
  733. ErrorCode::InternalError,
  734. None,
  735. 0,
  736. )))
  737. .await;
  738. }
  739. }
  740. },
  741. Error::DetachedTaskStopped,
  742. ex.clone(),
  743. );
  744. let req =
  745. JsonRequest::new("lookup", JsonValue::Array(vec![JsonValue::String(hash.clone())]));
  746. let rpc_client_lookup = RpcClient::new(self.endpoint.clone(), ex.clone()).await?;
  747. rpc_client_lookup.request(req).await?;
  748. let print_seeders = |info: &HashMap<String, JsonValue>| {
  749. let seeders = info.get("seeders").unwrap().get::<Vec<JsonValue>>().unwrap();
  750. for seeder in seeders {
  751. let seeder = seeder.get::<HashMap<String, JsonValue>>().unwrap();
  752. let node: HashMap<String, JsonValue> =
  753. seeder.get("node").unwrap().clone().try_into().unwrap();
  754. let node_id: String = node.get("id").unwrap().clone().try_into().unwrap();
  755. let addresses: Vec<JsonValue> =
  756. node.get("addresses").unwrap().clone().try_into().unwrap();
  757. let tree: Vec<_> = addresses
  758. .into_iter()
  759. .map(|addr| TreeNode::key(TryInto::<String>::try_into(addr).unwrap()))
  760. .collect();
  761. print_tree(node_id.as_str(), &tree);
  762. }
  763. };
  764. loop {
  765. match subscription.receive().await {
  766. JsonResult::Notification(n) => {
  767. let params = n.params.get::<HashMap<String, JsonValue>>().unwrap();
  768. let info =
  769. params.get("info").unwrap().get::<HashMap<String, JsonValue>>().unwrap();
  770. let hash_ = match info.get("hash") {
  771. Some(hash_value) => hash_value.get::<String>().unwrap(),
  772. None => continue,
  773. };
  774. if hash != *hash_ {
  775. continue;
  776. }
  777. if params.get("event").unwrap().get::<String>().unwrap().as_str() ==
  778. "seeders_found"
  779. {
  780. print_seeders(info);
  781. break
  782. }
  783. }
  784. JsonResult::Error(e) => {
  785. return Err(Error::UnexpectedJsonRpc(format!("Got error from JSON-RPC: {e:?}")))
  786. }
  787. x => {
  788. return Err(Error::UnexpectedJsonRpc(format!(
  789. "Got unexpected data from JSON-RPC: {x:?}"
  790. )))
  791. }
  792. }
  793. }
  794. Ok(())
  795. }
  796. }
  797. fn main() -> Result<()> {
  798. let args = Args::parse();
  799. setup_logging(args.verbose, None)?;
  800. let ex = Arc::new(smol::Executor::new());
  801. smol::block_on(async {
  802. ex.run(async {
  803. let rpc_client = Arc::new(RpcClient::new(args.endpoint.clone(), ex.clone()).await?);
  804. let fu = Fu { rpc_client, endpoint: args.endpoint.clone() };
  805. match args.command {
  806. Subcmd::Get { hash, path, files } => fu.get(hash, path, files, ex.clone()).await,
  807. Subcmd::Put { path } => fu.put(path, ex.clone()).await,
  808. Subcmd::Ls {} => fu.list_resources().await,
  809. Subcmd::Watch {} => fu.watch(ex.clone()).await,
  810. Subcmd::Rm { hash } => fu.remove(hash).await,
  811. Subcmd::Buckets {} => fu.buckets().await,
  812. Subcmd::Seeders {} => fu.seeders().await,
  813. Subcmd::Verify { files } => fu.verify(files).await,
  814. Subcmd::Lookup { hash } => fu.lookup(hash, ex.clone()).await,
  815. }?;
  816. Ok(())
  817. })
  818. .await
  819. })
  820. }