Struct Store 

pub struct Store { /* private fields */ }

Reads from and writes to EventSourcingDB, through the client SDK.

Create one with Store::builder, from a client and the source to use for all events it writes:

use architecturekit::{Store, eventsourcingdb::Client};

let client = Client::new("http://localhost:3000".parse().unwrap(), "secret");
let store = Store::builder(client, "https://library.eventsourcingdb.io").build();

The store checks the hash of every event it reads, before any upcaster or state sees it, and an event whose hash does not match its content makes reading fail with Permanent::Unverified. To turn this off, use StoreBuilder::without_hash_verification.

Implementations

impl Store

pub async fn catch_up_projection<P: Projection>( &self, subjects: &Subjects, projection: &P, ) -> Result<(), ProjectionError>

Applies the events of the subjects that are stored so far to a projection, from the first one on, and returns. Use it to build a read model once, for a batch job or in a test, instead of following the events as they are written. Dropping the future stops it, so a read model that is built only in part never looks complete.

Errors

Fails with the first failure, of reading, of verifying an event, or of the projection, including a panic of the projection, which is ProjectionError::Panicked.

pub async fn catch_up_resumable_projection<P: ResumableProjection>( &self, subjects: &Subjects, projection: &P, ) -> Result<(), ProjectionError>

Applies the events of the subjects that are stored so far to a resumable projection, from its checkpoint on, saving the checkpoint after every batch, and returns. See catch_up_projection.

Errors

Fails as catch_up_projection does, and if the checkpoint can not be read or saved.

pub async fn catch_up_transactional_projection<P: TransactionalProjection>( &self, subjects: &Subjects, projection: &P, ) -> Result<(), ProjectionError>

Applies the events of the subjects that are stored so far to a transactional projection, from its checkpoint on, one transaction per batch, and returns. See catch_up_projection.

Errors

Fails as catch_up_projection does, and if the checkpoint can not be read, or a transaction can not be begun or committed. A transaction that fails is rolled back, and a failure of the rollback is reported alongside the failure that caused it.

pub fn start_projection<P: Projection>( &self, subjects: Subjects, projection: Arc<P>, token: CancellationToken, options: ProjectionOptions, ) -> ProjectionRun

Starts a projection on a task of its own, which applies the events of the subjects stored so far, from the first one on, and then follows them as they are written, until the token is cancelled, or until a failure that trying again will not fix. It returns the run, which tells how far the projection has got (see ProjectionRun).

Reading that fails in a way that may pass, which includes a stream of events that ends, and a failure of the projection of the category Category::Transient, is tried again from the last event applied on, after a delay (see StoreBuilder::reconnect_delays). Any other failure ends the run, and so does a panic of the projection, as ProjectionError::Panicked, rather than the task. Cancelling the token stops waiting for the next event at once, while an event that is being applied is applied to the end, and the run then stops without a failure.

It needs a Tokio runtime with the time driver, as the client SDK does.

Panics

A name of the options that is empty is a programming error, so this panics.

pub fn start_resumable_projection<P: ResumableProjection>( &self, subjects: Subjects, projection: Arc<P>, token: CancellationToken, options: ProjectionOptions, ) -> ProjectionRun

Starts a resumable projection as start_projection does, from its checkpoint on. While it catches up, it saves the checkpoint after every batch, and once it follows the events, after every event.

Panics

As start_projection does.

pub fn start_transactional_projection<P: TransactionalProjection>( &self, subjects: Subjects, projection: Arc<P>, token: CancellationToken, options: ProjectionOptions, ) -> ProjectionRun

Starts a transactional projection as start_projection does, from its checkpoint on, one transaction per batch while it catches up, and one per event once it follows the events.

Panics

As start_projection does.

impl Store

pub fn builder(client: Client, source: impl Into<String>) -> StoreBuilder

Creates a builder for a store that reads and writes with the given client, and writes its events with the given source.

pub async fn load<S: State>(&self, subject: &Subject) -> Result<S, LoadError>

Reads the state of a subject, from the events of the subject alone, starting from Default, or from the latest event of the type the state names with State::from_latest. Use it for a query that needs the state of a single subject, rather than a view across many.

Every event is verified before the upcasters of the state, and then the state, see it. Dropping the future before the state is read completely returns nothing, so part of a state never reaches a caller.

Errors

Fails transiently if the database can not be reached, and permanently if an event fails its verification, if the state has no rule for an event, if the data of an event does not match its type, if the ID of an event is not a revision or its subject not a subject, neither of which the database ever writes, or if the database refuses the request.

