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

PostgreSQL event store implementation.

Implementations§

source§

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

source

pub async fn new(pool: PgPool, serde: S) -> Result<Self, Error>

Initializes the PostgreSQL DB and returns a new instance of PgEventStore.

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

Trait Implementations§

source§

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

source§

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

Returns a copy of the value. Read more
1.0.0 · source§

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

Performs copy-assignment from source. Read more
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.

source§

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

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.

source§

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 + Sync + '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

Auto Trait Implementations§

§

impl<E, S> !RefUnwindSafe for PgEventStore<E, S>

§

impl<E, S> Send for PgEventStore<E, S>
where E: Send,

§

impl<E, S> Sync for PgEventStore<E, S>
where E: Sync,

§

impl<E, S> Unpin for PgEventStore<E, S>
where E: Unpin, S: Unpin,

§

impl<E, S> !UnwindSafe for PgEventStore<E, S>

Blanket Implementations§

source§

impl<T> Any for T
where T: 'static + ?Sized,

source§

fn type_id(&self) -> TypeId

Gets the TypeId of self. Read more
source§

impl<T> Borrow<T> for T
where T: ?Sized,

source§

fn borrow(&self) -> &T

Immutably borrows from an owned value. Read more
source§

impl<T> BorrowMut<T> for T
where T: ?Sized,

source§

fn borrow_mut(&mut self) -> &mut T

Mutably borrows from an owned value. Read more
source§

impl<T> From<T> for T

source§

fn from(t: T) -> T

Returns the argument unchanged.

§

impl<T> Instrument for T

§

fn instrument(self, span: Span) -> Instrumented<Self>

Instruments this type with the provided [Span], returning an Instrumented wrapper. Read more
§

fn in_current_span(self) -> Instrumented<Self>

Instruments this type with the current Span, returning an Instrumented wrapper. Read more
source§

impl<T, U> Into<U> for T
where U: From<T>,

source§

fn into(self) -> U

Calls U::from(self).

That is, this conversion is whatever the implementation of From<T> for U chooses to do.

source§

impl<T> Same for T

§

type Output = T

Should always be Self
source§

impl<T> ToOwned for T
where T: Clone,

§

type Owned = T

The resulting type after obtaining ownership.
source§

fn to_owned(&self) -> T

Creates owned data from borrowed data, usually by cloning. Read more
source§

fn clone_into(&self, target: &mut T)

Uses borrowed data to replace owned data, usually by cloning. Read more
source§

impl<T, U> TryFrom<U> for T
where U: Into<T>,

§

type Error = Infallible

The type returned in the event of a conversion error.
source§

fn try_from(value: U) -> Result<T, <T as TryFrom<U>>::Error>

Performs the conversion.
source§

impl<T, U> TryInto<U> for T
where U: TryFrom<T>,

§

type Error = <U as TryFrom<T>>::Error

The type returned in the event of a conversion error.
source§

fn try_into(self) -> Result<U, <U as TryFrom<T>>::Error>

Performs the conversion.
§

impl<V, T> VZip<V> for T
where V: MultiLane<T>,

§

fn vzip(self) -> V

§

impl<T> WithSubscriber for T

§

fn with_subscriber<S>(self, subscriber: S) -> WithDispatch<Self>
where S: Into<Dispatch>,

Attaches the provided Subscriber to this type, returning a [WithDispatch] wrapper. Read more
§

fn with_current_subscriber(self) -> WithDispatch<Self>

Attaches the current default Subscriber to this type, returning a [WithDispatch] wrapper. Read more