diff --git a/Cargo.toml b/Cargo.toml index c57e0f21c..9de04213d 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -12,6 +12,7 @@ members = [ "neptune", "aries", "saturn", + "taurus", "lunar/src-tauri", "atlas", ] @@ -31,6 +32,7 @@ gemini = { path = "gemini" } vault = { path = "vault" } neptune = { path = "neptune" } saturn = { path = "saturn" } +taurus = { path = "taurus" } mega = { path = "mega" } anyhow = "1.0.86" serde = {version = "1.0.203", features = ["derive"]} diff --git a/firedbg/target.json b/firedbg/target.json new file mode 100644 index 000000000..636ddd0dd --- /dev/null +++ b/firedbg/target.json @@ -0,0 +1,6 @@ +{ + "binaries": [], + "examples": [], + "integration_tests": [], + "unit_tests": [] +} \ No newline at end of file diff --git a/firedbg/version.toml b/firedbg/version.toml new file mode 100644 index 000000000..c59a771e6 --- /dev/null +++ b/firedbg/version.toml @@ -0,0 +1 @@ +firedbg_cli = "1.74.0" diff --git a/gateway/Cargo.toml b/gateway/Cargo.toml index cc53b8326..8ebf9c7b7 100644 --- a/gateway/Cargo.toml +++ b/gateway/Cargo.toml @@ -18,6 +18,7 @@ gemini = { workspace = true } vault = { workspace = true } venus = { workspace = true } mercury = { workspace = true } +taurus = { workspace = true } anyhow = { workspace = true } axum = { workspace = true } axum-server = { version = "0.7", features = ["tls-rustls"] } @@ -37,7 +38,7 @@ tower-http = { workspace = true, features = [ "trace", "decompression-full", ] } -axum-extra = { workspace = true, features = ["typed-header"]} +axum-extra = { workspace = true, features = ["typed-header"] } tokio = { workspace = true, features = ["net"] } tokio-stream = { workspace = true } async-stream = { workspace = true } diff --git a/gateway/src/api/api_router.rs b/gateway/src/api/api_router.rs index c5f91c29e..3fff4acda 100644 --- a/gateway/src/api/api_router.rs +++ b/gateway/src/api/api_router.rs @@ -6,6 +6,7 @@ use axum::{ Json, Router, }; +use taurus::event::api_request::{ApiRequestEvent, ApiType}; use ceres::model::{ create_file::CreateFileInfo, publish_path::PublishPathInfo, @@ -39,6 +40,7 @@ async fn get_blob_object( Query(query): Query, state: State, ) -> Result>, (StatusCode, String)> { + ApiRequestEvent::notify(ApiType::Blob, &state.0.context.config); let res = state .api_handler(query.path.clone().into()) .await @@ -80,6 +82,7 @@ async fn create_file( state: State, Json(json): Json, ) -> Result>, (StatusCode, String)> { + ApiRequestEvent::notify(ApiType::CreateFile, &state.0.context.config); let res = state .api_handler(json.path.clone().into()) .await @@ -96,6 +99,7 @@ async fn get_latest_commit( Query(query): Query, state: State, ) -> Result, (StatusCode, String)> { + ApiRequestEvent::notify(ApiType::LastestCommit, &state.0.context.config); let res = state .api_handler(query.path.clone().into()) .await @@ -109,6 +113,7 @@ async fn get_tree_info( Query(query): Query, state: State, ) -> Result>>, (StatusCode, String)> { + ApiRequestEvent::notify(ApiType::TreeInfo, &state.0.context.config); let res = state .api_handler(query.path.clone().into()) .await @@ -125,6 +130,7 @@ async fn get_tree_commit_info( Query(query): Query, state: State, ) -> Result>>, (StatusCode, String)> { + ApiRequestEvent::notify(ApiType::CommitInfo, &state.0.context.config); let res = state .api_handler(query.path.clone().into()) .await @@ -141,6 +147,7 @@ async fn publish_path_to_repo( state: State, Json(json): Json, ) -> Result>, (StatusCode, String)> { + ApiRequestEvent::notify(ApiType::Publish, &state.0.context.config); let res = state .api_handler(json.path.clone().into()) .await diff --git a/gateway/src/api/mr_router.rs b/gateway/src/api/mr_router.rs index 35d8c9f26..18d843a81 100644 --- a/gateway/src/api/mr_router.rs +++ b/gateway/src/api/mr_router.rs @@ -10,6 +10,7 @@ use axum::{ use ceres::model::mr::{MRDetail, MrInfoItem}; use common::model::CommonResult; +use taurus::event::api_request::{ApiRequestEvent, ApiType}; use crate::api::ApiServiceState; pub fn routers() -> Router { @@ -24,11 +25,14 @@ async fn merge( Path(mr_id): Path, state: State, ) -> Result>, (StatusCode, String)> { + ApiRequestEvent::notify(ApiType::MergeRequest, &state.0.context.config); + let res = state.monorepo().merge_mr(mr_id).await; let res = match res { Ok(_) => CommonResult::success(None), Err(err) => CommonResult::failed(&err.to_string()), }; + ApiRequestEvent::notify(ApiType::MergeDone, &state.0.context.config); Ok(Json(res)) } @@ -36,6 +40,7 @@ async fn get_mr_list( Query(query): Query>, state: State, ) -> Result>>, (StatusCode, String)> { + ApiRequestEvent::notify(ApiType::MergeList, &state.0.context.config); let status = query.get("status").unwrap(); let res = state.monorepo().mr_list(status).await; let res = match res { @@ -49,6 +54,7 @@ async fn mr_detail( Path(mr_id): Path, state: State, ) -> Result>>, (StatusCode, String)> { + ApiRequestEvent::notify(ApiType::MergeDetail, &state.0.context.config); let res = state.monorepo().mr_detail(mr_id).await; let res = match res { Ok(data) => CommonResult::success(Some(data)), @@ -61,6 +67,7 @@ async fn get_mr_files( Path(mr_id): Path, state: State, ) -> Result>>, (StatusCode, String)> { + ApiRequestEvent::notify(ApiType::MergeFiles, &state.0.context.config); let res = state.monorepo().mr_tree_files(mr_id).await; let res = match res { Ok(data) => CommonResult::success(Some(data)), diff --git a/jupiter/callisto/src/lib.rs b/jupiter/callisto/src/lib.rs index ebd858ac6..0451807a1 100644 --- a/jupiter/callisto/src/lib.rs +++ b/jupiter/callisto/src/lib.rs @@ -26,3 +26,4 @@ pub mod mega_tree; pub mod raw_blob; pub mod ztm_node; pub mod ztm_repo_info; +pub mod mq_storage; diff --git a/jupiter/callisto/src/mq_storage.rs b/jupiter/callisto/src/mq_storage.rs new file mode 100644 index 000000000..148d731e2 --- /dev/null +++ b/jupiter/callisto/src/mq_storage.rs @@ -0,0 +1,19 @@ +//! `SeaORM` Entity. Generated by sea-orm-codegen 0.11.3 + +use sea_orm::entity::prelude::*; + +#[derive(Clone, Debug, PartialEq, DeriveEntityModel, Eq)] +#[sea_orm(table_name = "mq_storage")] +pub struct Model { + #[sea_orm(primary_key, auto_increment = false)] + pub id: i64, + pub category: String, + pub create_time: DateTime, + #[sea_orm(column_type = "Text", nullable)] + pub content: Option, +} + +#[derive(Copy, Clone, Debug, EnumIter, DeriveRelation)] +pub enum Relation {} + +impl ActiveModelBehavior for ActiveModel {} diff --git a/jupiter/src/context.rs b/jupiter/src/context.rs index 8d98df037..27e2eff41 100644 --- a/jupiter/src/context.rs +++ b/jupiter/src/context.rs @@ -3,8 +3,7 @@ use std::sync::Arc; use common::config::Config; use crate::storage::{ - git_db_storage::GitDbStorage, init::database_connection, lfs_storage::LfsStorage, - mega_storage::MegaStorage, ztm_storage::ZTMStorage, + git_db_storage::GitDbStorage, init::database_connection, lfs_storage::LfsStorage, mega_storage::MegaStorage, mq_storage::MQStorage, ztm_storage::ZTMStorage }; #[derive(Clone)] @@ -34,6 +33,7 @@ pub struct Service { pub git_db_storage: Arc, pub lfs_storage: Arc, pub ztm_storage: Arc, + pub mq_storage: Arc, } impl Service { @@ -48,6 +48,7 @@ impl Service { ), lfs_storage: Arc::new(LfsStorage::new(connection.clone()).await), ztm_storage: Arc::new(ZTMStorage::new(connection.clone()).await), + mq_storage: Arc::new(MQStorage::new(connection.clone()).await), } } @@ -61,6 +62,7 @@ impl Service { git_db_storage: Arc::new(GitDbStorage::mock()), lfs_storage: Arc::new(LfsStorage::mock()), ztm_storage: Arc::new(ZTMStorage::mock()), + mq_storage: Arc::new(MQStorage::mock()), }) } } diff --git a/jupiter/src/storage/mod.rs b/jupiter/src/storage/mod.rs index 204e4a9fa..59ac5d8f7 100644 --- a/jupiter/src/storage/mod.rs +++ b/jupiter/src/storage/mod.rs @@ -3,6 +3,7 @@ pub mod git_fs_storage; pub mod init; pub mod lfs_storage; pub mod mega_storage; +pub mod mq_storage; pub mod ztm_storage; use async_trait::async_trait; diff --git a/jupiter/src/storage/mq_storage.rs b/jupiter/src/storage/mq_storage.rs new file mode 100644 index 000000000..0bbbed5dc --- /dev/null +++ b/jupiter/src/storage/mq_storage.rs @@ -0,0 +1,46 @@ +use std::sync::Arc; + +use callisto::mq_storage::*; +use sea_orm::{DatabaseConnection, EntityTrait, QueryOrder, QuerySelect}; + +use super::batch_save_model; + + +#[derive(Clone)] +pub struct MQStorage { + pub connection: Arc, +} + +impl MQStorage { + pub fn get_connection(&self) -> &DatabaseConnection { + &self.connection + } + + pub async fn new(connection: Arc) -> Self { + MQStorage { connection } + } + + pub fn mock() -> Self { + MQStorage { + connection: Arc::new(DatabaseConnection::default()), + } + } + + pub async fn save_messages(&self, msgs: Vec) { + if msgs.is_empty() { + return; + } + + let msgs: Vec = msgs.into_iter().map(|m| m.into()).collect(); + batch_save_model(self.get_connection(), msgs).await.unwrap(); + } + + pub async fn get_latest_message(&self) -> Option { + Entity::find() + .order_by_desc(Column::Id) + .limit(1) + .one(self.get_connection()) + .await + .unwrap() + } +} diff --git a/mega/Cargo.toml b/mega/Cargo.toml index c0eeabc55..7ac1e1537 100644 --- a/mega/Cargo.toml +++ b/mega/Cargo.toml @@ -16,6 +16,7 @@ path = "src/main.rs" gateway = { workspace = true } common = { workspace = true } ceres = { workspace = true } +taurus = { workspace = true } serde = { workspace = true, features = ["derive"] } tokio = { workspace = true, features = ["macros"] } clap = { workspace = true, features = ["derive"] } diff --git a/mega/src/commands/service/mod.rs b/mega/src/commands/service/mod.rs index 4d12d1fde..824e357ac 100644 --- a/mega/src/commands/service/mod.rs +++ b/mega/src/commands/service/mod.rs @@ -25,6 +25,9 @@ pub fn cli() -> Command { // It determines which subcommand was used and calls the appropriate function. #[tokio::main] pub(crate) async fn exec(config: Config, args: &ArgMatches) -> MegaResult { + use taurus::init::init_mq; + init_mq(&config).await; + let (cmd, subcommand_args) = match args.subcommand() { Some((cmd, args)) => (cmd, args), _ => { diff --git a/saturn/src/context.rs b/saturn/src/context.rs index cb010954d..49fbc6bea 100644 --- a/saturn/src/context.rs +++ b/saturn/src/context.rs @@ -16,6 +16,7 @@ pub struct AppContext { schema: Schema, } +#[allow(dead_code)] #[derive(Debug, Error)] pub enum ContextError { #[error("{0}")] @@ -34,6 +35,7 @@ pub enum ContextError { Json(#[from] serde_json::Error), } +#[allow(dead_code)] #[derive(Debug, Error)] pub enum Error { #[error("Authorization Denied")] @@ -43,6 +45,7 @@ pub enum Error { } impl AppContext { + #[allow(dead_code)] pub fn new( entities: EntityStore, schema_path: impl Into, @@ -78,6 +81,7 @@ impl AppContext { } } + #[allow(dead_code)] pub fn is_authorized( &self, principal: impl AsRef, diff --git a/saturn/src/entitystore.rs b/saturn/src/entitystore.rs index a27901281..8d8a0500b 100644 --- a/saturn/src/entitystore.rs +++ b/saturn/src/entitystore.rs @@ -18,6 +18,7 @@ pub struct EntityStore { } impl EntityStore { + #[allow(dead_code)] pub fn as_entities(&self, schema: &Schema) -> Entities { let users = self.users.values().map(|user| user.clone().into()); let repos = self.repos.values().map(|repo| repo.clone().into()); @@ -32,6 +33,7 @@ impl EntityStore { Entities::from_entities(all, Some(schema)).unwrap() } + #[allow(dead_code)] pub fn merge(&mut self, other: EntityStore) { self.users.extend(other.users); self.repos.extend(other.repos); diff --git a/sql/postgres/pg_20240205__init.sql b/sql/postgres/pg_20240205__init.sql index 0d73841cf..591646e67 100644 --- a/sql/postgres/pg_20240205__init.sql +++ b/sql/postgres/pg_20240205__init.sql @@ -243,4 +243,11 @@ CREATE TABLE IF NOT EXISTS "ztm_repo_info" ( "origin" VARCHAR(64), "update_time" BIGINT NOT NULL, "commit" VARCHAR(64) -); \ No newline at end of file +); + +CREATE TABLE IF NOT EXISTS "mq_storage" ( + "id" BIGINT PRIMARY KEY, + "category" VARCHAR(64), + "create_time" TIMESTAMP NOT NULL, + "content" TEXT +); diff --git a/sql/sqlite/sqlite_20240711_init.sql b/sql/sqlite/sqlite_20240711_init.sql index 106e88992..fa8df4d75 100644 --- a/sql/sqlite/sqlite_20240711_init.sql +++ b/sql/sqlite/sqlite_20240711_init.sql @@ -243,4 +243,11 @@ CREATE TABLE IF NOT EXISTS "ztm_repo_info" ( "origin" TEXT, "update_time" INTEGER NOT NULL, "commit" TEXT -); \ No newline at end of file +); + +CREATE TABLE IF NOT EXISTS "mq_storage" ( + "id" INTEGER PRIMARY KEY, + "category" TEXT, + "create_time" TIMESTAMP NOT NULL, + "content" TEXT +); diff --git a/taurus/Cargo.toml b/taurus/Cargo.toml new file mode 100644 index 000000000..5eb8d0725 --- /dev/null +++ b/taurus/Cargo.toml @@ -0,0 +1,23 @@ +[package] +name = "taurus" +version = "0.1.0" +edition = "2021" + +[lib] +name = "taurus" +path = "src/lib.rs" + +[dependencies] +common = { workspace = true } +jupiter = { workspace = true } +callisto = { workspace = true } + +axum = { workspace = true } +async-trait = { workspace = true } +tokio = { workspace = true, features = ["rt-multi-thread"]} +tracing = { workspace = true } +thiserror = { workspace = true } +serde = { workspace = true } +serde_json = { workspace = true } +chrono = { workspace = true } +crossbeam-channel = "0.5.10" diff --git a/taurus/README.md b/taurus/README.md new file mode 100644 index 000000000..1af45fb61 --- /dev/null +++ b/taurus/README.md @@ -0,0 +1,2 @@ +# Message Queue Module +This module offers mega the ability to send and handle specific events. diff --git a/taurus/src/cache.rs b/taurus/src/cache.rs new file mode 100644 index 000000000..cfbb1cf22 --- /dev/null +++ b/taurus/src/cache.rs @@ -0,0 +1,98 @@ +use std::{mem::swap, sync::{atomic::{AtomicBool, AtomicI64}, Arc, Mutex, OnceLock}, time::Duration}; + +use chrono::Utc; + +use crate::{event::Message, queue::{get_mq, MessageQueue}}; + +const FLUSH_INTERVAL: u64 = 10; + +// Lazy initialized static MessageCache instance. +pub fn get_mcache() -> &'static MessageCache { + static MC: OnceLock = OnceLock::new(); + MC.get_or_init(|| { + let mc = MessageCache::new(); + mc.start(); + + mc + }) +} + +// Automatically flush message cache into database +// eveny 10 seconds or 1024 message. +pub struct MessageCache { + inner: Arc>>, + bound_mq: &'static MessageQueue, + last_flush: Arc, + stop: Arc, +} + +impl MessageCache { + fn new() -> Self { + let now: chrono::DateTime = Utc::now(); + + MessageCache { + inner: Arc::new(Mutex::new(Vec::new())), + bound_mq: get_mq(), + last_flush: Arc::new(AtomicI64::new(now.timestamp_millis())), + stop: Arc::new(AtomicBool::new(false)) + } + } + + fn start(&self) { + let stop = self.stop.clone(); + tokio::spawn(async move { + loop { + if stop.load(std::sync::atomic::Ordering::Acquire) { + return + } + tokio::time::sleep(Duration::from_secs(FLUSH_INTERVAL)).await; + + instant_flush().await; + } + }); + } + + fn get_cache(&self) -> Vec { + let mut res = Vec::new(); + let inner = self.inner.clone(); + + let mut locked = inner.lock().unwrap(); + if locked.len() != 0 { + swap(locked.as_mut(), &mut res); + } + + res + } + + pub(crate) async fn add(&self, msg: Message) -> &Self { + let inner = self.inner.clone(); + let should_flush: bool; + { + let mut locked = inner.lock().unwrap(); + let l = locked.len(); + should_flush = l >= 1; + locked.push(msg); + } + + if should_flush { + instant_flush().await + } + + self + } +} + +pub async fn instant_flush() { + use callisto::mq_storage::Model; + + let mc = get_mcache(); + let st = mc.bound_mq.context.services.mq_storage.clone(); + let data = mc + .get_cache() + .into_iter().map(Into::::into) + .collect::>(); + st.save_messages(data).await; + + let now = Utc::now(); + mc.last_flush.to_owned().store(now.timestamp_millis(), std::sync::atomic::Ordering::Relaxed); +} diff --git a/taurus/src/event/api_request.rs b/taurus/src/event/api_request.rs new file mode 100644 index 000000000..d2323348d --- /dev/null +++ b/taurus/src/event/api_request.rs @@ -0,0 +1,99 @@ +use common::config::Config; +use serde::{Deserialize, Serialize}; +use async_trait::async_trait; + +use crate::{event::EventBase, event::EventType, queue::get_mq}; + +/// # Api Request Event +/// --- +/// This is a example event definition for using message queue. \ +/// +/// Your customized event should implement `EventBase` trait. \ +/// Then the event can be wrapped and put into message queue. \ +/// The message `id` and `create_time` will be attached to your event +/// and then wrapped as a `Message`. \ +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct ApiRequestEvent { + pub api: ApiType, + pub config: common::config::Config, +} + +#[derive(Debug, Serialize, Deserialize, Clone)] +pub enum ApiType { + // Common Api enum for api_routers + CreateFile, + LastestCommit, + CommitInfo, + TreeInfo, + Blob, + Publish, + + // Merge Api enum for mr_routers + MergeRequest, + MergeDone, + MergeList, + MergeDetail, + MergeFiles, +} + +impl std::fmt::Display for ApiRequestEvent { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + write!(f, "Api Request Event: {:?}", self.api) + } +} + +#[async_trait] +impl EventBase for ApiRequestEvent { + async fn process(&self) { + tracing::info!("Handling Api Request event: [{}]", &self); + } +} + +impl ApiRequestEvent { + // Create and enqueue this event. + pub fn notify(api: ApiType, config: &Config) { + get_mq().send(EventType::ApiRequest(ApiRequestEvent { + api, + config: config.clone(), + })); + } +} + +// For storing the data into database. +impl From for serde_json::Value { + fn from(value: ApiRequestEvent) -> Self { + serde_json::to_value(value).unwrap() + } +} + +impl TryFrom for ApiRequestEvent { + type Error = crate::event::Error; + + fn try_from(value: serde_json::Value) -> Result { + let res: ApiRequestEvent = serde_json::from_value(value)?; + Ok(res) + } + +} + +#[cfg(test)] +mod tests { + use super::{ApiRequestEvent, ApiType}; + use common::config::Config; + use serde_json::Value; + + const SER: &str = + r#"{"api":"Blob","config":{"base_dir":"","database":{"db_path":"/tmp/.mega/mega.db","db_type":"sqlite","db_url":"postgres://mega:mega@localhost:5432/mega","max_connection":32,"min_connection":16,"sqlx_logging":false},"lfs":{"enable_split":true,"split_size":1073741824},"log":{"level":"info","log_path":"/tmp/.mega/logs","print_std":true},"monorepo":{"import_dir":"/third-part"},"oauth":{"github_client_id":"","github_client_secret":""},"pack":{"channel_message_size":1000000,"clean_cache_after_decode":true,"pack_decode_cache_path":"/tmp/.mega/cache","pack_decode_mem_size":4},"ssh":{"ssh_key_path":"/tmp/.mega/ssh"},"storage":{"big_obj_threshold":1024,"lfs_obj_local_path":"/tmp/.mega/lfs","obs_access_key":"","obs_endpoint":"https://obs.cn-east-3.myhuaweicloud.com","obs_region":"cn-east-3","obs_secret_key":"","raw_obj_local_path":"/tmp/.mega/objects","raw_obj_storage_type":"LOCAL"},"ztm":{"agent":"127.0.0.1:7777","ca":"127.0.0.1:9999","hub":"127.0.0.1:8888"}}}"#; + + #[test] + fn test_conversion() { + let evt = ApiRequestEvent {api: ApiType::Blob, config: Config::default()}; + + // Convert into value + let serialized: Value = Value::from(evt); + assert_eq!(serialized.to_string().as_str(), SER); + + // Convert from value + let _ = ApiRequestEvent::try_from(serialized).unwrap(); + } +} diff --git a/taurus/src/event/mod.rs b/taurus/src/event/mod.rs new file mode 100644 index 000000000..455b555a7 --- /dev/null +++ b/taurus/src/event/mod.rs @@ -0,0 +1,121 @@ +use std::fmt::Display; + +use api_request::ApiRequestEvent; + +use async_trait::async_trait; +use chrono::{DateTime, Utc}; +use serde::{Deserialize, Serialize}; +use serde_json::Value; +use thiserror::Error; + +pub mod api_request; + +#[allow(clippy::large_enum_variant)] +#[derive(Debug, Clone, Serialize, Deserialize)] +pub enum EventType { + ApiRequest(ApiRequestEvent), + + // Reserved + ErrorEvent, +} + +#[derive(Debug, Clone)] +pub struct Message { + pub(crate) id: i64, + pub(crate) create_time: DateTime, + pub(crate) evt: EventType, +} + +#[derive(Debug, Error)] +pub enum Error { + #[error("Error converting from database")] + MismatchedData(#[from] serde_json::error::Error), +} + +#[async_trait] +pub trait EventBase: + Send + Sync + std::fmt::Display + Into + TryFrom +{ + // defines the callback function for this event. + async fn process(&self); +} + +impl Display for EventType { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + write!(f, "{:?}", self) + } +} + +impl EventType { + pub(crate) async fn process(&self) { + match self { + // I can't easily add a trait bound for the enum members, + // so you have to manually add a process logic for your event here. + EventType::ApiRequest(evt) => evt.process().await, + // EventType::SomeOtherEvent(xxx) => xxx.process().await, + + // This won't happen unless failed to load events from database. + // And that's because of a event conversion error. + // You should recheck yout conversion code logic. + EventType::ErrorEvent => panic!("Got error event"), + } + } +} + +impl Display for Message { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + write!( + f, + "ID: {}, Created at: {}", + self.id, self.create_time + ) + } +} + +impl From for callisto::mq_storage::Model { + fn from(val: Message) -> Self { + use callisto::mq_storage::Model; + + let category = match val.evt { + EventType::ApiRequest(_) => "ApiRequestEvent".into(), + + #[allow(unreachable_patterns)] + _ => "Unknown".into(), + }; + + let content: Value = match val.evt { + EventType::ApiRequest(evt) => evt.into(), + + #[allow(unreachable_patterns)] + _ => Value::Null, + }; + + Model { + id: val.id, + category, + create_time: val.create_time.naive_utc(), + content: Some(content.to_string()), + } + } +} + +impl From for Message { + fn from(value: callisto::mq_storage::Model) -> Self { + let id = value.id; + let create_time = value.create_time.and_utc(); + let evt = match value.category.as_str() { + "ApiRequestEvent" => { + if let Some(s) = value.content { + let evt = serde_json::from_str(&s).unwrap(); + EventType::ApiRequest(evt) + } else { + EventType::ErrorEvent + } + }, + + _ => EventType::ErrorEvent + }; + + Self { id, create_time, evt } + } +} diff --git a/taurus/src/init.rs b/taurus/src/init.rs new file mode 100644 index 000000000..9d7b3cf0a --- /dev/null +++ b/taurus/src/init.rs @@ -0,0 +1,16 @@ +use common::config::Config; +use jupiter::context::Context; +use crate::queue::{MessageQueue, MQ}; + +pub async fn init_mq(config: &Config) { + let ctx = Context::new(config.clone()).await; + let seq = match ctx.services.mq_storage.get_latest_message().await { + Some(model) => model.id + 1, + None => 1, + }; + + let mq = MessageQueue::new(12, seq, ctx); + mq.start(); + + MQ.set(mq).unwrap(); +} diff --git a/taurus/src/lib.rs b/taurus/src/lib.rs new file mode 100644 index 000000000..55f636af0 --- /dev/null +++ b/taurus/src/lib.rs @@ -0,0 +1,4 @@ +pub mod init; +pub mod event; +pub mod queue; +pub mod cache; diff --git a/taurus/src/queue.rs b/taurus/src/queue.rs new file mode 100644 index 000000000..2e4149479 --- /dev/null +++ b/taurus/src/queue.rs @@ -0,0 +1,90 @@ +use std::fmt::Debug; +use std::sync::atomic::AtomicI64; +use std::sync::{Arc, OnceLock}; + +use chrono::Utc; +use crossbeam_channel::{unbounded, Sender}; +use crossbeam_channel::Receiver; +use jupiter::context::Context; +use tokio::runtime::{Builder, Runtime}; + +use crate::cache::get_mcache; +use crate::event::{Message, EventType}; + +// Lazy initialized static MessageQueue instance. +pub(crate) static MQ: OnceLock = OnceLock::new(); +pub fn get_mq() -> &'static MessageQueue { + MQ.get().unwrap() +} + +pub struct MessageQueue { + sender: Sender, + receiver: Receiver, + // sem: Arc, + runtime: Arc, + cur_id: Arc, + pub(crate) context: Context, +} + +unsafe impl Send for MessageQueue{} +unsafe impl Sync for MessageQueue{} + +impl Debug for MessageQueue { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + // Just ignore context field. + f.debug_struct("MessageQueue").field("sender", &self.sender).field("receiver", &self.receiver).field("runtime", &self.runtime).finish() + } +} + +impl MessageQueue { + // Should be singleton. + pub(crate) fn new(n_workers: usize, seq: i64, ctx: Context) -> Self { + let (s, r) = unbounded::(); + let rt = Builder::new_multi_thread() + .worker_threads(n_workers) + .build() + .unwrap(); + + MessageQueue { + sender: s.to_owned(), + receiver: r.to_owned(), + // sem: Arc::new(Semaphore::new(n_workers)), + runtime: Arc::new(rt), + cur_id: Arc::new(AtomicI64::new(seq)), + context: ctx, + } + } + + pub(crate) fn start(&self) { + let receiver = self.receiver.clone(); + // let sem = self.sem.clone(); + let rt = self.runtime.clone(); + + tokio::spawn(async move { + let mc = get_mcache(); + loop { + match receiver.recv() { + Ok(msg) => { + let stored = msg.clone(); + mc.add(stored).await; + rt.spawn(async move { + msg.evt.process().await; + }); + }, + Err(e) => { + // Should not error here. + panic!("Event Loop Panic: {e}"); + } + } + } + }); + } + + pub(crate) fn send(&self, evt: EventType) { + let _ = self.sender.send(Message { + id: self.cur_id.fetch_add(1, std::sync::atomic::Ordering::Relaxed), + create_time: Utc::now(), + evt + }); + } +}