Struct Processor 

pub struct Processor<R: Reducer, D: Db, T, E: EventHandler<R::Event>> { /* private fields */ }

Drives a Reducer over a Db: applies local writes (exec) and ingests remote log entries, advancing the HLC and per-peer cursors and emitting change events through its EventHandler. This is the concrete engine behind the Store/SqlStore surface (open one with open); it also implements the sync traits.

Implementations

impl<R: Reducer, D: Db, T: LogTracker<R::Op>, E: EventHandler<R::Event>> Processor<R, D, T, E>

pub async fn open( reducer: R, db: D, prefix: &str, self_id: Uuid, ) -> Result<Self, ProcessorError<R::Error>>

Open a processor against db: set up HLC storage and the tracker (both namespaced by prefix), seed the clock and cursor view from persisted state, and take ownership of reducer.

pub async fn exec( &self, server_user_id: Option<Uuid>, op: R::Op, ) -> Result<(), ProcessorError<R::Error>>

Apply a local write: mint a fresh entry under self_id and ingest it, advancing self’s cursor.

pub fn event_handler(&self) -> &E

The handler this processor emits into — subscribe off it.

Trait Implementations

impl<R: Reducer, D: Db, T: Send + Sync, E: EventHandler<R::Event> + Send + Sync> HasCursors for Processor<R, D, T, E>

where E::Publish: Send + Sync,

fn cursors<'life0, 'async_trait>( &'life0 self, ) -> Pin<Box<dyn Future<Output = Result<PeerCursors, SyncError>> + Send + 'async_trait>>

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

Snapshot of the current cursor vector.

fn watch_cursors(&self) -> CursorStream

Live cursor progress: a first Snapshot, then Advanced deltas.

impl<R: Reducer, D: Db, T: LogTracker<R::Op>, E: EventHandler<R::Event> + Send + Sync> LogProcessor<<R as Reducer>::Op> for Processor<R, D, T, E>

where R::Error: Error + Send + Sync + 'static, E::Publish: Send + Sync,

fn apply<'life0, 'async_trait>( &'life0 self, peer: Uuid, index: u64, entry: DecodedEntry<R::Op>, ) -> Pin<Box<dyn Future<Output = Result<Applied, SyncError>> + Send + 'async_trait>>

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

Apply one entry at (peer, index), advancing the cursor for peer to index + 1. Idempotent (see the trait docs).

impl<R: Reducer, D: Db, T: HistoryTracker<R::Op>, E: EventHandler<R::Event> + Send + Sync> LogSource<<R as Reducer>::Op> for Processor<R, D, T, E>

where R::Error: Error + Send + Sync + 'static, E::Publish: Send + Sync,

fn read_since<'life0, 'async_trait>( &'life0 self, peer: Uuid, from: u64, ) -> Pin<Box<dyn Future<Output = Result<Vec<(u64, DecodedEntry<R::Op>)>, SyncError>> + Send + 'async_trait>>

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

A bounded batch of peer’s entries at or after from, ascending (expunged markers included). Empty means drained at from; the caller loops with an advancing from until then.

impl<R: Reducer, D: Db, T: LogTracker<R::Op>> SqlStore<<R as Reducer>::Op, <R as Reducer>::Event> for Processor<R, D, T, EventBus<R::Event>>

where R::Event: RoutableEvent, R::Error: Error + Send + Sync + 'static, <R::Event as RoutableEvent>::Target: Send + Sync,

fn query<'life0, 'life1, 'life2, 'async_trait>( &'life0 self, sql: &'life1 str, params: &'life2 [DbValue], ) -> Pin<Box<dyn Future<Output = Result<Vec<DbRow>, DbError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait,

Run a read-only query against the backend.

fn dialect(&self) -> SqlDialect

The backend’s SQL dialect — lets a caller build dialect-correct SQL (placeholder style, quoting) to hand back to query.

impl<R: Reducer, D: Db, T: LogTracker<R::Op>> Store<<R as Reducer>::Op, ProcessorError<Box<dyn Error + Send + Sync>>, <R as Reducer>::Event> for Processor<R, D, T, EventBus<R::Event>>

where R::Event: RoutableEvent, R::Error: Error + Send + Sync + 'static, <R::Event as RoutableEvent>::Target: Send + Sync,

fn exec<'life0, 'async_trait>( &'life0 self, server_user_id: Option<Uuid>, op: R::Op, ) -> Pin<Box<dyn Future<Output = Result<(), ProcessorError<BoxError>>> + Send + 'async_trait>>

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

Apply a local write, minting a fresh log entry for it.

fn watch( &self, target: <R::Event as RoutableEvent>::Target, ) -> Subscription<R::Event>

Subscribe to events routed to target; dropping the subscription unsubscribes.