Chewing_Bever
/
rieter
1
0
Fork 0
Code Issues 11 Pull Requests Packages Releases Activity

refactor: switch to new repo actors

agents
Jef Roosens 2024-06-25 17:05:14 +02:00
parent 656df06b4e
commit 80d5291508
Signed by: Jef Roosens
GPG Key ID: B75D4F293C7052DB
4 changed files with 79 additions and 69 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

30
server/src/main.rs
View File

@ -23,7 +23,7 @@ pub const ANY_ARCH: &'static str = "any";
#[derive(Clone)] #[derive(Clone)]
pub struct Global { pub struct Global {
config: crate::config::Config, config: crate::config::Config,
mgr: Arc<repo::RepoMgr>, repo: repo::Handle,
db: sea_orm::DbConn, db: sea_orm::DbConn,
} }
@ -54,26 +54,32 @@ fn setup(rt: &runtime::Handle, config_file: PathBuf) -> crate::Result<Global> {
let db = rt.block_on(crate::db::connect(&config.db))?; let db = rt.block_on(crate::db::connect(&config.db))?;
rt.block_on(crate::db::Migrator::up(&db, None))?; rt.block_on(crate::db::Migrator::up(&db, None))?;
let mgr = match &config.fs { let repo = match &config.fs {
FsConfig::Local { data_dir } => { FsConfig::Local { data_dir } => {
rt.block_on(crate::repo::RepoMgr::new( crate::repo::Handle::start(
data_dir.join("repos"), data_dir.join("repos"),
db.clone(), db.clone(),
))? rt.clone(),
config.pkg_workers,
)?
//rt.block_on(crate::repo::RepoMgr::new(
// data_dir.join("repos"),
// db.clone(),
//))?
//RepoHandle::start(data_dir.join("repos"), db.clone(), config.pkg_workers, rt.clone())? //RepoHandle::start(data_dir.join("repos"), db.clone(), config.pkg_workers, rt.clone())?
} }
}; };
let mgr = Arc::new(mgr); //let mgr = Arc::new(mgr);
//
for _ in 0..config.pkg_workers { //for _ in 0..config.pkg_workers {
let clone = Arc::clone(&mgr); // let clone = Arc::clone(&mgr);
//
rt.spawn(async move { clone.pkg_parse_task().await }); // rt.spawn(async move { clone.pkg_parse_task().await });
} //}
Ok(Global { Ok(Global {
config: config.clone(), config: config.clone(),
mgr, repo,
db, db,
}) })
} }

2
server/src/repo/actor.rs
View File

