Struct KinesisProcessor

pub struct KinesisProcessor<P, C, S>
where
    P: RecordProcessor + Send + Sync + 'static,
    P::Item: Send + Sync + 'static,
    C: KinesisClientTrait + Send + Sync + Clone + 'static,
    S: CheckpointStore + Send + Sync + Clone + 'static,
{
    context: ProcessingContext<P, C, S>,
}

Main Kinesis stream processor

Handles the orchestration of:

• Shard discovery and management
• Record batch processing
• Checkpointing
• Monitoring
• Graceful shutdown

Examples

use go_zoom_kinesis::{KinesisProcessor, ProcessorConfig, RecordProcessor, ProcessorError};
use aws_sdk_kinesis::Client;
use go_zoom_kinesis::store::InMemoryCheckpointStore;

async fn run_processor(
    processor: impl RecordProcessor + 'static,
    client: Client,
    config: ProcessorConfig
) -> Result<(), ProcessorError> {
    let store = InMemoryCheckpointStore::new();
    let (shutdown_tx, shutdown_rx) = tokio::sync::watch::channel(false);

    let (processor, _monitoring_rx) = KinesisProcessor::new(
        config,
        processor,
        client,
        store
    );

    processor.run(shutdown_rx).await
}

Fields

context: ProcessingContext<P, C, S>

Implementations

impl<P, C, S> KinesisProcessor<P, C, S>

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

pub fn new( config: ProcessorConfig, processor: P, client: C, store: S, ) -> (Self, Option<Receiver<ProcessingEvent>>)

Creates a new processor instance

Arguments

• config - Processor configuration
• processor - Record processor implementation
• client - AWS Kinesis client
• store - Checkpoint storage implementation

Returns

Returns a tuple of the processor instance and an optional monitoring channel receiver

pub async fn run(&self, shutdown_rx: Receiver<bool>) -> Result<()>

Starts processing the Kinesis stream

Arguments

• shutdown - Channel receiver for shutdown signals

Returns

Returns Ok(()) on successful shutdown, or Error on processing failures

async fn process_stream(&self, shutdown_rx: &mut Receiver<bool>) -> Result<()>

Process all shards in the stream

async fn get_records_batch( ctx: &ProcessingContext<P, C, S>, shard_id: &str, iterator: &str, shutdown_rx: &mut Receiver<bool>, ) -> Result<(Vec<Record>, Option<String>)>

Get a batch of records from a shard

Arguments

• ctx - Processing context
• shard_id - ID of the shard being processed
• iterator - Shard iterator for getting records
• shutdown_rx - Channel receiver for shutdown signals

async fn initialize_checkpoint( ctx: &ProcessingContext<P, C, S>, shard_id: &str, ) -> Result<Option<String>>

Initialize checkpoint for a shard

Arguments

• ctx - Processing context
• shard_id - ID of the shard to initialize

async fn get_initial_iterator( ctx: &ProcessingContext<P, C, S>, shard_id: &str, checkpoint: &Option<String>, shutdown_rx: &mut Receiver<bool>, ) -> Result<String>

Get initial iterator for a shard

Arguments

• ctx - Processing context
• shard_id - ID of the shard
• checkpoint - Optional checkpoint to start from
• shutdown_rx - Channel receiver for shutdown signals

async fn process_batch( &self, shard_id: &str, iterator: &str, state: &mut ShardProcessingState, shutdown_rx: Receiver<bool>, ) -> Result<BatchResult>

Process a batch of records

async fn process_shard( &self, shard_id: &str, shutdown_rx: Receiver<bool>, ) -> Result<()>

Process a single shard

Arguments

• ctx - Processing context
• shard_id - ID of the shard to process
• shutdown_rx - Channel receiver for shutdown signals

fn handle_early_shutdown(shard_id: &str) -> Result<()>

Handle early shutdown request