Trait Handler 

pub trait Handler: Processor {
    type Store: SequentialStore;
    type Batch: Default + Send + Sync + 'static;

    const MAX_PENDING_ROWS: usize = 5000;
    const MAX_BATCH_CHECKPOINTS: usize = _;

    // Required methods
    fn batch(&self, batch: &mut Self::Batch, values: IntoIter<Self::Value>);
    fn commit<'a, 'life0, 'life1, 'life2, 'async_trait>(
        &'life0 self,
        batch: &'life1 Self::Batch,
        conn: &'life2 mut <Self::Store as Store>::Connection<'a>,
    ) -> 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;
}

Handlers implement the logic for a given indexing pipeline: How to process checkpoint data (by implementing Processor) into rows for their table, how to combine multiple rows into a single DB operation, and then how to write those rows atomically to the database.

The handler is also responsible for tuning the various parameters of the pipeline (provided as associated values).

Sequential handlers can only be used in sequential pipelines, where checkpoint data is processed out-of-order, but then gathered and written in order. If multiple checkpoints are available, the pipeline will attempt to combine their writes taking advantage of batching to avoid emitting redundant writes.

Back-pressure is handled by the bounded subscriber channel from the ingestion service, the same as concurrent pipelines: the channel blocks broadcaster sends when full, and the adaptive ingestion controller cuts fetch concurrency as the channel fills up.

Provided Associated Constants

const MAX_PENDING_ROWS: usize = 5000

Once this many rows are pending and the next checkpoint to write has arrived, the collector stops receiving processed checkpoints until it cuts a batch, which applies back-pressure to the processor and ingestion. While the next checkpoint is missing, the collector keeps receiving past this limit, because the missing checkpoint may still be in its input channel. Under back-pressure, a batch therefore gathers about this many rows, which is fewer than MAX_BATCH_CHECKPOINTS checkpoints when those hold more rows.

const MAX_BATCH_CHECKPOINTS: usize = _

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.

Required Associated Types

type Store: SequentialStore

type Batch: Default + Send + Sync + 'static

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).

Required Methods

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

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.

Returns BatchStatus::Ready if the batch is full and should be committed, or BatchStatus::Pending if the batch can accept more values.

Note: The handler can signal batch readiness via BatchStatus::Ready, but the framework may also decide to commit a batch based on the trait parameters above.

fn commit<'a, 'life0, 'life1, 'life2, 'async_trait>( &'life0 self, batch: &'life1 Self::Batch, conn: &'life2 mut <Self::Store as Store>::Connection<'a>, ) -> 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.

Dyn Compatibility

This trait is not dyn compatible.

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

Implementors