@ -1,4 +1,4 @@
use super::{archive, package, RepoHandle}; use super::{archive, package, Handle};
use crate::db; use crate::db;
use std::{ use std::{

6
server/src/repo/handle.rs
View File

@ -17,16 +17,16 @@ use tokio::{
}; };
#[derive(Clone)] #[derive(Clone)]
pub struct RepoHandle { pub struct Handle {
state: Arc<RepoSharedState>, state: Arc<RepoSharedState>,
} }
impl RepoHandle { impl Handle {
pub fn start( pub fn start(
repos_dir: impl AsRef<Path>, repos_dir: impl AsRef<Path>,
conn: DbConn, conn: DbConn,
actors: u32,
rt: runtime::Handle, rt: runtime::Handle,
actors: u32,
) -> crate::Result<Self> { ) -> crate::Result<Self> {
std::fs::create_dir_all(repos_dir.as_ref())?; std::fs::create_dir_all(repos_dir.as_ref())?;

110
server/src/repo/mod.rs
View File

@ -5,7 +5,7 @@ mod manager;
pub mod package; pub mod package;
pub use actor::{RepoActor, RepoCommand, RepoSharedState}; pub use actor::{RepoActor, RepoCommand, RepoSharedState};
pub use handle::RepoHandle; pub use handle::Handle;
pub use manager::RepoMgr; pub use manager::RepoMgr;
use crate::FsConfig; use crate::FsConfig;
@ -53,30 +53,31 @@ async fn get_file(
Path((distro, repo, arch, file_name)): Path<(String, String, String, String)>, Path((distro, repo, arch, file_name)): Path<(String, String, String, String)>,
req: Request<Body>, req: Request<Body>,
) -> crate::Result<impl IntoResponse> { ) -> crate::Result<impl IntoResponse> {
if let Some(repo_id) = global.mgr.get_repo(&distro, &repo).await? { Ok(StatusCode::NOT_FOUND)
match global.config.fs { //if let Some(repo_id) = global.mgr.get_repo(&distro, &repo).await? {
FsConfig::Local { data_dir } => { // match global.config.fs {
let repo_dir = data_dir.join("repos").join(repo_id.to_string()); // FsConfig::Local { data_dir } => {
// let repo_dir = data_dir.join("repos").join(repo_id.to_string());
let file_name = if file_name == format!("{}.db", repo) //
|| file_name == format!("{}.db.tar.gz", repo) // let file_name = if file_name == format!("{}.db", repo)
{ // || file_name == format!("{}.db.tar.gz", repo)
format!("{}.db.tar.gz", arch) // {
} else if file_name == format!("{}.files", repo) // format!("{}.db.tar.gz", arch)
|| file_name == format!("{}.files.tar.gz", repo) // } else if file_name == format!("{}.files", repo)
{ // || file_name == format!("{}.files.tar.gz", repo)
format!("{}.files.tar.gz", arch) // {
} else { // format!("{}.files.tar.gz", arch)
file_name // } else {
}; // file_name
// };
let path = repo_dir.join(file_name); //
Ok(ServeFile::new(path).oneshot(req).await) // let path = repo_dir.join(file_name);
} // Ok(ServeFile::new(path).oneshot(req).await)
} // }
} else { // }
Err(StatusCode::NOT_FOUND.into()) //} else {
} // Err(StatusCode::NOT_FOUND.into())
//}
} }
async fn post_package_archive( async fn post_package_archive(
@ -84,46 +85,49 @@ async fn post_package_archive(
Path((distro, repo)): Path<(String, String)>, Path((distro, repo)): Path<(String, String)>,
body: Body, body: Body,
) -> crate::Result<StatusCode> { ) -> crate::Result<StatusCode> {
let mut body = StreamReader::new(body.into_data_stream().map_err(std::io::Error::other)); Ok(StatusCode::NOT_FOUND)
let repo = global.mgr.get_or_create_repo(&distro, &repo).await?; //let mut body = StreamReader::new(body.into_data_stream().map_err(std::io::Error::other));
let [tmp_path] = global.mgr.random_file_paths(); //let repo = global.mgr.get_or_create_repo(&distro, &repo).await?;
//let [tmp_path] = global.mgr.random_file_paths();
let mut tmp_file = tokio::fs::File::create(&tmp_path).await?; //
tokio::io::copy(&mut body, &mut tmp_file).await?; //let mut tmp_file = tokio::fs::File::create(&tmp_path).await?;
//tokio::io::copy(&mut body, &mut tmp_file).await?;
global.mgr.queue_pkg(repo, tmp_path).await; //
//global.mgr.queue_pkg(repo, tmp_path).await;
Ok(StatusCode::ACCEPTED) //
//Ok(StatusCode::ACCEPTED)
} }
async fn delete_repo( async fn delete_repo(
State(global): State<crate::Global>, State(global): State<crate::Global>,
Path((distro, repo)): Path<(String, String)>, Path((distro, repo)): Path<(String, String)>,
) -> crate::Result<StatusCode> { ) -> crate::Result<StatusCode> {
if let Some(repo) = global.mgr.get_repo(&distro, &repo).await? { Ok(StatusCode::NOT_FOUND)
global.mgr.remove_repo(repo).await?; //if let Some(repo) = global.mgr.get_repo(&distro, &repo).await? {
// global.mgr.remove_repo(repo).await?;
tracing::info!("Removed repository {repo}"); //
// tracing::info!("Removed repository {repo}");
Ok(StatusCode::OK) //
} else { // Ok(StatusCode::OK)
Ok(StatusCode::NOT_FOUND) //} else {
} // Ok(StatusCode::NOT_FOUND)
//}
} }
async fn delete_arch_repo( async fn delete_arch_repo(
State(global): State<crate::Global>, State(global): State<crate::Global>,
Path((distro, repo, arch)): Path<(String, String, String)>, Path((distro, repo, arch)): Path<(String, String, String)>,
) -> crate::Result<StatusCode> { ) -> crate::Result<StatusCode> {
if let Some(repo) = global.mgr.get_repo(&distro, &repo).await? { Ok(StatusCode::NOT_FOUND)
global.mgr.remove_repo_arch(repo, &arch).await?; //if let Some(repo) = global.mgr.get_repo(&distro, &repo).await? {
// global.mgr.remove_repo_arch(repo, &arch).await?;
tracing::info!("Removed architecture '{arch}' from repository {repo}"); //
// tracing::info!("Removed architecture '{arch}' from repository {repo}");
Ok(StatusCode::OK) //
} else { // Ok(StatusCode::OK)
Ok(StatusCode::NOT_FOUND) //} else {
} // Ok(StatusCode::NOT_FOUND)
//}
//if let Some(mgr) = global.mgr.get_mgr(&distro).await { //if let Some(mgr) = global.mgr.get_mgr(&distro).await {
// let repo_removed = mgr.remove_repo_arch(&repo, &arch).await?; // let repo_removed = mgr.remove_repo_arch(&repo, &arch).await?;
// //