Struct PipelineAggregate 

pub struct PipelineAggregate { /* private fields */ }

Managing Processing Operations

Error Handling and Recovery

Pipeline Configuration Updates

Event Management

The aggregate generates domain events for all significant state changes:

• PipelineCreated: When a new pipeline is created
• PipelineUpdated: When pipeline configuration changes
• ProcessingStarted: When file processing begins
• ProcessingCompleted: When processing finishes successfully
• ProcessingFailed: When processing encounters errors

Concurrency and Thread Safety

The aggregate is designed for single-threaded access within a transaction boundary. Concurrent access should be managed through:

• Repository-level locking mechanisms
• Optimistic concurrency control using version numbers
• Event store transaction boundaries
• Application-level coordination

Performance Considerations

• Memory Usage: Scales with number of active processing contexts
• Event Storage: Uncommitted events held in memory until persistence
• Validation Overhead: All operations include business rule validation
• Version Tracking: Minimal overhead for optimistic concurrency control

Error Recovery

The aggregate provides robust error handling:

• Validation Errors: Prevent invalid state transitions
• Processing Failures: Tracked with detailed error information
• Event Application: Supports replay for crash recovery
• State Consistency: Maintains valid state even during failures

Implementations

impl PipelineAggregate

pub fn new(pipeline: Pipeline) -> Result<Self, PipelineError>

Creates a new pipeline aggregate

pub fn from_events(events: Vec<PipelineEvent>) -> Result<Self, PipelineError>

Loads aggregate from events (event sourcing)

pub fn pipeline(&self) -> &Pipeline

Gets the pipeline

pub fn version(&self) -> u64

Gets the aggregate version

pub fn uncommitted_events(&self) -> &[PipelineEvent]

Gets uncommitted events

pub fn mark_events_as_committed(&mut self)

Marks events as committed

pub fn update_pipeline( &mut self, updated_pipeline: Pipeline, ) -> Result<(), PipelineError>

Updates the pipeline configuration

pub fn start_processing( &mut self, input_path: String, output_path: String, file_size: u64, security_context: SecurityContext, ) -> Result<Uuid, PipelineError>

Starts processing a file

pub fn complete_processing( &mut self, processing_id: Uuid, metrics: ProcessingMetrics, output_size: u64, ) -> Result<(), PipelineError>

Completes processing

pub fn fail_processing( &mut self, processing_id: Uuid, error_message: String, error_code: String, stage_name: Option<String>, partial_metrics: Option<ProcessingMetrics>, ) -> Result<(), PipelineError>

Fails processing

pub fn active_processing_contexts(&self) -> &HashMap<Uuid, ProcessingContext>

Gets active processing contexts

pub fn get_processing_context( &self, processing_id: Uuid, ) -> Option<&ProcessingContext>

Gets a specific processing context

pub fn update_processing_context( &mut self, processing_id: Uuid, context: ProcessingContext, ) -> Result<(), PipelineError>

Updates a processing context

pub fn validate(&self) -> Result<(), PipelineError>

Validates the aggregate state

pub fn id(&self) -> Uuid

Gets the aggregate ID

pub fn has_uncommitted_events(&self) -> bool

Checks if the aggregate has uncommitted events

pub fn active_processing_count(&self) -> usize

Gets the number of active processing contexts

pub fn is_processing_active(&self) -> bool

Checks if processing is active

Trait Implementations

impl Clone for PipelineAggregate

fn clone(&self) -> PipelineAggregate

Returns a duplicate of the value.

fn clone_from(&mut self, source: &Self)

Performs copy-assignment from source.

impl Debug for PipelineAggregate

fn fmt(&self, f: &mut Formatter<'_>) -> Result

Formats the value using the given formatter.