es-entity 0.12.21

Event Sourcing Entity Framework
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
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115

mod helpers;

use es_entity::operation::{
    AtomicOperation, DbOp, SavepointOperation,
    hooks::{CommitHook, HookOperation, PreCommitRet},
};
use std::{
    any::TypeId,
    sync::{Arc, Mutex},
};

#[derive(Debug)]
struct CountingHook {
    label: String,
    post: Arc<Mutex<Vec<String>>>,
}

impl CommitHook for CountingHook {
    async fn pre_commit(
        self,
        op: HookOperation<'_>,
    ) -> Result<PreCommitRet<'_, Self>, sqlx::Error> {
        PreCommitRet::ok(self, op)
    }

    fn post_commit(self) {
        self.post.lock().unwrap().push(self.label);
    }
}

/// The pin: if `AtomicOperation` ever regains a method that is generic
/// without `Self: Sized`, this stops compiling with E0038.
fn assert_object_safe(op: &mut dyn AtomicOperation) -> &mut dyn AtomicOperation {
    op
}

#[tokio::test]
async fn dyn_atomic_operation_drives_hooks_and_savepoints() -> anyhow::Result<()> {
    let pool = helpers::init_pool().await?;
    let post = Arc::new(Mutex::new(Vec::new()));
    let mut db_op = DbOp::init(&pool).await?;

    {
        let op: &mut dyn AtomicOperation = assert_object_safe(&mut db_op);

        assert!(op.supports_hooks());
        let _ = op.maybe_now();
        let _ = op.clock();
        let _ = op.connection();

        let hook = CountingHook {
            label: "root".to_string(),
            post: post.clone(),
        };
        op.add_commit_hook_dyn(TypeId::of::<CountingHook>(), Box::new(hook))
            .ok()
            .expect("dyn op must accept a hook");

        let registered = op
            .commit_hook_dyn(TypeId::of::<CountingHook>())
            .and_then(|h| h.as_any().downcast_ref::<CountingHook>())
            .expect("the hook just registered must be readable back through the dyn op");
        assert_eq!(registered.label, "root");

        // Savepoints work uniformly through the trait object, too.
        let result = op
            .with_savepoint(async |sp| {
                let sp_op: &mut dyn AtomicOperation = sp;
                assert!(sp_op.supports_hooks());
                sp_op
                    .add_commit_hook_dyn(
                        TypeId::of::<CountingHook>(),
                        Box::new(CountingHook {
                            label: "savepoint".to_string(),
                            post: post.clone(),
                        }),
                    )
                    .ok()
                    .expect("savepoint must accept a hook via the dyn op");
                Ok::<_, sqlx::Error>(())
            })
            .await?;
        assert!(result.is_ok());
    }

    db_op.commit().await?;

    let mut ran = post.lock().unwrap().clone();
    ran.sort();
    assert_eq!(ran, vec!["root".to_string(), "savepoint".to_string()]);

    Ok(())
}

#[tokio::test]
async fn dyn_atomic_operation_generic_wrapper_still_works() -> anyhow::Result<()> {
    let pool = helpers::init_pool().await?;
    let post = Arc::new(Mutex::new(Vec::new()));
    let mut db_op = DbOp::init(&pool).await?;

    // The Sized-only generic wrappers are unaffected on a concrete (non-dyn)
    // operation: they route through the same erased methods internally.
    db_op
        .add_commit_hook(CountingHook {
            label: "typed".to_string(),
            post: post.clone(),
        })
        .unwrap();
    assert_eq!(db_op.commit_hook::<CountingHook>().unwrap().label, "typed");

    db_op.commit().await?;
    assert_eq!(post.lock().unwrap().clone(), vec!["typed".to_string()]);

    Ok(())
}