From 68d9e90e15f3e882b5a5a9e5b053acd0517b1aeb Mon Sep 17 00:00:00 2001 From: yyjeqhc <1772413353@qq.com> Date: Tue, 3 Jun 2025 09:09:08 +0800 Subject: [PATCH] feat: add load file while load the dirs from the server. --- ceres/src/api_service/mod.rs | 86 +++++++++++++++ mono/src/api/api_router.rs | 25 +++++ scorpio/src/dicfuse/async_io.rs | 15 +-- scorpio/src/dicfuse/content_store.rs | 37 +++++++ scorpio/src/dicfuse/mod.rs | 42 ++----- scorpio/src/dicfuse/store.rs | 159 ++++++++++++++++++++++++--- scorpio/src/main.rs | 1 + scorpio/tests/dir_test.rs | 140 +++++++++++++++-------- vault/src/pgp.rs | 4 +- 9 files changed, 403 insertions(+), 106 deletions(-) create mode 100644 scorpio/src/dicfuse/content_store.rs diff --git a/ceres/src/api_service/mod.rs b/ceres/src/api_service/mod.rs index 82c38205b..19babcdf1 100644 --- a/ceres/src/api_service/mod.rs +++ b/ceres/src/api_service/mod.rs @@ -216,6 +216,92 @@ pub trait ApiHandler: Send + Sync { } } + /// the dir's hash as same as old,file's hash is the content hash + /// may think about change dir'hash as the content + /// for now,only change the file's hash + async fn get_tree_content_hash(&self, path: PathBuf) -> Result, GitError> { + let mut cache = GitObjectCache::default(); + match self.search_tree_by_path(&path).await? { + Some(tree) => { + let mut item_to_commit = HashMap::new(); + + self.add_trees_to_map( + &mut item_to_commit, + tree.tree_items + .iter() + .filter(|x| x.mode == TreeItemMode::Tree) + .map(|x| x.id.to_string()) + .collect(), + ) + .await; + + // self.add_blobs_to_map( + // &mut item_to_commit, + // tree.tree_items + // .iter() + // .filter(|x| x.mode == TreeItemMode::Blob) + // .map(|x| x.id.to_string()) + // .collect(), + // ) + // .await; + let content_items: Vec = tree + .tree_items + .iter() + .filter(|x| x.mode == TreeItemMode::Blob) + .map(|x| TreeCommitItem { + oid: x.id.to_string(), + name: x.name.clone(), + content_type: "file".to_owned(), + message: String::new(), + date: String::new(), + }) + .collect(); + let commit_ids: HashSet = item_to_commit.values().cloned().collect(); + let commits = self + .get_commits_by_hashes(commit_ids.into_iter().collect()) + .await + .unwrap(); + let commit_map: HashMap = + commits.into_iter().map(|x| (x.id.to_string(), x)).collect(); + + let mut root_commit: Option = None; + let mut item_to_commit_map: HashMap> = HashMap::new(); + for item in tree.tree_items { + if item.mode == TreeItemMode::Blob { + continue; + } + if let Some(commit_id) = item_to_commit.get(&item.id.to_string()) { + let commit = if let Some(commit) = commit_map.get(commit_id) { + commit.to_owned() + } else { + tracing::warn!("failed fetch from commit map: {}", commit_id); + if root_commit.is_none() { + root_commit = Some(self.get_root_commit().await); + } + let root_commit = root_commit.as_ref().unwrap().clone(); + self.traverse_commit_history(&path, &root_commit, &item, &mut cache) + .await + }; + item_to_commit_map.insert(item, Some(commit)); + } + } + let mut items: Vec = item_to_commit_map + .into_iter() + .map(TreeCommitItem::from) + .collect(); + // sort with type and date + items.sort_by(|a, b| { + a.content_type + .cmp(&b.content_type) + .then(a.name.cmp(&b.name)) + }); + items.extend(content_items); + Ok(items) + } + None => Ok(Vec::new()), + } + } + /// return the dir's hash only async fn get_tree_dir_hash( &self, diff --git a/mono/src/api/api_router.rs b/mono/src/api/api_router.rs index 9a36edc3d..f7ac64cbe 100644 --- a/mono/src/api/api_router.rs +++ b/mono/src/api/api_router.rs @@ -31,6 +31,7 @@ pub fn routers() -> OpenApiRouter { .routes(routes!(create_file)) .routes(routes!(get_latest_commit)) .routes(routes!(get_tree_commit_info)) + .routes(routes!(get_tree_content_hash)) .routes(routes!(get_tree_dir_hash)) .routes(routes!(path_can_be_cloned)) .routes(routes!(get_tree_info)) @@ -173,6 +174,30 @@ async fn get_tree_commit_info( Ok(Json(CommonResult::success(Some(data)))) } +/// Get tree content hash,the dir's hash as same as old,file's hash is the content hash +#[utoipa::path( + get, + path = "/tree/content-hash", + params( + CodePreviewQuery + ), + responses( + (status = 200, body = CommonResult>, content_type = "application/json") + ), + tag = GIT_TAG +)] +async fn get_tree_content_hash( + Query(query): Query, + state: State, +) -> Result>>, ApiError> { + let data = state + .api_handler(query.path.clone().into()) + .await? + .get_tree_content_hash(query.path.into()) + .await?; + Ok(Json(CommonResult::success(Some(data)))) +} + /// return the dir's hash #[utoipa::path( get, diff --git a/scorpio/src/dicfuse/async_io.rs b/scorpio/src/dicfuse/async_io.rs index 75dcdb95b..c81b47bf7 100644 --- a/scorpio/src/dicfuse/async_io.rs +++ b/scorpio/src/dicfuse/async_io.rs @@ -32,11 +32,8 @@ impl Filesystem for Dicfuse { .ok_or_else(|| std::io::Error::from_raw_os_error(libc::ENODATA))?; ppath.push(name.to_string_lossy().into_owned()); - let chil = store.get_by_path(&ppath.to_string()).await?; - if self.open_buff.read().await.get(&chil.get_inode()).is_none() { - let _ = self.load_one_file(parent, name).await; - } - let re = self.get_stat(chil).await; + let child = store.get_by_path(&ppath.to_string()).await?; + let re = self.get_stat(child).await; Ok(re) } /// initialize filesystem. Called before any other filesystem method. @@ -122,9 +119,9 @@ impl Filesystem for Dicfuse { data: Bytes::from("".as_bytes()), }); } - let read_lock = self.open_buff.read().await; - let datas = read_lock - .get(&inode) + let datas = self + .store + .get_file_content(inode) .ok_or_else(|| std::io::Error::from_raw_os_error(libc::ENOENT))?; let _offset = offset as usize; let end = (_offset + size as usize).min(datas.len()); @@ -201,8 +198,6 @@ impl Filesystem for Dicfuse { let items = self.store.do_readdir(parent, fh, offset).await?; let mut d: Vec> = Vec::new(); - let parent_item = self.store.get_inode(parent).await?; - self.load_files(parent_item, &items).await; for (index, item) in items.into_iter().enumerate() { if index as u64 >= offset { let attr = self.get_stat(item.clone()).await; diff --git a/scorpio/src/dicfuse/content_store.rs b/scorpio/src/dicfuse/content_store.rs new file mode 100644 index 000000000..452503c13 --- /dev/null +++ b/scorpio/src/dicfuse/content_store.rs @@ -0,0 +1,37 @@ +use crate::util::config; +use sled::Db; +use std::io; +use std::io::{Error, ErrorKind}; + +pub struct ContentStorage { + db: Db, +} +#[allow(unused)] +impl ContentStorage { + pub fn new_from_db(db: Db) -> Self { + ContentStorage { db } + } + pub fn new() -> io::Result { + let store_path = config::store_path(); + let path = format!("{}/content.db", store_path); + let db = sled::open(path)?; + Ok(ContentStorage { db }) + } + pub fn insert_file(&self, inode: u64, content: &[u8]) -> io::Result<()> { + self.db + .insert(inode.to_be_bytes(), content) + .map_err(Error::other)?; + Ok(()) + } + + pub fn get_file_content(&self, inode: u64) -> io::Result> { + match self.db.get(inode.to_be_bytes())? { + Some(value) => Ok(value.to_vec()), + None => Err(Error::new(ErrorKind::NotFound, "File not found")), + } + } + pub fn remove_file(&self, inode: u64) -> std::io::Result<()> { + self.db.remove(inode.to_be_bytes())?; + Ok(()) + } +} diff --git a/scorpio/src/dicfuse/mod.rs b/scorpio/src/dicfuse/mod.rs index 7606da542..59bcd4df0 100644 --- a/scorpio/src/dicfuse/mod.rs +++ b/scorpio/src/dicfuse/mod.rs @@ -1,11 +1,11 @@ mod abi; mod async_io; +mod content_store; pub mod store; mod tree_store; use crate::manager::fetch::fetch_tree; use crate::util::config; use std::{ - collections::HashMap, ffi::{OsStr, OsString}, sync::Arc, }; @@ -19,7 +19,6 @@ use tree_store::StorageItem; pub struct Dicfuse { readable: bool, pub store: Arc, - open_buff: Arc>>>, } #[allow(unused)] impl Dicfuse { @@ -27,15 +26,11 @@ impl Dicfuse { Self { readable: config::dicfuse_readable(), store: DictionaryStore::new().await.into(), // Assuming DictionaryStore has a new() method - open_buff: Arc::new(tokio::sync::RwLock::new(HashMap::new())), } } pub async fn get_stat(&self, item: StorageItem) -> ReplyEntry { let mut e = item.get_stat(); - let rl = self.open_buff.read().await; - if let Some(datas) = rl.get(&item.get_inode()) { - e.attr.size = datas.len() as u64; - } + e.attr.size = self.store.get_file_len(item.get_inode()); e } async fn load_one_file(&self, parent: u64, name: &OsStr) -> std::io::Result<()> { @@ -72,11 +67,7 @@ impl Dicfuse { parent_item.push(i.name.clone()); let it_temp = self.store.get_by_path(&parent_item.to_string()).await?; - - self.open_buff - .write() - .await - .insert(it_temp.get_inode(), data); + self.store.save_file(it_temp.get_inode(), data); } else { eprintln!("Request failed with status: {}", response.status()); } @@ -88,12 +79,7 @@ impl Dicfuse { if !self.readable { return; } - if self - .open_buff - .read() - .await - .contains_key(&parent_item.get_inode()) - { + if self.store.file_exists(parent_item.get_inode()) { return; } let gpath = self.store.find_path(parent_item.get_inode()).await.unwrap(); @@ -131,26 +117,18 @@ impl Dicfuse { // Look up the buff, find Loaded file. if is_first { - match self.open_buff.write().await.get(&hit_inodes) { - Some(_) => { - break; - // is loaded , no need to reload ; - } - None => { - // this dictionary is not loaded , just go ahead. - is_first = false; - } + if self.store.file_exists(hit_inodes) { + // if the file is already exists, no need to load again. + break; } + self.store.save_file(hit_inodes, data); + is_first = false; } - self.open_buff.write().await.insert(hit_inodes, data); } else { eprintln!("Request failed with status: {}", response.status()); } } - self.open_buff - .write() - .await - .insert(parent_item.get_inode(), Vec::new()); + self.store.save_file(parent_item.get_inode(), Vec::new()); } } diff --git a/scorpio/src/dicfuse/store.rs b/scorpio/src/dicfuse/store.rs index 0f544d796..86f05ea5d 100644 --- a/scorpio/src/dicfuse/store.rs +++ b/scorpio/src/dicfuse/store.rs @@ -3,6 +3,7 @@ use async_recursion::async_recursion; /// Read only file system for obtaining and displaying monorepo directory information use core::panic; use crossbeam::queue::SegQueue; +use dashmap::mapref::one::Ref; use dashmap::DashMap; use fuse3::raw::reply::ReplyEntry; use fuse3::FileType; @@ -19,6 +20,7 @@ use tokio::sync::Mutex; use tokio::sync::Notify; use super::abi::{default_dic_entry, default_file_entry}; +use super::content_store::ContentStorage; use super::tree_store::{StorageItem, TreeStorage}; use crate::util::{config, GPath}; const UNKNOW_INODE: u64 = 0; // illegal inode number; @@ -190,13 +192,44 @@ async fn fetch_tree(path: &str) -> Result { Err(e) => Err(e.into()), } } + +/// Download a file from the server using its OID/hash +async fn fetch_file(oid: &str) -> Vec { + let file_blob_endpoint = config::file_blob_endpoint(); + let url = format!("{}/{}", file_blob_endpoint, oid); + let client = Client::new(); + + // Send GET request + let response = match client.get(url).send().await { + Ok(resp) => resp, + Err(_) => { + eprintln!("Failed to fetch file with OID: {}", oid); + return Vec::new(); // Return empty vector on error + } + }; + + // Ensure that the response status is successful + if response.status().is_success() { + // Get the binary data from the response body + let content = match response.bytes().await { + Ok(bytes) => bytes, + Err(_) => { + eprintln!("Failed to read content for OID: {}", oid); + return Vec::new(); // Return empty vector on error + } + }; + return content.to_vec(); + } + Vec::new() +} + async fn fetch_dir(path: &str) -> Result { static CLIENT: Lazy = Lazy::new(Client::new); let client = CLIENT.clone(); let clean_path = path.trim_start_matches('/'); let url = format!( - "{}/api/v1/tree/commit-info?path=/{}", + "{}/api/v1/tree/content-hash?path=/{}", config::base_url(), clean_path ); @@ -345,6 +378,8 @@ pub struct DictionaryStore { persistent_path_store: Arc, // persistent path store for saving and retrieving file paths max_depth: Arc, // max depth for loading directories init_notify: Arc, // used in dir_test to notify the start of the test.. + persistent_content_store: Arc, // persistent content store for saving and retrieving file contents + open_buff: Arc>>, // buffer for open files } #[allow(unused)] @@ -359,14 +394,16 @@ impl DictionaryStore { dirs: Arc::new(DashMap::new()), max_depth: Arc::new(config::load_dir_depth()), init_notify: Arc::new(Notify::new()), + persistent_content_store: Arc::new( + ContentStorage::new().expect("Failed to create ContentStorage"), + ), + open_buff: Arc::new(DashMap::new()), } } - #[inline(always)] pub fn max_depth(&self) -> usize { *self.max_depth } - pub async fn wait_for_ready(&self) { // Wait for the store to be initialized self.init_notify.notified().await; @@ -512,6 +549,8 @@ impl DictionaryStore { depth_items } + /// When a file changed,the parent directory's hash changed too. + /// So we need to update the ancestors' hash . pub async fn update_ancestors_hash(&self, inode: u64) { let item = self.persistent_path_store.get_item(inode).unwrap(); let mut parent_inode = item.get_parent(); @@ -534,7 +573,15 @@ impl DictionaryStore { parent_inode = item.get_parent(); } } + /// When scorpio start,if the db is not empty, we need to load all the files to the memory. + fn load_file(&self, inode: u64) -> Result<(), io::Error> { + let file_content = self.persistent_content_store.get_file_content(inode)?; + let _ = self.open_buff.insert(inode, file_content); + Ok(()) + } + #[async_recursion] + /// Loads directories recursively from the parent path into memory. async fn load_dirs(&self, path: PathBuf, parent_inode: u64) -> Result<(), io::Error> { let root_item = self.persistent_path_store.get_item(parent_inode)?; self.dirs.insert( @@ -561,10 +608,14 @@ impl DictionaryStore { if child_item.is_dir() { self.load_dirs(child_path, child).await?; + } else { + self.load_file(child); } } Ok(()) } + + /// When the scorpio start,we need to load the directories and files from db to the memory. pub async fn load_db(&self) -> Result<(), io::Error> { let mut path = PathBuf::from("/"); self.load_dirs(path, 1).await?; @@ -632,7 +683,7 @@ impl DictionaryStore { self.get_inode(inode).await } - /// get the inode from path + /// Get the inode from path pub async fn get_inode_from_path(&self, path: &str) -> Result { let inode = if path.is_empty() || path == "/" { 1 @@ -681,6 +732,48 @@ impl DictionaryStore { } } +/// File operations interface for in-memory file management +/// Provides functions to handle file content stored in memory buffer (open_buff) +impl DictionaryStore { + pub fn get_file_len(&self, inode: u64) -> u64 { + self.open_buff.get(&inode).map_or(0, |v| v.len() as u64) + } + pub fn remove_file_by_node(&self, inode: u64) -> Result<(), io::Error> { + self.persistent_content_store.remove_file(inode)?; + self.open_buff.remove(&inode); + Ok(()) + } + /// Save to db and then save in the memory. + pub fn save_file(&self, inode: u64, content: Vec) { + self.persistent_content_store + .insert_file(inode, &content) + .expect("Failed to save file content"); + self.open_buff.insert(inode, content); + } + /// Check if the file exists in the memory. + pub fn file_exists(&self, inode: u64) -> bool { + self.open_buff.contains_key(&inode) + } + /// Get the file content from the memory. + pub fn get_file_content(&self, inode: u64) -> Option>> { + self.open_buff.get(&inode) + } + + /// Doanload the file content from the server and save it to the db and memory. + pub async fn fetch_file_content(&self, inode: u64, oid: &str) { + let content = fetch_file(oid).await; + self.save_file(inode, content); + } + /// Return the content of a file by its path. + pub async fn get_file_content_by_path(&self, path: &str) -> Result, io::Error> { + let inode = self.get_inode_from_path(path).await?; + if let Some(content) = self.get_file_content(inode) { + Ok(content.to_vec()) + } else { + Err(io::Error::new(io::ErrorKind::NotFound, "File not found")) + } + } +} /// Loads subdirectories from a remote server into an empty parent directory up to a specified depth. /// /// # Arguments @@ -715,6 +808,8 @@ pub async fn load_dir_depth(store: Arc, parent_path: String, ma file_list: HashMap::new(), }, ); + } else { + store.fetch_file_content(it_inode, it.hash.as_str()).await; } } } @@ -773,6 +868,11 @@ pub async fn load_dir_depth(store: Arc, parent_path: String, ma file_list: HashMap::new(), }, ); + } else { + // If it's a file, fetch its content + store + .fetch_file_content(new_inode, newit.hash.as_str()) + .await; } } } @@ -956,13 +1056,22 @@ pub async fn load_dir(store: Arc, parent_path: String, max_dept if is_dir { println!("hash changes dir {:?}", path); load_dir(store.clone(), path.to_owned(), max_depth).await; + } else { + let inode = store.get_inode_from_path(&path).await.unwrap(); + let item = store.persistent_path_store.get_item(inode).unwrap(); + if item.hash != it.hash { + // update the hash in the db. + tree_db.update_item_hash(inode, it.hash.to_owned()).unwrap(); + store.fetch_file_content(inode, &it.hash).await + } } } else { dirs.get_mut(&parent_path) .unwrap() .file_list .insert(path.to_owned(), true); - let _ = store.update_inode(parent_inode, it.clone()).await.unwrap(); + println!("load dir add new file {:?}", path); + let new_node = store.update_inode(parent_inode, it.clone()).await.unwrap(); //fetch a new dir. if is_dir { println!("add dir {:?}", path); @@ -974,6 +1083,8 @@ pub async fn load_dir(store: Arc, parent_path: String, max_dept }, ); load_dir_depth(store.clone(), path.to_owned(), max_depth).await; + } else { + store.fetch_file_content(new_node, &it.hash).await } } } @@ -994,6 +1105,7 @@ pub async fn load_dir(store: Arc, parent_path: String, max_dept let inode = store.get_inode_from_path(&item).await.unwrap(); println!("delete {:?} {} ", inode, item); tree_db.remove_item(inode).unwrap(); + let _ = store.remove_file_by_node(inode); } return true; } @@ -1021,30 +1133,39 @@ pub async fn update_dir(store: Arc, parent_path: String) { .unwrap() .file_list .insert(path.to_owned(), true); - if is_dir && dirs.get(&path).unwrap().hash != it.hash { - //when the dir's hash changed,fetch the dir. - // If the path already exists, update the hash - update_dir(store.clone(), path.to_owned()).await; - - let mut dir_it = dirs.get_mut(&path).unwrap(); - dir_it.hash = it.hash.to_owned(); - //also update the hash in the db. - let inode = store.get_inode_from_path(&path).await.unwrap(); + + let inode = store.get_inode_from_path(&path).await.unwrap(); + let item = store.persistent_path_store.get_item(inode).unwrap(); + if item.hash != it.hash { + if is_dir { + //when the dir's hash changed,fetch the dir. + // If the path already exists, update the hash + update_dir(store.clone(), path.to_owned()).await; + + let mut dir_it = dirs.get_mut(&path).unwrap(); + dir_it.hash = it.hash.to_owned(); + //also update the hash in the db. + + println!("modify dir {:?}", path); + } else { + // If it's a file, fetch its content + // update the hash in the db. + store.fetch_file_content(inode, &it.hash).await + } tree_db.update_item_hash(inode, it.hash).unwrap(); - println!("modify dir {:?}", path); } } else { dirs.get_mut(&parent_path) .unwrap() .file_list .insert(path.to_owned(), true); + println!("update_dir new add file {:?}", path); let parent_inode = store.get_inode_from_path(&parent_path).await.unwrap(); - let _ = store.update_inode(parent_inode, it.clone()).await.unwrap(); + let new_node = store.update_inode(parent_inode, it.clone()).await.unwrap(); //fetch a new dir. if is_dir { println!("add dir {:?}", path); - dirs.insert( path, DirItem { @@ -1052,6 +1173,9 @@ pub async fn update_dir(store: Arc, parent_path: String) { file_list: HashMap::new(), }, ); + } else { + // If it's a file, fetch its content + store.fetch_file_content(new_node, &it.hash).await; } } } @@ -1073,6 +1197,7 @@ pub async fn update_dir(store: Arc, parent_path: String) { let inode = store.get_inode_from_path(&item).await.unwrap(); println!("delete {:?} {} ", inode, item); tree_db.remove_item(inode).unwrap(); + let _ = store.remove_file_by_node(inode); } } diff --git a/scorpio/src/main.rs b/scorpio/src/main.rs index ef93bcc19..fde02398f 100644 --- a/scorpio/src/main.rs +++ b/scorpio/src/main.rs @@ -43,6 +43,7 @@ async fn main() { let mountpoint = OsStr::new(workspace); let lgfs = LoggingFileSystem::new(fuse_interface.clone()); let mut mount_handle = mount_filesystem(lgfs, mountpoint).await; + let handle = &mut mount_handle; // spawn the server running function. diff --git a/scorpio/tests/dir_test.rs b/scorpio/tests/dir_test.rs index 73b186519..a4638630d 100644 --- a/scorpio/tests/dir_test.rs +++ b/scorpio/tests/dir_test.rs @@ -1,5 +1,5 @@ use core::panic; -use http::Method; +// use http::Method; use lazy_static::lazy_static; use libfuse_fs::passthrough::newlogfs::LoggingFileSystem; use scorpio::dicfuse::store; @@ -19,7 +19,8 @@ use std::time::SystemTime; use std::time::UNIX_EPOCH; use std::{env, fs, io}; use std::{ffi::OsStr, sync::Arc}; -use testcontainers::core::wait::HttpWaitStrategy; +// use testcontainers::core::wait::HttpWaitStrategy; +use testcontainers::core::wait::LogWaitStrategy; use testcontainers::{ core::{IntoContainerPort, Mount, ReuseDirective, WaitFor}, runners::AsyncRunner, @@ -35,6 +36,7 @@ enum SCORCommand { GitAddFile(String, String), // add a new file and update to check the watch_dir GitDeleteFile(String), // remove a new file and update to check the watch_dir Shutdown, // finish and close the file system service + ReadFileContent(String), // read the content of a file } #[derive(Debug, Clone, Serialize, Deserialize)] @@ -42,7 +44,8 @@ enum CommandResult { StoreDirectoryStructure(HashMap>), //used to return the directory structure Success, Error(String), - InitFinish(usize), // used to indicate the initialization is finished + InitFinish(usize), // used to indicate the initialization is finished + FileContent(Vec), // used to return the content of a file } lazy_static! { @@ -231,11 +234,12 @@ async fn mono_container(mapping_port: u16) -> ContainerAsync { GenericImage::new("ubuntu", "latest") .with_exposed_port(mapping_port.tcp()) - .with_wait_for(WaitFor::Http(Box::new( - HttpWaitStrategy::new("/") - .with_method(Method::GET) - .with_expected_status_code(404_u16), - ))) + // .with_wait_for(WaitFor::Http(Box::new( + // HttpWaitStrategy::new("/") + // .with_method(Method::GET) + // .with_expected_status_code(404_u16), + // ))) + .with_wait_for(WaitFor::Log(LogWaitStrategy::stdout("CommonHttpOptions"))) .with_mapped_port(mapping_port, mapping_port.tcp()) .with_mount(Mount::bind_mount(MONO.to_str().unwrap(), "/root/mono")) .with_working_dir("/root") @@ -308,50 +312,78 @@ async fn test_scorpio_service_with_containers() { println!("\n===== add a file and WatchDir ====="); let test_file = "test_file.txt"; - let test_content = format!( - "now: {}", - SystemTime::now() - .duration_since(UNIX_EPOCH) - .unwrap() - .as_secs() - ); - - cmd_tx - .send(SCORCommand::GitAddFile(test_file.to_string(), test_content)) - .await - .unwrap(); - - if let Some(result) = result_rx.recv().await { - match result { - CommandResult::Success => { - println!("git add success."); - } - _ => { - cmd_tx.send(SCORCommand::Shutdown).await.unwrap(); - // let _ = scorpio_handle.await; - let _ = result_rx.recv().await; + for i in 0..2 { + let test_content = format!( + "now: {} {}", + i, + SystemTime::now() + .duration_since(UNIX_EPOCH) + .unwrap() + .as_secs() + ); + + cmd_tx + .send(SCORCommand::GitAddFile( + test_file.to_string(), + test_content.to_owned(), + )) + .await + .unwrap(); + + if let Some(result) = result_rx.recv().await { + match result { + CommandResult::Success => { + println!("git add success."); + } + _ => { + cmd_tx.send(SCORCommand::Shutdown).await.unwrap(); + // let _ = scorpio_handle.await; + let _ = result_rx.recv().await; - panic!("git add file error."); + panic!("git add file error."); + } } } - } - cmd_tx.send(SCORCommand::WatchDir()).await.unwrap(); + cmd_tx.send(SCORCommand::WatchDir()).await.unwrap(); - if let Some(result) = result_rx.recv().await { - match result { - CommandResult::StoreDirectoryStructure(result) => { - assert!( - result.get(&0).unwrap().contains(&test_file.to_string()), - "WatchDir fail: did not find the added file" - ); + if let Some(result) = result_rx.recv().await { + match result { + CommandResult::StoreDirectoryStructure(result) => { + assert!( + result.get(&0).unwrap().contains(&test_file.to_string()), + "WatchDir fail: did not find the added file" + ); + } + _ => { + cmd_tx.send(SCORCommand::Shutdown).await.unwrap(); + // let _ = scorpio_handle.await; + let _ = result_rx.recv().await; + + panic!("WatchDir error."); + } } - _ => { - cmd_tx.send(SCORCommand::Shutdown).await.unwrap(); - // let _ = scorpio_handle.await; - let _ = result_rx.recv().await; + } + cmd_tx + .send(SCORCommand::ReadFileContent(test_file.to_string())) + .await + .unwrap(); + if let Some(result) = result_rx.recv().await { + match result { + CommandResult::FileContent(result) => { + assert_eq!( + result, + test_content.as_bytes(), + "ReadFileContent failed to get the correct content" + ); + } + _ => { + cmd_tx.send(SCORCommand::Shutdown).await.unwrap(); + // let _ = scorpio_handle.await; + let _ = result_rx.recv().await; - panic!("WatchDir error."); + panic!("WatchDir error."); + } } } } @@ -523,6 +555,24 @@ async fn test_scorpio_dir( .await; } } + SCORCommand::ReadFileContent(path) => { + env::set_current_dir(&repo_dir).unwrap(); + let base_path = "/third-party/dir_test/"; + + let file_path = base_path.to_string() + &path; + // if !file_path.exists() { + // let _ = result_tx + // .send(CommandResult::Error(format!("File not found: {}", path))) + // .await; + // continue; + // } + println!("ReadFileContent {:?}", file_path); + let content = store + .get_file_content_by_path(&file_path) + .await + .expect("Failed to get file content"); + let _ = result_tx.send(CommandResult::FileContent(content)).await; + } SCORCommand::GitAddFile(path, content) => { env::set_current_dir(&repo_dir).unwrap(); let _ = fs::write(repo_dir.join(&path), content); diff --git a/vault/src/pgp.rs b/vault/src/pgp.rs index 926262986..f9db57bd8 100644 --- a/vault/src/pgp.rs +++ b/vault/src/pgp.rs @@ -4,10 +4,10 @@ /// using asynchronous operations. use smallvec::smallvec; -use pgp::{SecretKeyParams, SecretKeyParamsBuilder, SubkeyParamsBuilder}; -use pgp::types::SecretKeyTrait; pub use pgp::composed::{Deserializable, SignedPublicKey, SignedSecretKey}; +use pgp::types::SecretKeyTrait; pub use pgp::KeyType; +use pgp::{SecretKeyParams, SecretKeyParamsBuilder, SubkeyParamsBuilder}; use crate::vault::{delete_secret, read_secret, write_secret};