Struct QueuedRepository 

pub struct QueuedRepository<R, L = InMemoryLockManager> { /* private fields */ }

Repository wrapper that serializes access with process-local per-stream locks.

Locking reads (get and get_many) intentionally keep matching locks held after returning. Call commit to release those locks after a successful write, or call abort/unlock when the loaded entity is no longer being written. Dropping a loaded entity without commit or abort leaves its in-memory lock held until an explicit unlock.

Commit releases held locks only after the inner repository succeeds. On commit errors, locks remain held so callers can inspect state, retry, or explicitly abort.

Implementations

impl<R> QueuedRepository<R>

pub fn new(inner: R) -> Self

impl<R, L> QueuedRepository<R, L>

pub fn inner(&self) -> &R

Access the inner repository.

pub fn lock_manager(&self) -> &L

Access the lock manager.

impl<R, L: LockManager> QueuedRepository<R, L>

pub fn with_lock_manager(inner: R, lock_manager: L) -> Self

Create a QueuedRepository with a custom async lock manager.

Trait Implementations

impl<R: Clone, L> Clone for QueuedRepository<R, L>

fn clone(&self) -> Self

Returns a duplicate of the value.

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

Performs copy-assignment from source.

impl<R, L> GetAllWithOpts for QueuedRepository<R, L>

where R: GetStream, L: LockManager,

fn get_streams_with<'a>( &'a self, identities: &'a [StreamIdentity], opts: ReadOpts, ) -> impl Future<Output = Result<Vec<Entity>, RepositoryError>> + Send + 'a

impl<R, L> GetStream for QueuedRepository<R, L>

where R: GetStream, L: LockManager,

fn get_stream<'a>( &'a self, identity: &'a StreamIdentity, ) -> impl Future<Output = Result<Option<Entity>, RepositoryError>> + Send + 'a

fn get_streams<'a>( &'a self, identities: &'a [StreamIdentity], ) -> impl Future<Output = Result<Vec<Entity>, RepositoryError>> + Send + 'a

Load the streams for the provided identities, skipping missing ones.

fn get_stream_tail<'a>( &'a self, identity: &'a StreamIdentity, after_version: u64, ) -> impl Future<Output = Result<Option<Entity>, RepositoryError>> + Send + 'a

Load only the events with sequence > after_version as a tail-only Entity (see Entity::load_tail_from_history).

impl<R, L> GetWithOpts for QueuedRepository<R, L>

where R: GetStream, L: LockManager,

fn get_stream_with<'a>( &'a self, identity: &'a StreamIdentity, opts: ReadOpts, ) -> impl Future<Output = Result<Option<Entity>, RepositoryError>> + Send + 'a

impl<R, L> HasOutboxStore for QueuedRepository<R, L>

where R: HasOutboxStore,

type OutboxStore = <R as HasOutboxStore>::OutboxStore

The concrete outbox store this repository produces.

fn outbox_store(&self) -> Self::OutboxStore

Produce a handle to the durable outbox store.

impl<R, L> InboxStore for QueuedRepository<R, L>

where R: InboxStore, L: LockManager,

fn inbox_contains<'a>( &'a self, consumer: &'a str, message_id: &'a str, ) -> impl Future<Output = Result<bool, RepositoryError>> + Send + 'a

fn purge_inbox_older_than( &self, age: Duration, ) -> impl Future<Output = Result<u64, RepositoryError>> + Send

Purge inbox receipts older than age, returning the number removed.

impl<R, L> ReadModelWritePlanStore for QueuedRepository<R, L>

where R: ReadModelWritePlanStore, L: LockManager,

fn read_model_capabilities(&self) -> TableAdapterCapabilities

fn commit_write_plan( &self, plan: TableWritePlan, ) -> impl Future<Output = Result<TableCommitOutcome, TableStoreError>> + Send + '_

impl<R, L> RelationalReadModelQueryStore for QueuedRepository<R, L>

where R: RelationalReadModelQueryStore, L: LockManager,

fn read_model_query_capabilities(&self) -> ReadModelQueryCapabilities

fn load_graph( &self, request: ReadModelLoadRequest, ) -> impl Future<Output = Result<ReadModelLoadGraph, TableStoreError>> + Send + '_

impl<R, L> SnapshotStore for QueuedRepository<R, L>

where R: SnapshotStore, L: LockManager,

fn get_snapshot<'a>( &'a self, identity: &'a StreamIdentity, ) -> impl Future<Output = Result<Option<SnapshotRecord>, RepositoryError>> + Send + 'a

fn get_snapshots<'a>( &'a self, identities: &'a [StreamIdentity], ) -> impl Future<Output = Result<Vec<SnapshotRecord>, RepositoryError>> + Send + 'a

Load the snapshots for the provided identities, skipping identities without one. Each returned record carries its own aggregate type/id, so callers can pair records back to identities.

fn save_snapshot<'a>( &'a self, identity: &'a StreamIdentity, record: SnapshotRecord, ) -> impl Future<Output = Result<(), RepositoryError>> + Send + 'a

fn delete_snapshot<'a>( &'a self, identity: &'a StreamIdentity, ) -> impl Future<Output = Result<bool, RepositoryError>> + Send + 'a

impl<R, L> TransactionalCommit for QueuedRepository<R, L>

where R: TransactionalCommit, L: LockManager,

fn commit_batch<'a>( &'a self, batch: CommitBatch<'a>, ) -> impl Future<Output = Result<(), RepositoryError>> + Send + 'a

impl<R, L: LockManager> UnlockableRepository for QueuedRepository<R, L>

where R: Send + Sync,

fn unlock<'a>( &'a self, identity: &'a StreamIdentity, ) -> impl Future<Output = Result<(), RepositoryError>> + Send + 'a

Release the lock held for a stream.

fn abort<'a>( &'a self, identity: &'a StreamIdentity, ) -> impl Future<Output = Result<(), RepositoryError>> + Send + 'a

Release a lock for an aborted load (alias for unlock).