Struct BatchProcessor

struct BatchProcessor<P, C, S>
where
    P: RecordProcessor + Send + Sync + 'static,
    C: KinesisClientTrait + Send + Sync + Clone + 'static,
    S: CheckpointStore + Send + Sync + Clone + 'static,
{
    ctx: ProcessingContext<P, C, S>,
}

Fields

ctx: ProcessingContext<P, C, S>

Implementations

impl<P, C, S> BatchProcessor<P, C, S>

where P: RecordProcessor + Send + Sync + 'static, C: KinesisClientTrait + Send + Sync + Clone + 'static, S: CheckpointStore + Send + Sync + Clone + 'static,

async fn process_batch( &self, shard_id: &str, records: &[Record], shutdown_rx: &mut Receiver<bool>, ) -> Result<BatchProcessingResult>

async fn process_single_record( &self, record: &Record, shard_id: &str, shutdown_rx: &mut Receiver<bool>, ) -> Result<RecordProcessingResult<P::Item>>

async fn attempt_process_record( &self, record: &Record, shard_id: &str, attempt_number: u32, ) -> Result<Option<P::Item>, ProcessingError>

fn update_batch_result( &self, result: &mut BatchProcessingResult, processed_items: &mut Vec<P::Item>, record_result: RecordProcessingResult<P::Item>, )

async fn handle_checkpointing( &self, shard_id: &str, result: &BatchProcessingResult, processed_items: &[P::Item], shutdown_rx: &mut Receiver<bool>, ) -> Result<()>

async fn try_checkpoint( &self, shard_id: &str, sequence: &str, processed_items: &[P::Item], metadata: &CheckpointMetadata<'_>, ) -> Result<(), BeforeCheckpointError>

async fn send_success_event(&self, shard_id: &str, sequence: &str)

async fn send_failure_event( &self, shard_id: &str, sequence: &str, error: String, )

async fn send_attempt_event( &self, shard_id: &str, sequence: &str, attempt: u32, success: bool, error: Option<String>, )

async fn send_checkpoint_success(&self, shard_id: &str, sequence: &str)

async fn send_batch_metrics( &self, shard_id: &str, result: &BatchProcessingResult, duration: Duration, )