Struct Effects 

pub struct Effects;

Pipeline marker for effects.

Trait Implementations

impl Handler for Effects

type Store = Store<RpcStoreSchema>

type Batch = Vec<Row>

A type to combine multiple Self::Value-s into. This can be used to avoid redundant writes by combining multiple rows into one (e.g. if one row supersedes another, the latter can be omitted).

fn batch(&self, batch: &mut Self::Batch, values: IntoIter<Row>)

Add values from processing a checkpoint to the current batch. Checkpoints are guaranteed to be presented to the batch in checkpoint order. The handler takes ownership of the iterator and consumes all values.

fn commit<'a, 'life0, 'life1, 'life2, 'async_trait>( &'life0 self, batch: &'life1 Self::Batch, conn: &'life2 mut Connection<'a, Schema>, ) -> Pin<Box<dyn Future<Output = Result<usize>> + Send + 'async_trait>>

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

Take a batch of values and commit them to the database, returning the number of rows affected.

const MIN_EAGER_ROWS: usize = 50usize

If at least this many rows are pending, the committer will commit them eagerly.

const MAX_PENDING_ROWS: usize = 5_000usize

Soft cap: once this many rows are pending, the collector stops eagerly draining its input channel and yields to the flush phase. Receive is never hard-gated — unlike concurrent pipelines, a missing predecessor may be buried in the input channel, and blocking receive would risk deadlock. The cap only bounds receive-to-flush latency in the happy path.

const MAX_BATCH_CHECKPOINTS: usize = 300usize

Maximum number of checkpoints to try and write in a single batch. The larger this number is, the more chances the pipeline has to merge redundant writes, but the longer each write transaction is likely to be.

impl Processor for Effects

const NAME: &'static str = "effects"

Used to identify the pipeline in logs and metrics.

type Value = Row

The type of value being inserted by the handler.

fn process<'life0, 'life1, 'async_trait>( &'life0 self, checkpoint: &'life1 Arc<Checkpoint>, ) -> Pin<Box<dyn Future<Output = Result<Vec<Row>>> + Send + 'async_trait>>

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

The processing logic for turning a checkpoint into rows of the table.