Merge branch 'dev'

Author: LgnMs

.cargo/config.toml

@@ -0,0 +1,2 @@
+[build]
+rustflags = ["--cfg", "tokio_unstable"]

.env

@@ -1,6 +1,6 @@
-DATABASE_URL=postgres://postgres:[email protected].40.3:5432/data_process_web
-HOST=127.0.0.1
+DATABASE_URL=mysql://root:[email protected].10.74:3306/data_process_web
+HOST=0.0.0.0
 PORT=18000
-CACHE_DATABASE_URL=postgres://postgres:[email protected].40.3:5432/data_process_cache
+CACHE_DATABASE_URL=mysql://root:[email protected].10.74:3306/data_process_cache
 JWT_SECRET=data_process
 LOG_LEVEL=DEBUG

.zed/settings.json

@@ -7,6 +7,7 @@ use reqwest::{
     header::{self, HeaderName, HeaderValue},
     Method,
 };
+use sea_orm::{DbBackend, Statement};
 use serde::{Deserialize, Serialize};
 use serde_json::Value;
 use tracing::{debug, error};
@@ -156,8 +157,6 @@ impl Serde for Http {
                 ) {
                     Err(_) => {}
                     Ok(x) => {
-                        let a = x.to_string();
-                        println!("a {a}");
                         self.data = x;
                     }
                 }
@@ -206,40 +205,72 @@ pub fn generate_sql_list(template_sql: &str, data: &Value) -> Result<Vec<String>
     }
 
     let mut key_vec = vec![];
+    let mut rel_key_vec = vec![];
     let mut i = 0;
+    let char_size1 = "$".as_bytes().len();
+    let char_size2 = "'".as_bytes().len();
+    let char_size3 = "}".as_bytes().len();
+
     while i < temp_index_vec.len() {
-        let one_index = temp_index_vec[i].0 - "$".as_bytes().len(); // 取"{"前$的字节索引
-        let two_index = temp_index_vec[i + 1].0;
+        rel_key_vec
+            .push(template_sql[temp_index_vec[i].0 + 1..temp_index_vec[i + 1].0].to_string());
+
+        let mut one_index = temp_index_vec[i].0 - char_size1; // 取"{"前$的字节索引
+        let mut two_index = temp_index_vec[i + 1].0;
+
+        let pre_one = &template_sql[(one_index - char_size2)..one_index];
+        let post_two =
+            &template_sql[(two_index + char_size3)..(two_index + char_size3 + char_size2)];
 
-        key_vec.push(template_sql[one_index..two_index + "}".as_bytes().len()].to_string());
+        if pre_one == "'" && post_two == "'" {
+            one_index -= char_size1;
+            two_index += char_size2;
+        }
+
+        key_vec.push(template_sql[one_index..two_index + char_size3].to_string());
 
         i += 2;
     }
+    let mut template_sql = template_sql.to_string();
 
-    let mut result_vec: Vec<String> = vec![];
+    for item in &key_vec {
+        // template_sql = template_sql.replace(item, format!("${}", index + 1).as_str());
+        template_sql = template_sql.replace(item, "?");
+    }
 
