obix 0.2.26

Implementation of outbox backed by PG / sqlx
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

use std::{marker::PhantomData, sync::Arc};

use es_entity::hooks::{CommitHook, HookOperation, PreCommitRet};
use serde::{Serialize, de::DeserializeOwned};
use tokio::sync::broadcast;

use crate::out::event::{EphemeralEventType, EphemeralOutboxEvent};
use crate::tables::MailboxTables;

pub struct EphemeralEvent<P> {
    pub event_type: EphemeralEventType,
    pub payload: P,
}

pub struct PersistEphemeralEvents<P, Tables>
where
    P: Serialize + DeserializeOwned + Send + Sync + 'static + Unpin,
    Tables: MailboxTables,
{
    sender: broadcast::Sender<Arc<EphemeralOutboxEvent<P>>>,
    pre_commit_events: Vec<EphemeralEvent<P>>,
    post_commit_events: Vec<EphemeralOutboxEvent<P>>,
    _phantom: PhantomData<Tables>,
}

impl<P, Tables> PersistEphemeralEvents<P, Tables>
where
    P: Serialize + DeserializeOwned + Send + Sync + 'static + Unpin,
    Tables: MailboxTables,
{
    pub fn new(
        sender: broadcast::Sender<Arc<EphemeralOutboxEvent<P>>>,
        event_type: EphemeralEventType,
        event: impl Into<P>,
    ) -> Self {
        Self {
            sender,
            pre_commit_events: vec![EphemeralEvent {
                event_type,
                payload: event.into(),
            }],
            post_commit_events: Vec::new(),
            _phantom: PhantomData,
        }
    }
}

impl<P, Tables> CommitHook for PersistEphemeralEvents<P, Tables>
where
    P: Serialize + DeserializeOwned + Send + Sync + 'static + Unpin,
    Tables: MailboxTables,
{
    async fn pre_commit(
        mut self,
        mut op: HookOperation<'_>,
    ) -> Result<PreCommitRet<'_, Self>, sqlx::Error> {
        for event in self.pre_commit_events.drain(..) {
            let persisted =
                Tables::persist_ephemeral_event_in_op(&mut op, event.event_type, event.payload)
                    .await?;
            self.post_commit_events.push(persisted);
        }
        PreCommitRet::ok(self, op)
    }

    fn post_commit(self) {
        for event in self.post_commit_events {
            let _ = self.sender.send(Arc::new(event));
        }
    }

    fn merge(&mut self, other: &mut Self) -> bool {
        self.pre_commit_events.append(&mut other.pre_commit_events);
        true
    }
}