Struct PgIdIndexer

pub struct PgIdIndexer<E: Event + Clone> { /* private fields */ }

The PgIdIndexer is a helper to index existing fields that have been newly tagged with the #[id] attribute in events.

Overview

The PgIdIndexer is an EventListener responsible for indexing fields in the event store that are tagged with the #[id] attribute. This allows querying of old events based on this new domain identifier once the indexing is complete. The PgIdIndexer listens to events and updates the event table in the database with the appropriate values.

Workflow

After you have tagged an existing field in your event structure with the #[id] attribute to mark it as a domain identifier:

use disintegrate_macros::Event;
use serde::{Serialize, Deserialize};
#[derive(Event, Clone, Serialize, Deserialize)]
struct MyEvent {
    #[id]
    existing_id: String,
    other_field: String,
}

1. Register the PgIdIndexer as an EventListener: Integrate the indexer with the event listener system to process the newly tagged domain identifier:

   use disintegrate_postgres::PgIdIndexer;
   use disintegrate_postgres::PgEventListenerConfig;
   use disintegrate_postgres::PgEventListener;
   use disintegrate_postgres::PgEventStore;
   use std::time::Duration;
   use disintegrate_macros::Event;
   use disintegrate::serde::json::Json;
   use serde::{Serialize, Deserialize};
   use sqlx::PgPool;
   
   #[derive(Event, Clone, Serialize, Deserialize)]
   struct MyEvent {
       #[id]
       existing_id: String,
       other_field: String,
   }
   
   async fn setup_listener(pool: PgPool, event_store: PgEventStore<MyEvent, Json<MyEvent>>) {
       let id_indexer = PgIdIndexer::<MyEvent>::new("index_exsting_id", pool);
       PgEventListener::builder(event_store)
           .register_listener(
               id_indexer,
               PgEventListenerConfig::poller(Duration::from_secs(5)).with_notifier()
           )
           .start_with_shutdown(shutdown())
           .await
           .expect("start event listener failed");
   }
   
   async fn shutdown() {
       tokio::signal::ctrl_c().await.expect("ctrl_c signal failed");
   }

2. Deploy the application: Start the application with the updated event structure and the PgIdIndexer integration. Newly created events with the new domain identifier will automatically have the identifier indexed.

Once the indexing process is complete, you can query the event store using the new domain identifier to fetch events.

If indexing is done, you can remove the PgIdIndexer from the list of registered event listeners.

Implementations

impl<E: Event + Clone> PgIdIndexer<E>

pub fn new(id: &'static str, pool: PgPool) -> Self

Creates a new PgIdIndexer instance for indexing events.

Arguments

• id - A unique identifier for the listener, used to store the last processed event_id in the database.
• pool - A PgPool instance for Postgres.

Trait Implementations

impl<E: Event + Clone + Send + Sync> EventListener<i64, E> for PgIdIndexer<E>

type Error = Error

The type of error that may occur during the handle of an event.

fn id(&self) -> &'static str

Returns the unique identifier of the event listener.

fn query(&self) -> &StreamQuery<PgEventId, E>

Returns the stream query used by the event listener.

fn handle<'life0, 'async_trait>( &'life0 self, event: PersistedEvent<PgEventId, E>, ) -> Pin<Box<dyn Future<Output = Result<(), Self::Error>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Handles an event.