use log::info; use std::collections::HashMap; use sqlx::postgres::{ PgConnectOptions, PgPool, PgPoolOptions, }; pub static INSERT_EVENT: &str = " INSERT INTO events ( aggregate_type, aggregate_id, sequence, payload, metadata ) VALUES ( $1, $2, $3, $4, $5 ); "; pub static SELECT_EVENTS_WITH_METADATA: &str = " SELECT sequence, payload, metadata FROM events WHERE aggregate_type = $1 AND aggregate_id = $2 ORDER BY sequence; "; pub static UPDATE_EVENTS: &str = " UPDATE events SET payload = $4, metadata = $5 WHERE aggregate_type = $1 AND aggregate_id = $2 AND sequence = $3; "; async fn get_pool() -> Result { // "postgres://demo_user:demo_pass@localhost:8087/demo" let options = PgConnectOptions::new() .host("localhost") .port(8087) .database("demo") .username("demo_user") .password("demo_pass"); let pool = PgPoolOptions::new() .max_connections(5) .connect_with(options) .await?; Ok(pool) } async fn test_connect() -> Result<(), sqlx::Error> { let pool = get_pool().await?; // Make a simple query to return the given parameter let row: (i64,) = sqlx::query_as("SELECT $1") .bind(150_i64) .fetch_one(&pool) .await?; assert_eq!(row.0, 150); info!("Received {}", row.0); Ok(()) } async fn test_insert_select_update() -> Result<(), sqlx::Error> { let pool = get_pool().await?; // insert let mut payload = HashMap::new(); payload.insert("k".to_string(), "v".to_string()); let payload = match serde_json::to_value(payload) { Ok(x) => x, Err(e) => { panic!( "payload serialization error '{}'", e.to_string() ); }, }; let mut metadata = HashMap::new(); metadata.insert("k1".to_string(), "v1".to_string()); let metadata = match serde_json::to_value(metadata) { Ok(x) => x, Err(e) => { panic!( "metadata serialization error '{}'", e.to_string() ); }, }; let aggregate_type = uuid::Uuid::new_v4().to_string(); let aggregate_id = uuid::Uuid::new_v4().to_string(); let rows_affected = sqlx::query(INSERT_EVENT) .bind(&aggregate_type) .bind(&aggregate_id) .bind(1) .bind(payload) .bind(metadata) .execute(&pool) .await? .rows_affected(); info!( "Insert affected '{}' rows", rows_affected ); // select let rows: Vec<( i64, serde_json::Value, serde_json::Value, )> = sqlx::query_as(SELECT_EVENTS_WITH_METADATA) .bind(&aggregate_type) .bind(&aggregate_id) .fetch_all(&pool) .await?; info!("select success"); for row in rows { let payload: HashMap = serde_json::from_value(row.1).unwrap(); let metadata: HashMap = serde_json::from_value(row.2).unwrap(); info!( "{},{:?},{:?}", &row.0, payload, metadata ) } // update let mut payload = HashMap::new(); payload.insert("k2".to_string(), "v2".to_string()); let payload = match serde_json::to_value(payload) { Ok(x) => x, Err(e) => { panic!( "payload serialization error '{}'", e.to_string() ); }, }; let mut metadata = HashMap::new(); metadata.insert("k3".to_string(), "v3".to_string()); let metadata = match serde_json::to_value(metadata) { Ok(x) => x, Err(e) => { panic!( "metadata serialization error '{}'", e.to_string() ); }, }; let rows_affected = sqlx::query(UPDATE_EVENTS) .bind(&aggregate_type) .bind(&aggregate_id) .bind(1) .bind(payload) .bind(metadata) .execute(&pool) .await? .rows_affected(); info!( "Insert affected '{}' rows", rows_affected ); // select let rows: Vec<( i64, serde_json::Value, serde_json::Value, )> = sqlx::query_as(SELECT_EVENTS_WITH_METADATA) .bind(&aggregate_type) .bind(&aggregate_id) .fetch_all(&pool) .await?; info!("select success"); for row in rows { let payload: HashMap = serde_json::from_value(row.1).unwrap(); let metadata: HashMap = serde_json::from_value(row.2).unwrap(); info!( "{},{:?},{:?}", &row.0, payload, metadata ) } Ok(()) } pub async fn check_postgres() -> Result<(), sqlx::Error> { test_connect().await?; test_insert_select_update().await?; Ok(()) }