Struct disintegrate_postgres::PgEventStore

pub struct PgEventStore<E, S>where
    S: Serde<E> + Send + Sync,{ /* private fields */ }

PostgreSQL event store implementation.

Implementations

impl<E, S> PgEventStore<E, S>where S: Serde<E> + Send + Sync,

pub fn new(pool: PgPool, serde: S) -> Self

Creates a new instance of PgEventStore.

Arguments

• pool - The PostgreSQL connection pool.
• serde - The serialization implementation for the event payload.

Trait Implementations

impl<E: Clone, S> Clone for PgEventStore<E, S>where S: Serde<E> + Send + Sync + Clone,

fn clone(&self) -> PgEventStore<E, S>

Returns a copy of the value.

fn clone_from(&mut self, source: &Self)

Performs copy-assignment from source.

impl<E, S> EventStore<E> for PgEventStore<E, S>where E: Event + Send + Sync, S: Serde<E> + Send + Sync,

Implementation of the event store using PostgreSQL.

This module provides the implementation of the EventStore trait for PgEventStore, allowing interaction with a PostgreSQL event store. It enables streaming events based on a query and appending new events to the event store.

fn stream<'a, QE>( &'a self, query: &'a StreamQuery<QE> ) -> Result<BoxStream<'_, PersistedEvent<QE>>, Self::Error>where QE: TryFrom<E> + Event + Send + Sync + Clone, <QE as TryFrom<E>>::Error: Debug + Send,

Streams events based on the provided query.

This function fetches events from the PostgreSQL event store based on the provided query. It constructs a SQL query using the SqlEventsCriteriaBuilder and executes the query using the sqlx crate. The fetched events are then converted into PersistedEvent instances and streamed as a boxed stream.

Arguments

• query - The stream query specifying the criteria for filtering events.

Returns

A Result containing a boxed stream of PersistedEvent that matches the query criteria, or an error of type Self::Error.

fn append<'life0, 'async_trait, QE>( &'life0 self, events: Vec<E>, query: StreamQuery<QE>, last_event_id: i64 ) -> Pin<Box<dyn Future<Output = Result<Vec<PersistedEvent<E>>, Self::Error>> + Send + 'async_trait>>where E: Clone + 'async_trait, QE: Event + Clone + Send + 'async_trait, Self: 'async_trait, 'life0: 'async_trait,

Appends new events to the event store.

This function inserts the provided events into the PostgreSQL event store by performing two separate inserts. First, it inserts the events into the event_sequence table to reclaim a set of IDs for the events. Then, it inserts the events into the event table along with their IDs, event types, domain identifiers, and payloads. Finally, it marks the event IDs as consumed in the event sequence table. If marking the event IDs as consumed fails (e.g., another process has already consumed the IDs), a conflict error is raised. This conflict indicates that the data retrieved by the query is stale, meaning that the events generated are no longer valid due to being generated from an old version of the event store.

Arguments

• events - A vector of events to be appended.
• query - The stream query specifying the criteria for filtering events.
• last_event_id - The ID of the last consumed event.

Returns

A Result containing a vector of PersistedEvent representing the appended events, or an error of type Self::Error.

type Error = Error

impl<E, Srd> StateStore<E> for PgEventStore<E, Srd>where E: Event + Clone + Send + Sync, Srd: Serde<E> + Send + Sync,

Implementation of the StateStore trait for a PostgreSQL event store.

This struct allows hydrating and saving states using a PostgreSQL event store. It requires the event type E to implement the Event traits. Additionally, it requires the serializer type Srd to implement Serde traits.

fn hydrate<'life0, 'async_trait, QE, S>( &'life0 self, default: S ) -> Pin<Box<dyn Future<Output = Result<Hydrated<S>, Self::Error>> + Send + 'async_trait>>where S: State<Event = QE> + 'async_trait, QE: TryFrom<E> + Event + Clone + Send + Sync + 'async_trait, <QE as TryFrom<E>>::Error: Debug + Send, Self: 'async_trait, 'life0: 'async_trait,

Hydrates the given state using the events stored in the PostgreSQL event store.

It retrieves events from the event store and applies them to the default state, resulting in a hydrated state. The function returns a Hydrated object containing the hydrated state and the version of the state.

Arguments

• default - The default state to be hydrated.

Returns

A Result containing the hydrated state wrapped in a Hydrated object and the version of the state if successful, or an Error if an error occurs during the hydration process.

fn save<'life0, 'life1, 'async_trait, QE, S>( &'life0 self, state: &'life1 mut Hydrated<S> ) -> Pin<Box<dyn Future<Output = Result<(), Self::Error>> + Send + 'async_trait>>where S: State<Event = QE> + 'async_trait, QE: Into<E> + Event + Clone + Send + Sync + 'async_trait, Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Saves the changes from the provided hydrated state into the PostgreSQL event store.

It appends the changes in the hydrated state as events to the event store, updating the state’s query and version.

Arguments

• state - The hydrated state containing the changes to be saved.

Returns

A Result indicating success or failure of the save operation. Returns Ok(()) if the changes are successfully saved, or an Error if an error occurs during the save process.

type Error = Error