Refactor judger code

judger/Cargo.toml

@@ -43,12 +43,4 @@ chrono = { version = "0.4", features = ["serde"] }
 anyhow = "1"
 thiserror = "1"
 
-uuid = { version = "1.4", features = ["serde", "v4"] }
-
-[[bin]]
-name ="judger-cli"
-path ="src/cli/main.rs"
-
-[[bin]]
-name ="judger-server"
-path ="src/server/main.rs"
+uuid = { version = "1.4", features = ["serde", "v4"] }

judger/src/server/client/httpclient.rs → judger/src/agent/http.rs

@@ -8,9 +8,7 @@ impl HttpClient {
         let client = reqwest::Client::new();
         Self { client, base_url }
     }
-    pub fn _get(&self, path: String) -> reqwest::RequestBuilder {
-        self.client.get(format!("{}{}", self.base_url, path))
-    }
+
     pub fn post(&self, path: String) -> reqwest::RequestBuilder {
         self.client.post(format!("{}{}", self.base_url, path))
     }

judger/src/agent/mod.rs

@@ -0,0 +1,3 @@
+pub mod http;
+pub mod rclone;
+pub mod platform;

judger/src/agent/platform/mod.rs

@@ -0,0 +1,99 @@
+use super::http::HttpClient;
+use judge_core::{compiler::Language, judge::result::JudgeResultInfo};
+
+pub struct PlatformClient {
+    client: HttpClient,
+}
+
+impl PlatformClient {
+    pub fn new(base_url: String) -> Self {
+        Self {
+            client: HttpClient::new(base_url),
+        }
+    }
+
+    pub async fn pick_task(&self) -> Result<JudgeTask, anyhow::Error> {
+        pick_task(&self.client).await
+    }
+
+    pub async fn report_task(
+        &self,
+        stream_id: &str,
+        results: Vec<JudgeResultInfo>,
+    ) -> Result<(), anyhow::Error> {
+        report_task(&self.client, stream_id, results).await
+    }
+}
+
+#[derive(Deserialize, Debug)]
+pub struct JudgeTask {
+    #[serde(rename = "submissionUID")]
+    pub submission_uid: String,
+    #[serde(rename = "problemSlug")]
+    pub problem_slug: String,
+    pub code: String,
+    pub language: Language,
+    #[serde(rename = "redisStreamID")]
+    pub redis_stream_id: String,
+}
+#[derive(Serialize)]
+struct PickTaskBody {
+    consumer: String,
+}
+#[derive(Deserialize, Debug)]
+struct PickTaskResponse {
+    task: JudgeTask,
+}
+
+async fn pick_task(client: &HttpClient) -> Result<JudgeTask, anyhow::Error> {
+    let pick_url = "/task/pick";
+    let body = PickTaskBody {
+        consumer: "".to_string(),
+    };
+    let response = client.post(pick_url.to_string()).json(&body).send().await?;
+
+    match response.status() {
+        reqwest::StatusCode::OK => Ok(response.json::<PickTaskResponse>().await?.task),
+        _ => Err(anyhow::anyhow!("Queue is empty")),
+    }
+}
+
+#[derive(Serialize)]
+struct ReportTaskBody {
+    consumer: String,
+    stream_id: String,
+    verdict_json: String,
+}
+#[derive(Deserialize, Debug)]
+struct ReportTaskResponse {
+    message: String,
+}
+
+async fn report_task(
+    client: &HttpClient,
+    stream_id: &str,
+    results: Vec<JudgeResultInfo>,
+) -> Result<(), anyhow::Error> {
+    let report_url = "/task/report";
+    let body = ReportTaskBody {
+        consumer: "".to_string(),
+        stream_id: stream_id.to_owned(),
+        verdict_json: serde_json::to_string(&results).unwrap(),
+    };
+    let response = client
+        .post(report_url.to_string())
+        .json(&body)
+        .send()
+        .await?;
+
+    match response.status() {
+        reqwest::StatusCode::OK => {
+            log::debug!(
+                "Report message: {:?}",
+                response.json::<ReportTaskResponse>().await?.message
+            );
+            Ok(())
+        }
+        _ => Err(anyhow::anyhow!("Report Failed")),
+    }
+}

judger/src/agent/rclone.rs

