Struct AnalyticsHandler 

pub struct AnalyticsHandler<P>
where
    P: Processor,
{ /* private fields */ }

Generic wrapper that implements Handler for any Processor with analytics batching.

This adapter wraps a Processor and provides the common batching and commit logic for writing analytics data to object stores. Uses a copy-on-write pattern for atomic commits: accumulated rows are only swapped after successful upload.

Note: Accumulated rows and pipeline config are stored on the AnalyticsStore, not on this handler. This makes handlers stateless and allows state to be shared.

Implementations

impl<P: Processor> AnalyticsHandler<P>

where P::Value: Row,

pub fn new(processor: P) -> Self

Create a new analytics handler and register the pipeline with the store.

Trait Implementations

impl<P, T> Handler for AnalyticsHandler<P>

where P: Processor<Value = T> + Send + Sync, T: Row + 'static,

type Store = AnalyticsStore

type Batch = Batch

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

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.

const MIN_EAGER_ROWS: usize = 50usize

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

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<P> Processor for AnalyticsHandler<P>

where P: Processor + Send + Sync, P::Value: Send + Sync,

const NAME: &'static str = P::NAME

Used to identify the pipeline in logs and metrics.

type Value = <P as Processor>::Value

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<Self::Value>>> + 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.