esrs 0.18.0

A Prima.it-opinionated library to achieve cqrs/es
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
116
117
118
119
120
121
122
123
124
125
126
127

use std::ops::Deref;

use async_trait::async_trait;
use uuid::Uuid;

use crate::store::StoreEvent;
use crate::Aggregate;

/// This trait is used to implement an [`EventHandler`]. An event handler is intended to be an entity
/// which can create, update and delete a read side and perform side effects.
///
/// The main purpose of an [`EventHandler`] is to have an eventually persistent processor.
#[async_trait]
pub trait EventHandler<A>: Sync
where
    A: Aggregate,
{
    /// Handle an event and perform an action. This action could be over a read model or a side-effect.
    /// All the errors should be handled from within the [`EventHandler`] and shouldn't panic.
    async fn handle(&self, event: &StoreEvent<A::Event>);

    /// Perform a deletion of a resource using the given aggregate_id.
    async fn delete(&self, _aggregate_id: Uuid) {}

    /// The name of the event handler. By default, this is the type name of the event handler,
    /// but it can be overridden to provide a custom name. This name is used as
    /// part of tracing spans, to identify the event handler being run.
    fn name(&self) -> &'static str {
        std::any::type_name::<Self>()
    }
}

#[async_trait]
impl<A, Q, T> EventHandler<A> for T
where
    A: Aggregate,
    A::Event: Send + Sync,
    Q: EventHandler<A>,
    T: Deref<Target = Q> + Clone + Send + Sync,
{
    /// Deref call to [`EventHandler::handle`].
    async fn handle(&self, event: &StoreEvent<A::Event>) {
        self.deref().handle(event).await;
    }

    /// Deref call to [`EventHandler::handle`].
    async fn delete(&self, aggregate_id: Uuid) {
        self.deref().delete(aggregate_id).await;
    }

    /// Deref call to [`EventHandler::handle`].
    fn name(&self) -> &'static str {
        self.deref().name()
    }
}

/// This trait is used to implement a [`TransactionalEventHandler`]. A transactional event handler is
/// intended to be an entity which can create, update and delete a read side. No side effects must be
/// performed inside of this kind on handler.
///
/// An `handle` operation will result in a _deadlock_ if the implementation of this trait is used to
/// apply an event on an [`Aggregate`].
#[async_trait]
pub trait TransactionalEventHandler<A, Er, Ex>: Sync
where
    A: Aggregate,
{
    /// Handle an event in a transactional fashion and perform a read side crate, update or delete.
    /// If an error is returned the transaction will be aborted and the handling of a command by an
    /// aggregate will return an error.
    async fn handle(&self, event: &StoreEvent<A::Event>, executor: &mut Ex) -> Result<(), Er>;

    /// Perform a deletion of a read side projection using the given aggregate_id.
    async fn delete(&self, _aggregate_id: Uuid, _executor: &mut Ex) -> Result<(), Er> {
        Ok(())
    }

    /// The name of the event handler. By default, this is the type name of the event handler,
    /// but it can be overridden to provide a custom name. This name is used as
    /// part of tracing spans, to identify the event handler being run.
    fn name(&self) -> &'static str {
        std::any::type_name::<Self>()
    }
}

#[async_trait]
impl<A, Er, Ex, Q, T> TransactionalEventHandler<A, Er, Ex> for T
where
    A: Aggregate,
    A::Event: Send + Sync,
    Ex: Send,
    Q: TransactionalEventHandler<A, Er, Ex>,
    T: Deref<Target = Q> + Send + Sync,
{
    /// Deref call to [`TransactionalEventHandler::handle`].
    async fn handle(&self, event: &StoreEvent<A::Event>, executor: &mut Ex) -> Result<(), Er> {
        self.deref().handle(event, executor).await
    }

    /// Deref call to [`TransactionalEventHandler::delete`].
    async fn delete(&self, aggregate_id: Uuid, executor: &mut Ex) -> Result<(), Er> {
        self.deref().delete(aggregate_id, executor).await
    }

    /// Deref call to [`TransactionalEventHandler::name`].
    fn name(&self) -> &'static str {
        self.deref().name()
    }
}

/// The [`ReplayableEventHandler`] trait is used to add the `replay` behavior on an [`EventHandler`].
///
/// Being replayable means that the operation performed by this EventHandler should be idempotent
/// and should be intended to be "eventually consistent".
/// In other words it means that they should not perform external API calls, generate random numbers
/// or do anything that relies on external state and might change the outcome of this function.
///
/// The most common use case for this is when rebuilding read models: [`EventHandler`]s that write on
/// the database should be marked as replayable.
///
/// Another use case could be if there's the need to implement a retry logic for this event handler.
pub trait ReplayableEventHandler<A>: Sync
where
    Self: EventHandler<A>,
    A: Aggregate,
{
}