Skip to main content

EventLog

eventsdb_core::log

Trait EventLog 

Source 
pub trait EventLog: Send + Sync {
    // Required methods
    fn stream<'life0, 'life1, 'async_trait>(
        &'life0 self,
        id: &'life1 str,
    ) -> Pin<Box<dyn Future<Output = Result<Box<dyn EventStore>>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait,
             'life1: 'async_trait;
    fn read_all<'life0, 'life1, 'async_trait>(
        &'life0 self,
        from: Position,
        filter: &'life1 Filter,
        limit: usize,
    ) -> Pin<Box<dyn Future<Output = Result<Vec<Recorded>>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait,
             'life1: 'async_trait;
    fn head_position<'life0, 'async_trait>(
        &'life0 self,
    ) -> Pin<Box<dyn Future<Output = Result<Position>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait;
    fn subscribe(
        &self,
        from: Position,
        filter: Filter,
    ) -> Result<BoxStream<'static, Result<Recorded>>>;
    fn checkpoint_load<'life0, 'life1, 'async_trait>(
        &'life0 self,
        consumer: &'life1 str,
    ) -> Pin<Box<dyn Future<Output = Result<Position>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait,
             'life1: 'async_trait;
    fn checkpoint_save<'life0, 'life1, 'async_trait>(
        &'life0 self,
        consumer: &'life1 str,
        at: Position,
    ) -> Pin<Box<dyn Future<Output = Result<()>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait,
             'life1: 'async_trait;

    // Provided methods
    fn export<'life0, 'life1, 'async_trait>(
        &'life0 self,
        from: Position,
        filter: &'life1 Filter,
        limit: usize,
    ) -> Pin<Box<dyn Future<Output = Result<Vec<ExportedEvent>>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait,
             'life1: 'async_trait { ... }
    fn import<'life0, 'async_trait>(
        &'life0 self,
        events: Vec<ExportedEvent>,
    ) -> Pin<Box<dyn Future<Output = Result<ImportReport>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait { ... }
}

Required Methods§

Source

fn stream<'life0, 'life1, 'async_trait>( &'life0 self, id: &'life1 str, ) -> Pin<Box<dyn Future<Output = Result<Box<dyn EventStore>>> + Send + 'async_trait>>
where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

A handle on one stream. Everything below this is the per-stream SPI, unchanged.

Source

fn read_all<'life0, 'life1, 'async_trait>( &'life0 self, from: Position, filter: &'life1 Filter, limit: usize, ) -> Pin<Box<dyn Future<Output = Result<Vec<Recorded>>> + Send + 'async_trait>>
where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Events with position > from, across every stream, in position order, at most limit.

Exclusive on from so a cursor can be fed straight back in: Position::BEGINNING reads from the start, and the position of the last event handled reads the next batch.

Source

fn head_position<'life0, 'async_trait>( &'life0 self, ) -> Pin<Box<dyn Future<Output = Result<Position>> + Send + 'async_trait>>
where Self: 'async_trait, 'life0: 'async_trait,

The newest position in the log, or Position::BEGINNING if it is empty.

Source

fn subscribe( &self, from: Position, filter: Filter, ) -> Result<BoxStream<'static, Result<Recorded>>>

Catch up from from, then stay live.

There is no seam a consumer has to handle: the live tail is the same range read, resumed. Ordering is by position and nothing is skipped — see Position for why that holds without gap detection.

How the live half learns of a write is the backend’s business. The SQLite backend wakes subscribers on the same log directly, and falls back to polling for anything else, because SQLite has no notification a writer elsewhere could send.

“Anything else” includes a second log opened on the same file in this same process — the wake-up channel belongs to the log, not to the database. Nothing is lost either way; the difference is latency, and it is about three orders of magnitude [measured: 552µs woken directly against 552ms on a 600ms poll, tests/two_logs.rs]. Open the file once per process and share the log if that matters.

Source

fn checkpoint_load<'life0, 'life1, 'async_trait>( &'life0 self, consumer: &'life1 str, ) -> Pin<Box<dyn Future<Output = Result<Position>> + Send + 'async_trait>>
where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

How far consumer has got, or Position::BEGINNING if it has never reported.

Source

fn checkpoint_save<'life0, 'life1, 'async_trait>( &'life0 self, consumer: &'life1 str, at: Position, ) -> Pin<Box<dyn Future<Output = Result<()>> + Send + 'async_trait>>
where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Record how far consumer has got.

Callers that need the checkpoint to move in the same transaction as the work it accounts for must not use this — it is its own write. That is what a projection runner is for.

Provided Methods§

Source

fn export<'life0, 'life1, 'async_trait>( &'life0 self, from: Position, filter: &'life1 Filter, limit: usize, ) -> Pin<Box<dyn Future<Output = Result<Vec<ExportedEvent>>> + Send + 'async_trait>>
where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Read events out in position order, as they are stored.

The one read that does not run the upcaster chain. Every other read wants the current shape; a transfer wants the bytes, so the receiving log can hold exactly what this one held and run its own chain over them. Upcasting on the way out would bake this build’s reading of an old event into the copy and lose the original.

Page with from and limit, feeding the last returned position back in; a short batch is the end. See crate::transfer for what travels and what the receiving log reassigns.

The default declines, because a log with no stored form has nothing to hand over that another log could hold.

Source

fn import<'life0, 'async_trait>( &'life0 self, events: Vec<ExportedEvent>, ) -> Pin<Box<dyn Future<Output = Result<ImportReport>> + Send + 'async_trait>>
where Self: 'async_trait, 'life0: 'async_trait,

Write exported events into this log, in the order given.

seq and position are this log’s to assign; everything else travels unchanged, epoch_ms and _schema_version included. Keeping the version is what leaves an old event within reach of the upcaster written for it, and ImportReport::reproduced_coordinates reports whether the batch landed where it came from, so a migration can check rather than assume.

The default declines rather than appending one at a time. A backend with no transaction could only offer a partial import, and a transfer that stopped half way is worse than one that refused: from the outside there is no way to tell how far it got.

Dyn Compatibility§

This trait is dyn compatible.

In older versions of Rust, dyn compatibility was called "object safety".

Implementors§