Class: Mammoth::DeliveryProcessor

Inherits:
CDC::Core::Processor
  • Object
show all
Defined in:
lib/mammoth/delivery_processor.rb

Overview

Adapter object used by CDC::Concurrent::ProcessorPool.

The processor keeps cdc-concurrent integration narrow: cdc-concurrent owns I/O-heavy fan-out mechanics, while DeliveryWorker owns Mammoth relay semantics such as retries, dead letters, and checkpoint writes.

Instance Attribute Summary collapse

Class Method Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(delivery_worker:, delivery_unit: :event, progress_coordinator: nil) ⇒ DeliveryProcessor

Returns a new instance of DeliveryProcessor.

Parameters:



40
41
42
43
44
45
 
# File 'lib/mammoth/delivery_processor.rb', line 40

def initialize(delivery_worker:, delivery_unit: :event, progress_coordinator: nil)
  super()
  @delivery_worker = delivery_worker
  @delivery_unit = delivery_unit.to_sym
  @progress_coordinator = progress_coordinator
end

Instance Attribute Details

#delivery_unit ⇒ Object (readonly)

Returns the value of attribute delivery_unit.



35
36
37
 
# File 'lib/mammoth/delivery_processor.rb', line 35

def delivery_unit
  @delivery_unit
end

#delivery_worker ⇒ Object (readonly)

Returns the value of attribute delivery_worker.



35
36
37
 
# File 'lib/mammoth/delivery_processor.rb', line 35

def delivery_worker
  @delivery_worker
end

#progress_coordinator ⇒ Object (readonly)

Returns the value of attribute progress_coordinator.



35
36
37
 
# File 'lib/mammoth/delivery_processor.rb', line 35

def progress_coordinator
  @progress_coordinator
end

Class Method Details

.concurrent_safe ⇒ Boolean

Returns true when this processor has explicitly opted in to cdc-concurrent execution.

Returns:

  • (Boolean) —

    true when this processor has explicitly opted in to cdc-concurrent execution.



30
31
32
 
# File 'lib/mammoth/delivery_processor.rb', line 30

def concurrent_safe?
  @concurrent_safe == true
end

.concurrent_safe! ⇒ true

Mark this processor as safe for CDC::Concurrent::ProcessorPool.

DeliveryProcessor itself is intentionally stateless after initialization; per-delivery retry, dead-letter, and checkpoint behavior remains owned by the injected DeliveryWorker.

Returns:

  • (true) 


20
21
22
 
# File 'lib/mammoth/delivery_processor.rb', line 20

def concurrent_safe!
  @concurrent_safe = true
end

.concurrent_safe? ⇒ Boolean

Returns true when this processor has explicitly opted in to cdc-concurrent execution.

Returns:

  • (Boolean) —

    true when this processor has explicitly opted in to cdc-concurrent execution.



26
27
28
 
# File 'lib/mammoth/delivery_processor.rb', line 26

def concurrent_safe?
  @concurrent_safe == true
end

Instance Method Details

#call(work) ⇒ CDC::Core::ProcessorResult

Process one work item from CDC::Concurrent::ProcessorPool.

Parameters:

  • work (Object) —

    event or transaction envelope

Returns:

  • (CDC::Core::ProcessorResult) —

    normalized processor result



58
59
60
 
# File 'lib/mammoth/delivery_processor.rb', line 58

def call(work)
  process(work)
end

#concurrent_safe? ⇒ Boolean Also known as: concurrent_safe

Returns true when this processor instance is safe for concurrent execution.

Returns:

  • (Boolean) —

    true when this processor instance is safe for concurrent execution.



48
49
50
 
# File 'lib/mammoth/delivery_processor.rb', line 48

def concurrent_safe?
  self.class.concurrent_safe?
end

#process(work) ⇒ CDC::Core::ProcessorResult

Process one work item using the configured delivery unit.

Parameters:

  • work (Object) —

    event or transaction envelope

Returns:

  • (CDC::Core::ProcessorResult) —

    normalized processor result



66
67
68
69
70
71
72
 
# File 'lib/mammoth/delivery_processor.rb', line 66

def process(work)
  summary = deliver(work)
  progress_coordinator&.complete(work)
  build_result(work, summary)
rescue StandardError => e
  failure_result(work, e, retryable: e.is_a?(DeliveryError))
end