1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
use cqrs_es::persist::PersistedEventStore;
use cqrs_es::{Aggregate, CqrsFramework, Query};
use crate::{MysqlCqrs, MysqlEventRepository};
use sqlx::mysql::MySqlPoolOptions;
use sqlx::{MySql, Pool};
pub async fn default_mysql_pool(connection_string: &str) -> Pool<MySql> {
MySqlPoolOptions::new()
.max_connections(10)
.connect(connection_string)
.await
.expect("unable to connect to database")
}
pub fn mysql_cqrs<A>(pool: Pool<MySql>, query_processor: Vec<Box<dyn Query<A>>>) -> MysqlCqrs<A>
where
A: Aggregate,
{
let repo = MysqlEventRepository::new(pool);
let store = PersistedEventStore::new_event_store(repo);
CqrsFramework::new(store, query_processor)
}
pub fn mysql_snapshot_cqrs<A>(
pool: Pool<MySql>,
query_processor: Vec<Box<dyn Query<A>>>,
snapshot_size: usize,
) -> MysqlCqrs<A>
where
A: Aggregate,
{
let repo = MysqlEventRepository::new(pool);
let store = PersistedEventStore::new_snapshot_store(repo, snapshot_size);
CqrsFramework::new(store, query_processor)
}
pub fn mysql_aggregate_cqrs<A>(
pool: Pool<MySql>,
query_processor: Vec<Box<dyn Query<A>>>,
) -> MysqlCqrs<A>
where
A: Aggregate,
{
let repo = MysqlEventRepository::new(pool);
let store = PersistedEventStore::new_aggregate_store(repo);
CqrsFramework::new(store, query_processor)
}
#[cfg(test)]
mod test {
use crate::testing::tests::{
TestAggregate, TestQueryRepository, TestView, TEST_CONNECTION_STRING,
};
use crate::{default_mysql_pool, mysql_cqrs, MysqlViewRepository};
use std::sync::Arc;
#[tokio::test]
async fn test_valid_cqrs_framework() {
let pool = default_mysql_pool(TEST_CONNECTION_STRING).await;
let repo = MysqlViewRepository::<TestView, TestAggregate>::new("test_view", pool.clone());
let query = TestQueryRepository::new(Arc::new(repo));
let _ps = mysql_cqrs(pool, vec![Box::new(query)]);
}
}