karaolidis/qrust
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add RabbitMQ messaging

Browse Source 
Signed-off-by: Nikolaos Karaolidis <nick@karaolidis.com>
This commit is contained in:
Nick Karaolidis
2023-08-28 13:45:51 +03:00
parent fa509934ae
commit 9d3d51f23c
22 changed files with 1232 additions and 151 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
backend/assets/Cargo.toml
View File

@@ -25,3 +25,6 @@ deadpool = { version = "0.9.5", features = [
] }
serde = "1.0.188"
log = "0.4.20"
lapin = "2.3.1"
serde_json = "1.0.105"
log4rs = "1.2.0"

45
backend/assets/src/main.rs
View File

@@ -2,26 +2,46 @@ use axum::{
routing::{delete, get, post},
Extension, Router, Server,
};
use common::alpaca::create_alpaca_pool;
use common::pool::{
create_alpaca_pool_from_env, create_database_pool_from_env, create_rabbitmq_pool_from_env,
};
use deadpool::managed::{Hook, HookError, HookErrorCause};
use dotenv::dotenv;
use lapin::ExchangeKind;
use log::info;
use sqlx::PgPool;
use std::{env, error::Error, net::SocketAddr};
use std::{error::Error, net::SocketAddr};
mod routes;
#[tokio::main]
async fn main() -> Result<(), Box<dyn Error + Send + Sync>> {
dotenv().ok();
log4rs::init_file("log4rs.yaml", Default::default()).unwrap();
let database_pool = PgPool::connect(&env::var("DATABASE_URL").unwrap()).await?;
let alpaca_pool = create_alpaca_pool(
&env::var("APCA_API_BASE_URL").unwrap(),
&env::var("APCA_API_KEY_ID").unwrap(),
&env::var("APCA_API_SECRET_KEY").unwrap(),
10,
)?;
let num_clients = 10;
let database_pool = create_database_pool_from_env(num_clients).await?;
let alpaca_pool = create_alpaca_pool_from_env(num_clients).await?;
let rabbitmq_pool = create_rabbitmq_pool_from_env(
num_clients,
Hook::async_fn(|connection: &mut lapin::Connection, _| {
Box::pin(async move {
connection
.create_channel()
.await
.map_err(|e| HookError::Abort(HookErrorCause::Backend(e)))?
.exchange_declare(
"assets",
ExchangeKind::Topic,
Default::default(),
Default::default(),
)
.await
.map_err(|e| HookError::Abort(HookErrorCause::Backend(e)))?;
Ok(())
})
}),
)
.await?;
let app = Router::new()
.route("/assets", get(routes::get_assets))
@@ -30,7 +50,8 @@ async fn main() -> Result<(), Box<dyn Error + Send + Sync>> {
.route("/assets/:symbol", post(routes::update_asset))
.route("/assets/:symbol", delete(routes::delete_asset))
.layer(Extension(database_pool))
.layer(Extension(alpaca_pool));
.layer(Extension(alpaca_pool))
.layer(Extension(rabbitmq_pool));
let addr = SocketAddr::from(([0, 0, 0, 0], 7878));
info!("Listening on {}...", addr);

86
backend/assets/src/routes.rs
View File

@@ -1,11 +1,15 @@
use apca::api::v2::asset::{self, Symbol};
use axum::{extract::Path, http::StatusCode, Extension, Json};
use common::{
alpaca::AlpacaPool,
database::{Asset, Class, Exchange},
pool::{AlpacaPool, RabbitmqPool},
};
use serde::Deserialize;
use sqlx::{query, query_as, types::Uuid, PgPool};
use sqlx::{
query_as,
types::{time::OffsetDateTime, Uuid},
PgPool,
};
pub async fn get_assets(
Extension(database_pool): Extension<PgPool>,
@@ -40,6 +44,7 @@ pub struct AddAssetRequest {
pub async fn add_asset(
Extension(database_pool): Extension<PgPool>,
Extension(alpaca_pool): Extension<AlpacaPool>,
Extension(rabbitmq_pool): Extension<RabbitmqPool>,
Json(request): Json<AddAssetRequest>,
) -> Result<StatusCode, StatusCode> {
if query_as!(Asset, r#"SELECT id, symbol, class as "class: Class", exchange as "exchange: Exchange", trading, date_added FROM assets WHERE symbol = $1"#, request.symbol)
@@ -58,17 +63,36 @@ pub async fn add_asset(
.await
.map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?;
query!(
r#"INSERT INTO assets (id, symbol, class, exchange, trading) VALUES ($1, $2, $3::CLASS, $4::EXCHANGE, $5)"#,
Uuid::parse_str(&asset.id.to_string()).unwrap(),
let asset = query_as!(
Asset,
r#"INSERT INTO assets (id, symbol, class, exchange, trading, date_added) VALUES ($1, $2, $3::CLASS, $4::EXCHANGE, $5, $6) RETURNING id, symbol, class as "class: Class", exchange as "exchange: Exchange", trading, date_added"#,
Uuid::parse_str(&asset.id.to_string()).map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?,
asset.symbol,
Class::from(asset.class) as Class,
Exchange::from(asset.exchange) as Exchange,
request.trading
request.trading.unwrap_or(false),
OffsetDateTime::now_utc(),
)
.execute(&database_pool)
.fetch_one(&database_pool)
.await
.unwrap();
.map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?;
rabbitmq_pool
.get()
.await
.map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?
.create_channel()
.await
.map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?
.basic_publish(
"assets",
"assets.added",
Default::default(),
&serde_json::to_vec(&asset).map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?,
Default::default(),
)
.await
.map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?;
Ok(StatusCode::CREATED)
}
@@ -81,28 +105,64 @@ pub struct UpdateAssetRequest {
pub async fn update_asset(
Extension(database_pool): Extension<PgPool>,
Extension(rabbitmq_pool): Extension<RabbitmqPool>,
Path(symbol): Path<String>,
Json(request): Json<UpdateAssetRequest>,
) -> Result<StatusCode, StatusCode> {
query_as!(
let asset = query_as!(
Asset,
r#"UPDATE assets SET trading = $1 WHERE symbol = $2"#,
r#"UPDATE assets SET trading = $1 WHERE symbol = $2 RETURNING id, symbol, class as "class: Class", exchange as "exchange: Exchange", trading, date_added"#,
request.trading,
symbol
)
.execute(&database_pool)
.fetch_one(&database_pool)
.await
.map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?;
rabbitmq_pool
.get()
.await
.map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?
.create_channel()
.await
.map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?
.basic_publish(
"assets",
"assets.updated",
Default::default(),
&serde_json::to_vec(&asset).map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?,
Default::default(),
)
.await
.map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?;
Ok(StatusCode::OK)
}
pub async fn delete_asset(
Extension(database_pool): Extension<PgPool>,
Extension(rabbitmq_pool): Extension<RabbitmqPool>,
Path(symbol): Path<String>,
) -> Result<StatusCode, StatusCode> {
query!(r#"DELETE FROM assets WHERE symbol = $1"#, symbol)
.execute(&database_pool)
let asset = query_as!(Asset, r#"DELETE FROM assets WHERE symbol = $1 RETURNING id, symbol, class as "class: Class", exchange as "exchange: Exchange", trading, date_added"#, symbol)
.fetch_one(&database_pool)
.await
.map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?;
rabbitmq_pool
.get()
.await
.map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?
.create_channel()
.await
.map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?
.basic_publish(
"assets",
"assets.deleted",
Default::default(),
&serde_json::to_vec(&asset).map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?,
Default::default(),
)
.await
.map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?;