event_hex 0.0.3

A pragmatic Rust toolkit for Domain-Driven Design with first-class support for event sourcing and CQRS.
Documentation 
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
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95

use std::{any::Any, sync::Arc};

use async_trait::async_trait;
use mongodb::{
    options::{ReadPreference, SelectionCriteria}, Client,
    ClientSession,
};

use crate::errors::EventHexError;
use crate::persistence::transaction::{
    ErasedResult, EventTransactionContext, EventTransactionHandler, EventTransactionManager,
};

// Mongo context implementation
pub struct MongoContext {
    pub session: ClientSession,
}

// Abstract transaction context implementation
impl EventTransactionContext for MongoContext {
    fn as_any_mut(&mut self) -> &mut dyn Any {
        self
    }
}

// Context for working without transactions (standalone mode)
struct NoopTransactionContext;

impl EventTransactionContext for NoopTransactionContext {
    fn as_any_mut(&mut self) -> &mut dyn Any {
        self
    }
}

pub struct MongoTransactionManager {
    client: Arc<Client>,
    use_transactions: bool,
}

impl MongoTransactionManager {
    pub async fn new(client: Arc<Client>) -> Self {
        let use_transactions = Self::detect_transaction_support(&client).await;
        Self { client, use_transactions }
    }

    async fn detect_transaction_support(client: &Client) -> bool {
        match client.start_session().await {
            Ok(mut session) => {
                match session.start_transaction().await {
                    Ok(_) => {
                        let _ = session.abort_transaction().await;
                        true
                    }
                    Err(_) => false,
                }
            }
            Err(_) => false,
        }
    }
}

#[async_trait]
impl EventTransactionManager for MongoTransactionManager {
    async fn run_transaction(&self, handler: EventTransactionHandler) -> Result<ErasedResult, EventHexError> {
        if self.use_transactions {
            // Transaction logic
            let mut session = self.client.start_session().await.map_err(|e| EventHexError::MongoError(e))?;

            session
                .start_transaction()
                .selection_criteria(SelectionCriteria::ReadPreference(ReadPreference::Primary))
                .await
                .map_err(|e| EventHexError::MongoError(e))?;

            let mut ctx = MongoContext { session };

            let result = handler(&mut ctx).await;

            match result {
                Ok(value) => {
                    ctx.session.commit_transaction().await.map_err(|e| EventHexError::MongoError(e))?;
                    Ok(value)
                }
                Err(e) => {
                    ctx.session.abort_transaction().await.map_err(|e| EventHexError::MongoError(e))?;
                    Err(e)
                }
            }
        } else {
            // Without transaction (standalone mode)
            let mut ctx = NoopTransactionContext;
            handler(&mut ctx).await
        }
    }
}