Panics

A state that starts from an event type it has no rule for is a programming error, so this panics (see State::from_latest).

pub async fn read<'a>( &'a self, subjects: &'a Subjects, options: &'a ReadOptions, ) -> Result<impl Stream<Item = Result<Event, ReadError>> + 'a, ReadError>

Hands out the events of the given subjects as they are stored, for a read that neither load nor a projection fits: one page of a long stream, the events up to a certain one, or a single event. Every event is verified like everything else the store reads, before the caller sees it. Neither upcasters nor states apply, since there is no state.

Without options, it hands out every event of the subjects, oldest first. The bounds and the order of ReadOptions narrow that down. The stream ends with the first failure, and dropping it stops reading.

Errors

Fails before the database is asked, as check_read_options does, with ReadError::EmptyRange for bounds that leave no room for an event. Otherwise it fails, or the stream ends, with ReadError::Transient if the database can not be reached, and with ReadError::Permanent if an event fails its verification or the database refuses the read.

Panics

Panics for options that contradict each other (see LowerBound::FromLatestEvent).

pub async fn execute<C, S, D>( &self, decider: &Decider<C, S, D>, command: C, ) -> Result<Vec<Event>, ExecuteError<D>>

where C: Command + Clone, S: State,

Executes a command: reads the state of its subject, lets the decider decide on it, and writes the events it decided on to the subject, under the preconditions the command declares. It returns the events as the database recorded them, or none if the decider decided on none.

Precondition::on_state_read is filled in with the last event read, so that the events are only written if the state they were decided on still holds. A conflict is reported, and the caller decides what to do about it, unless the store was built with StoreBuilder::conflict_retries and the conflict may come from the state that was read, in which case it decides again first.

The decider decides on variants of the events of the state, so the state has a rule for every event that is written, and can read the subject for the next command. If the data of one of the events can not be encoded as JSON, none of them is written, and executing fails permanently, naming the event type, the subject and the reason, since trying again would fail the same way.

Dropping the future before the write has begun writes nothing, also while the decider decides. Once the write has begun, it runs on a task of its own, as with Store::write, so that it is finished even if the caller stops waiting for it.

Errors

Fails with ExecuteError::Decider if the decider refuses the command, with ExecuteError::Transient, including a Transient::Conflict for a precondition that did not hold, or with ExecuteError::Permanent if nothing was written, and with ExecuteError::OutcomeUnknown if the events may have been stored nonetheless. The preconditions are checked before anything is read, as check_preconditions checks them.

Panics

A state that starts from an event type it has no rule for is a programming error, so this panics (see State::from_latest).

pub async fn write( &self, events: Vec<EventOn>, preconditions: Vec<Precondition>, ) -> Result<Vec<Event>, WriteError>

Writes events to one or several subjects at once, so that either all of them are written or none, under the given preconditions. Use it for what is not a decision on the state of a single subject, such as a result derived from many of them.

Like a command, a write declares at least one precondition, such as Precondition::on_pristine_subject, or Precondition::unconditionally to write without any. Precondition::on_state_read has nothing to guard, since a write reads no state. Writing no events writes nothing, but the preconditions are checked first.

Once the write has begun, it runs on a task of its own, so that it is finished even if the caller stops waiting for it: a write cut off in the middle would leave open whether the events were stored. If it fails after its caller has stopped waiting, the failure is logged.

A write knows no state, so every state that reads one of the subjects needs a variant for each event type written to it, or reading fails afterwards.

Errors

Fails with WriteError::Transient, including a Transient::Conflict for a precondition that did not hold, or with WriteError::Permanent if the write certainly stored nothing, and with WriteError::OutcomeUnknown if it may have been stored nonetheless.

pub async fn register_schemas( &self, schemas: impl IntoIterator<Item = EventSchema>, ) -> Result<(), RegisterSchemasError>

Registers the schemas of the given events with the database, such as those of BookEvent::schemas(). It is meant to be called on every start: for an event type the database knows already, it checks that the registered schema is the given one.

A registered schema can not change. If it differs from the given one, or if the database refuses the given one, for example because stored events of the type do not match it, this fails permanently. To change the shape of an event, introduce a new event type and an upcaster instead. A registration that has begun runs on a task of its own, as a write does.

Errors

Fails transiently if the database can not be reached, and permanently if a schema differs from the registered one, if the same event type comes with two different schemas, or if the database refuses a schema.

Trait Implementations

impl Debug for Store

fn fmt(&self, f: &mut Formatter<'_>) -> Result

Formats the value using the given formatter.