@@ -0,0 +1,41 @@
+use anyhow::{self, Error};
+use std::path::PathBuf;
+use std::process::Command;
+
+pub struct RcloneClient {
+    config_path: PathBuf,
+}
+
+impl RcloneClient {
+    pub fn new(config_path: PathBuf) -> Self {
+        Self { config_path }
+    }
+
+    pub fn is_avaliable(&self) -> bool {
+        let status = Command::new("rclone")
+            .arg("--config")
+            .arg(format!("{}", self.config_path.to_string_lossy()))
+            .arg("ls")
+            .arg("minio:")
+            .status()
+            .expect("Failed to rclone");
+
+        status.success()
+    }
+
+    pub fn sync_bucket(&self, bucket_name: &str, target_dir: PathBuf) -> Result<(), Error> {
+        let status = Command::new("rclone")
+            .arg("--config")
+            .arg(format!("{}", self.config_path.to_string_lossy()))
+            .arg("sync")
+            .arg(format!("minio:{}", bucket_name))
+            .arg(format!("{}", target_dir.to_string_lossy()))
+            .status()
+            .expect("Failed to rclone");
+        if status.success() {
+            Ok(())
+        } else {
+            Err(anyhow::anyhow!("rclone sync failed, please check config."))
+        }
+    }
+}

judger/src/server/environment/mod.rs → judger/src/env/mod.rs

@@ -12,9 +12,12 @@ pub struct JudgeServerOpt {
     #[structopt(env = "PORT", default_value = "8080")]
     pub port: u16,
 
-    #[structopt(long, default_value = "data/dev-problem-package")]
+    #[structopt(long, default_value = "data/problem-package")]
     pub problem_package_dir: PathBuf,
 
+    #[structopt(long, default_value = "data/rclone.conf")]
+    pub rclone_config: PathBuf,
+
     #[structopt(env = "BASE_URL", default_value = "http://localhost:8080/api/v1/judge")]
     pub base_url: String,
 

judger/src/server/error.rs → judger/src/error.rs

@@ -2,7 +2,6 @@
 
 use actix_web::{HttpResponse, ResponseError};
 use judge_core::error::JudgeCoreError;
-use judger::service::error::JudgeServiceError;
 
 #[derive(Debug, thiserror::Error)]
 pub enum ServiceError {
@@ -23,7 +22,6 @@ pub enum ServiceError {
 #[derive(Debug)]
 pub enum ClientError {
     InternalError(anyhow::Error),
-    PackageError(JudgeServiceError),
 }
 
 #[derive(Serialize)]

judger/src/server/service/mod.rs → judger/src/handler/mod.rs

@@ -1,5 +1,4 @@
 mod greet;
-mod judge;
 pub mod state;
 
 use actix_web::web;
@@ -15,7 +14,6 @@ pub struct ApiDoc;
 pub fn route(cfg: &mut web::ServiceConfig) {
     cfg.service(
         web::scope("/api/v1")
-            .configure(judge::route)
             .service(greet::greet)
             .configure(state::route),
     )
@@ -25,10 +23,6 @@ pub fn route(cfg: &mut web::ServiceConfig) {
                 utoipa_swagger_ui::Url::new("root", "/api-docs/openapi.json"),
                 ApiDoc::openapi(),
             ),
-            (
-                utoipa_swagger_ui::Url::new("judge", "/api-docs/judge.json"),
-                judge::JudgeApiDoc::openapi(),
-            ),
             (
                 utoipa_swagger_ui::Url::new("state", "/api-docs/state.json"),
                 state::StateApiDoc::openapi(),

judger/src/main.rs

@@ -0,0 +1,50 @@
+mod agent;
+mod env;
+mod error;
+mod handler;
+mod worker;
+
+#[macro_use]
+extern crate serde_derive;
+extern crate lazy_static;
+
+use actix_web::{web::Data, App, HttpServer};
+
+#[actix_web::main] // or #[tokio::main]
+async fn main() -> std::io::Result<()> {
+    let opt = env::load_option();
+    env::setup_logger();
+
+    // TODO: Send heartbeat here to a remote host
+
+    let worker = match worker::JudgeWorker::new(
+        opt.base_url,
+        opt.interval as u64,
+        opt.rclone_config,
+        opt.problem_package_dir.clone(),
+    ) {
+        Ok(maybe_worker) => {
+            if let Some(worker) = maybe_worker {
+                worker
+            } else {
+                log::error!("Failed to create worker");
+                return Ok(());
+            }
+        }
+        Err(e) => {
+            log::error!("Failed to create worker: {:?}", e);
+            return Ok(());
+        }
+    };
+    tokio::spawn(async move { worker.run().await });
+
+    HttpServer::new(move || {
+        App::new()
+            .wrap(actix_web::middleware::Logger::default())
+            .app_data(Data::new(opt.problem_package_dir.clone()))
+            .configure(handler::route)
+    })
+    .bind(("0.0.0.0", opt.port))?
+    .run()
+    .await
+}