diff --git a/libra/Cargo.toml b/libra/Cargo.toml index c1ded7889..3f203f4fc 100644 --- a/libra/Cargo.toml +++ b/libra/Cargo.toml @@ -34,6 +34,7 @@ pathdiff = "0.2.1" url = "2.5.0" futures-util = "0.3.30" rpassword = "7.3.1" +indicatif = "0.17.8" [target.'cfg(unix)'.dependencies] # only on Unix pager = "0.16.0" diff --git a/libra/src/command/fetch.rs b/libra/src/command/fetch.rs index 0d99470e3..96ae56ed9 100644 --- a/libra/src/command/fetch.rs +++ b/libra/src/command/fetch.rs @@ -1,10 +1,14 @@ +use std::io; +use std::vec; use std::{collections::HashSet, fs, io::Write}; use ceres::protocol::ServiceType::UploadPack; use clap::Parser; -use futures::StreamExt; +use indicatif::ProgressBar; use mercury::internal::object::commit::Commit; use mercury::{errors::GitError, hash::SHA1}; +use tokio::io::{AsyncRead, AsyncReadExt}; +use tokio_util::io::StreamReader; use url::Url; use crate::command::{ask_basic_auth, load_object}; @@ -97,34 +101,51 @@ pub async fn fetch_repository(remote_config: &RemoteConfig) { .await .unwrap(); - let mut buffer = vec![]; - while let Some(item) = result_stream.next().await { - let item = item.unwrap(); - buffer.extend(item); - } - - // pase pkt line - if let Some(pack_pos) = buffer.windows(4).position(|w| w == b"PACK") { - tracing::info!("pack data found at: {}", pack_pos); - let readable_output = std::str::from_utf8(&buffer[..pack_pos]).unwrap(); - tracing::debug!("stdout readable: \n{}", readable_output); - tracing::info!("pack length: {}", buffer.len() - pack_pos); - assert!(buffer[pack_pos..pack_pos + 4].eq(b"PACK")); - - buffer = buffer[pack_pos..].to_vec(); - } else { - tracing::error!( - "no pack data found, stdout is: \n{}", - std::str::from_utf8(&buffer).unwrap() - ); - panic!("no pack data found"); - } + let mut reader = StreamReader::new(&mut result_stream); + let mut pack_data = Vec::new(); + let mut reach_pack = false; + let bar = ProgressBar::new_spinner(); + loop { + let (len, data) = read_pkt_line(&mut reader).await.unwrap(); + if len == 0 { + break; + } + if data.len() >= 5 && &data[1..5] == b"PACK" { + reach_pack = true; + tracing::debug!("Receiving PACK data..."); + } + if reach_pack { // 2.PACK data + bar.tick(); + // Side-Band Capability, should be enabled if Server Support + let code = data[0]; + let data = &data[1..]; + match code { + 1 => { // Data + pack_data.extend(data); // TODO: decode meanwhile & calc progress + } + 2 => { // Progress + print!("{}", String::from_utf8_lossy(data)); + std::io::stdout().flush().unwrap(); + } + 3 => { // Error + eprintln!("{}", String::from_utf8_lossy(data)); + } + _ => { + eprintln!("unknown side-band-64k code: {}", code); + } + } + } else if &data != b"NAK\n" { // 1.front info (server progress), ignore NAK (first line) + print!("{}", String::from_utf8_lossy(&data)); // data contains '\r' & '\n' at end + std::io::stdout().flush().unwrap(); + } + }; + bar.finish(); /* save pack file */ let pack_file = { - let hash = SHA1::new(&buffer[..buffer.len() - 20].to_vec()); + let hash = SHA1::new(&pack_data[..pack_data.len() - 20].to_vec()); - let checksum = SHA1::from_bytes(&buffer[buffer.len() - 20..]); + let checksum = SHA1::from_bytes(&pack_data[pack_data.len() - 20..]); assert_eq!(hash, checksum); let checksum = checksum.to_plain_str(); println!("checksum: {}", checksum); @@ -133,7 +154,7 @@ pub async fn fetch_repository(remote_config: &RemoteConfig) { .join("pack") .join(format!("pack-{}.pack", checksum)); let mut file = fs::File::create(pack_file.clone()).unwrap(); - file.write_all(&buffer).expect("write failed"); + file.write_all(&pack_data).expect("write failed"); pack_file.to_string_or_panic() }; @@ -223,3 +244,22 @@ async fn current_have() -> Vec { have } + +/// Read 4 bytes hex number +async fn read_hex_4(reader: &mut (impl AsyncRead + Unpin)) -> io::Result { + let mut buf = [0u8; 4]; + reader.read_exact(&mut buf).await?; + let hex_str = std::str::from_utf8(&buf).unwrap(); + u32::from_str_radix(hex_str, 16).map_err(|e| io::Error::new(io::ErrorKind::InvalidData, e)) +} +/// async version of `read_pkt_line` +/// - return (raw length, data) +async fn read_pkt_line(reader: &mut (impl AsyncRead + Unpin)) -> io::Result<(usize, Vec)> { + let len = read_hex_4(reader).await?; + if len == 0 { + return Ok((0, Vec::new())); + } + let mut data = vec![0u8; (len - 4) as usize]; + reader.read_exact(&mut data).await?; + Ok((len as usize, data)) +} \ No newline at end of file diff --git a/libra/src/command/push.rs b/libra/src/command/push.rs index 84748b4d8..7f8aad6f4 100644 --- a/libra/src/command/push.rs +++ b/libra/src/command/push.rs @@ -1,7 +1,9 @@ use std::collections::{HashSet, VecDeque}; +use std::io::Write; use std::str::FromStr; use bytes::BytesMut; use clap::Parser; +use colored::Colorize; use tokio::sync::mpsc; use url::Url; use ceres::protocol::ServiceType::ReceivePack; @@ -114,13 +116,12 @@ pub async fn execute(args: PushArgs) { SHA1::from_str(&commit_hash).unwrap(), SHA1::from_str(&remote_hash).unwrap() ); - println!("Counting objects: {}", objs.len()); // let (tx, rx) = mpsc::channel::(); let (entry_tx, entry_rx) = mpsc::channel(1_000_000); let (stream_tx, mut stream_rx) = mpsc::channel(1_000_000); - let encoder = PackEncoder::new(objs.len(), 5, stream_tx); + let encoder = PackEncoder::new(objs.len(), 0, stream_tx); // TODO: diff slow, so window_size = 0 encoder.encode_async(entry_rx).await.unwrap(); for entry in objs { @@ -128,13 +129,14 @@ pub async fn execute(args: PushArgs) { entry_tx.send(entry).await.unwrap(); } drop(entry_tx); - println!("Delta compression done."); + println!("Compression..."); let mut pack_data = Vec::new(); while let Some(chunk) = stream_rx.recv().await { pack_data.extend(chunk); } data.extend_from_slice(&pack_data); + println!("Delta compression done."); let res = client.send_pack(data.freeze(), auth).await.unwrap(); @@ -155,7 +157,7 @@ pub async fn execute(args: PushArgs) { let (len, _) = read_pkt_line(&mut data); assert_eq!(len, 0); - println!("Push success"); + println!("{}", "Push success".green()); // set after push success if args.set_upstream { @@ -184,6 +186,8 @@ fn collect_history_commits(commit_id: &SHA1) -> HashSet { } fn incremental_objs(local_ref: SHA1, remote_ref: SHA1) -> HashSet { + tracing::debug!("local_ref: {}, remote_ref: {}", local_ref, remote_ref); + // just fast-forward optimization if remote_ref != SHA1::default() { // remote exists let mut commit = Commit::load(&local_ref); @@ -234,12 +238,16 @@ fn incremental_objs(local_ref: SHA1, remote_ref: SHA1) -> HashSet { } } for parent in parents.iter() { - objs.extend(diff_tree_objs(Some(parent), &commit.tree_id)); + let parent_tree = Commit::load(parent).tree_id; + objs.extend(diff_tree_objs(Some(&parent_tree), &commit.tree_id)); if !exist_commits.contains(parent) { queue.push_back(*parent); } } objs.insert(commit.into()); + + print!("Counting objects: {}\r", objs.len()); + std::io::stdout().flush().unwrap(); } // root commit has no parent @@ -248,12 +256,13 @@ fn incremental_objs(local_ref: SHA1, remote_ref: SHA1) -> HashSet { objs.extend(diff_tree_objs(None, &root_tree)); } + println!("Counting objects: {} done.", objs.len()); objs } /// calc objects that in `new_tree` but not in `old_tree` /// - if `old_tree` is None, return all objects in `new_tree` (include tree itself) -fn diff_tree_objs(old_tree: Option<&SHA1>, new_tree: &SHA1) -> HashSet { +fn diff_tree_objs(old_tree: Option<&SHA1>, new_tree: &SHA1) -> HashSet { // TODO: skip objs that has been added in caller let mut objs = HashSet::new(); if let Some(old_tree) = old_tree { if old_tree == new_tree { @@ -280,7 +289,10 @@ fn diff_tree_objs(old_tree: Option<&SHA1>, new_tree: &SHA1) -> HashSet { TreeItemMode::Tree => { objs.extend(diff_tree_objs(None, &item.id)); //TODO optimize, find same name tree } - _ => { + _ => { // TODO: submodule (TreeItemMode: Commit) + if item.mode == TreeItemMode::Commit { // (160000)| Gitlink (Submodule) + eprintln!("fatal: submodule not supported"); + } let blob = Blob::load(&item.id); objs.insert(blob.into()); } diff --git a/libra/src/internal/protocol/https_client.rs b/libra/src/internal/protocol/https_client.rs index f1c5995c6..d6341c584 100644 --- a/libra/src/internal/protocol/https_client.rs +++ b/libra/src/internal/protocol/https_client.rs @@ -126,7 +126,7 @@ impl HttpsClient { refs = refs.trim(); if !read_first_line { if hash == SHA1::default().to_plain_str() { - break; // empty repo, return empty list + break; // empty repo, return empty list // TODO: parse capability } let (head, caps) = refs.split_once('\0').unwrap(); if service == UploadPack.to_string() { @@ -215,16 +215,17 @@ impl HttpsClient { request.send().await } } - +/// for fetching async fn generate_upload_pack_content(have: &Vec, want: &Vec) -> Bytes { let mut buf = BytesMut::new(); let mut write_first_line = false; + let capability = ["side-band-64k", "ofs-delta"].join(" "); for w in want { if !write_first_line { add_pkt_line_string( &mut buf, - format!("want {}\0agent=libra/0.1.0\n", w).to_string(), + format!("want {} {} agent=libra/0.1.0\n", w, capability).to_string(), ); write_first_line = true; } else {