fix & improve libra for mega

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"

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<String> {
 
     have
 }
+
+/// Read 4 bytes hex number
+async fn read_hex_4(reader: &mut (impl AsyncRead + Unpin)) -> io::Result<u32> {
+    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<u8>)> {
+    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))
+}

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,27 +116,27 @@ 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::<Entry>();
     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 {
         // TODO progress bar
         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<SHA1> {
 }
 
 fn incremental_objs(local_ref: SHA1, remote_ref: SHA1) -> HashSet<Entry> {
+    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<Entry> {
             }
         }
         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<Entry> {
         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<Entry> {
+fn diff_tree_objs(old_tree: Option<&SHA1>, new_tree: &SHA1) -> HashSet<Entry> { // 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<Entry> {
                 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());
                 }

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<String>, want: &Vec<String>) -> 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 {