-    for key in key_vec {
-        let rel_key = &key[2..key.len() - 1];
-        let value = find_value(rel_key, data, true)
-            .map_err(|err| anyhow!("{err} 未在rel_key: {rel_key} data:{}中找到数据", data))?;
+    let mut result_vec: Vec<String> = vec![];
+    let mut result_values: Vec<Vec<String>> = vec![];
+    for key in &rel_key_vec {
+        // let rel_key = &key[2..key.len() - 1];
+        let value = find_value(key, data, true)
+            .map_err(|err| anyhow!("{err} 未在rel_key: {key} data:{}中找到数据", data))?;
         if let Some(list) = value.as_array() {
             for i in 0..list.len() {
-                let item: &str;
+                let mut item;
                 let temp_string = list[i].to_string();
                 if let Some(x) = list[i].as_str() {
-                    item = x;
+                    item = x.to_string();
                 } else {
-                    item = temp_string.as_str();
+                    item = temp_string;
                 }
-
-                if result_vec.get(i).is_none() {
-                    result_vec.push(template_sql.replace(key.as_str(), item));
+                // TIPS 做这个转换是为了防止值影响以;来切割SQL语句的方法 查看：crates/process_web/src/service/collect_config_service.rs 219行
+                item = item.replace(';', r#"\:"#);
+                if result_values.get(i).is_none() {
+                    result_values.push(vec![item.to_string()]);
                 } else {
-                    result_vec[i] = result_vec[i].replace(key.as_str(), item);
+                    result_values[i].push(item.to_string());
                 }
             }
         }
     }
+    for values in result_values {
+        let sql = Statement::from_sql_and_values(
+            DbBackend::MySql,
+            template_sql.clone(),
+            values.iter().map(|x| x.into()),
+        );
+        result_vec.push(sql.to_string());
+    }
 
     Ok(result_vec)
 }

Cargo.lock

@@ -10,6 +10,7 @@ sea-orm = { version = "0.12", features = [ "sqlx-postgres", "sqlx-mysql", "runti
 anyhow = "1.0.75"
 dotenvy = "0.15.7"
 tokio = { version = "1.34.0", features = ["full"] }
+tokio-util = "0.7.10"
 tracing = "0.1"
 tracing-subscriber = "0.3.0"
 tracing-appender = "0.2"

crates/process_core/src/http.rs

@@ -1,32 +1,34 @@
 pub use sea_orm_migration::prelude::*;
 
-mod m20240119_000002_create_collect_log_table;
-mod m20240119_030002_create_sync_log_table;
 mod m20240119_000001_create_collect_config_table;
+mod m20240119_000002_create_collect_log_table;
 mod m20240119_023953_create_sync_config_table;
+mod m20240119_030002_create_sync_log_table;
 mod m20240226_015923_create_data_source_list;
 mod m20240227_022320_create_data_sharing_config_table;
 mod m20240304_063328_create_sharing_request_log;
 mod m20240319_061549_update_data_sharing_config_table;
 mod m20240327_063820_update_sharing_request_log;
 mod m20240402_033637_update_collect_config_table;
+mod m20240408_033448_update_collect_log_table;
 
 pub struct Migrator;
 
 #[async_trait::async_trait]
 impl MigratorTrait for Migrator {
     fn migrations() -> Vec<Box<dyn MigrationTrait>> {
         vec![
-            Box::new(m20240119_000001_create_collect_config_table::Migration),
             Box::new(m20240119_000002_create_collect_log_table::Migration),
-            Box::new(m20240119_023953_create_sync_config_table::Migration),
             Box::new(m20240119_030002_create_sync_log_table::Migration),
+            Box::new(m20240119_000001_create_collect_config_table::Migration),
+            Box::new(m20240119_023953_create_sync_config_table::Migration),
             Box::new(m20240226_015923_create_data_source_list::Migration),
             Box::new(m20240227_022320_create_data_sharing_config_table::Migration),
             Box::new(m20240304_063328_create_sharing_request_log::Migration),
             Box::new(m20240319_061549_update_data_sharing_config_table::Migration),
             Box::new(m20240327_063820_update_sharing_request_log::Migration),
             Box::new(m20240402_033637_update_collect_config_table::Migration),
+            Box::new(m20240408_033448_update_collect_log_table::Migration),
         ]
     }
 }

crates/process_web/Cargo.toml

@@ -77,6 +77,7 @@ pub enum CollectLog {
     Id,
     RunningLog,
     CollectConfigId,
+    TaskId,
     Status,
     UpdateTime,
     CreateTime,

crates/process_web/migration/src/lib.rs

@@ -8,7 +8,6 @@ pub struct Migration;
 #[async_trait::async_trait]
 impl MigrationTrait for Migration {
     async fn up(&self, manager: &SchemaManager) -> Result<(), DbErr> {
-
         manager
             .alter_table(
                 Table::alter()
@@ -18,9 +17,8 @@ impl MigrationTrait for Migration {
                             .json()
                             .comment(r#"数据库列配置2"#),
                     )
-                    .to_owned()
+                    .to_owned(),
             )
             .await
     }
-
 }

crates/process_web/migration/src/m20240119_000002_create_collect_log_table.rs

@@ -0,0 +1,24 @@
+use sea_orm_migration::prelude::*;
+
+use crate::m20240119_000002_create_collect_log_table::CollectLog;
+
+#[derive(DeriveMigrationName)]
+pub struct Migration;
+
+#[async_trait::async_trait]
+impl MigrationTrait for Migration {
+    async fn up(&self, manager: &SchemaManager) -> Result<(), DbErr> {
+        manager
+            .alter_table(
+                Table::alter()
+                    .table(CollectLog::Table)
+                    .add_column(
+                        ColumnDef::new(CollectLog::TaskId)
+                            .string()
+                            .comment(r#"正在执行的任务的id"#),
+                    )
+                    .to_owned(),
+            )
+            .await
+    }
+}

crates/process_web/migration/src/m20240402_033637_update_collect_config_table.rs

@@ -7,12 +7,16 @@ use axum::{
     Json, Router,
 };
 use serde::Deserialize;
+use tracing::debug;
 use ts_rs::TS;
+use uuid::Uuid;
 
-use crate::api::common::{AppError, AppState, PaginationPayload, ResJson, ResJsonWithPagination};
+use crate::api::common::{
+    AppError, AppState, LogTask, PaginationPayload, ResJson, ResJsonWithPagination,
+};
 use crate::entity::collect_config::Model;
 use crate::service::collect_config_service::CollectConfigService;
-use crate::{bool_response, data_response, pagination_response, res_template_ok};
+use crate::{bool_response, data_response, pagination_response};
 
 pub fn set_routes() -> Router<Arc<AppState>> {
     Router::new()
@@ -57,6 +61,7 @@ async fn list(
 
     pagination_response!(res, payload.current, payload.page_size)
 }
+
 async fn add(
     state: State<Arc<AppState>>,
     Json(payload): Json<Model>,
@@ -89,11 +94,23 @@ pub async fn execute(
 ) -> Result<ResJson<bool>, AppError> {
     let data = CollectConfigService::find_by_id(&state.conn, id).await?;
 
+    let task_id = Uuid::new_v4().simple();
+    let st = state.clone();
+    let log_task = LogTask::new();
+    let cloned_token = log_task.token.clone();
+    let mut task = state.log_task.write().await;
+    task.insert(task_id, log_task);
+    drop(task);
+
     tokio::task::spawn(async move {
-        CollectConfigService::execute_task(&state, &data).await;
+        tokio::select! {
+            _ = cloned_token.cancelled() => {
+                // The token was cancelled, task can shut down
+                debug!("cloned_token {cloned_token:?}");
+            }
+            _ = CollectConfigService::execute_task(&st, &data, task_id) => {}
+        }
     });
-    // https://docs.rs/tokio/1.35.1/tokio/task/index.html#yield_now
-    // tokio::task::yield_now().await;
-
-    Ok(Json(res_template_ok!(Some(true))))
+    let res: anyhow::Result<bool> = Ok(true);
+    bool_response!(res)
 }

crates/process_web/migration/src/m20240408_033448_update_collect_log_table.rs

@@ -7,20 +7,23 @@ use axum::{
 };
 use serde::Deserialize;
 use std::sync::Arc;
+use tracing::error;
 use ts_rs::TS;
+use uuid::Uuid;
 
 use crate::api::common::{AppError, AppState, PaginationPayload, ResJson, ResJsonWithPagination};
-use crate::entity::collect_log::Model;
+use crate::entity::collect_log::{self, Model};
 use crate::service::collect_log_service::CollectLogService;
 use crate::{bool_response, data_response, pagination_response};
 
 pub fn set_routes() -> Router<Arc<AppState>> {
-     Router::new()
-        .route("/find_by_id/:id", post(find_by_id))
+    Router::new()
+        .route("/find_by_id/:id", get(find_by_id))
         .route("/list", post(list))
         .route("/add", post(add))
         .route("/update_by_id/:id", post(update_by_id))
         .route("/delete/:id", get(del))
+        .route("/stop_task/:id", get(stop_task))
 }
 
 async fn find_by_id(
@@ -81,3 +84,29 @@ async fn del(state: State<Arc<AppState>>, Path(id): Path<i32>) -> Result<ResJson
 
     bool_response!(res)
 }
+
+pub async fn stop_task(
+    state: State<Arc<AppState>>,
+    Path(id): Path<String>,
+) -> Result<ResJson<bool>, AppError> {
+    let log_task_id = Uuid::parse_str(id.as_str())?.simple();
+
+    let log_id = state.stop_log_task(log_task_id).await;
+    if let Some(log_id) = log_id {
+        if let Err(err) = CollectLogService::update_by_id(
+            &state.conn,
+            log_id,
+            collect_log::Model {
+                status: 5,
+                running_log: "用户手动停止".to_string(),
+                ..Default::default()
+            },
+        )
+        .await
+        {
+            error!("{}", err);
+        }
+    }
+
+    bool_response!(anyhow::Ok::<bool>(true))
+}