Class: Mammoth::ConcurrentDeliveryRuntime

Inherits:
Object
  • Object
show all
Defined in:
lib/mammoth/concurrent_delivery_runtime.rb

Overview

Delivery runtime backed by CDC::Concurrent::ProcessorPool.

Mammoth keeps a single upstream replication stream and delegates downstream I/O fan-out to cdc-concurrent. This class is intentionally small so the runtime boundary remains easy to test and replace.

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(processor:, concurrency:, timeout:, preserve_order:, observer: CDC::Core::Observer.new) ⇒ ConcurrentDeliveryRuntime

Returns a new instance of ConcurrentDeliveryRuntime.

Parameters:

  • processor (#process) —

    delivery processor

  • concurrency (Integer) —

    number of concurrent delivery workers

  • timeout (Numeric, nil) —

    optional per-item timeout

  • preserve_order (Boolean) —

    preserve output order when supported

  • observer (CDC::Core::Observer) (defaults to: CDC::Core::Observer.new) —

    dispatch lifecycle observer



17
18
19
20
21
22
23
24
 
# File 'lib/mammoth/concurrent_delivery_runtime.rb', line 17

def initialize(processor:, concurrency:, timeout:, preserve_order:, observer: CDC::Core::Observer.new)
  @processor = processor
  @concurrency = concurrency
  @timeout = timeout
  @preserve_order = preserve_order
  @observer = observer
  @pool = build_pool
end

Instance Attribute Details

#concurrency ⇒ Object (readonly)

Returns the value of attribute concurrency.



10
11
12
 
# File 'lib/mammoth/concurrent_delivery_runtime.rb', line 10

def concurrency
  @concurrency
end

#observer ⇒ Object (readonly)

Returns the value of attribute observer.



10
11
12
 
# File 'lib/mammoth/concurrent_delivery_runtime.rb', line 10

def observer
  @observer
end

#pool ⇒ Object (readonly)

Returns the value of attribute pool.



10
11
12
 
# File 'lib/mammoth/concurrent_delivery_runtime.rb', line 10

def pool
  @pool
end

#preserve_order ⇒ Object (readonly)

Returns the value of attribute preserve_order.



10
11
12
 
# File 'lib/mammoth/concurrent_delivery_runtime.rb', line 10

def preserve_order
  @preserve_order
end

#processor ⇒ Object (readonly)

Returns the value of attribute processor.



10
11
12
 
# File 'lib/mammoth/concurrent_delivery_runtime.rb', line 10

def processor
  @processor
end

#timeout ⇒ Object (readonly)

Returns the value of attribute timeout.



10
11
12
 
# File 'lib/mammoth/concurrent_delivery_runtime.rb', line 10

def timeout
  @timeout
end

Instance Method Details

#process_many(items) ⇒ Array<Object>

Process many work items through cdc-concurrent.

Parameters:

  • items (Array<Object>) —

    CDC work units

Returns:

  • (Array<Object>) —

    processor results



30
31
32
33
34
35
36
37
 
# File 'lib/mammoth/concurrent_delivery_runtime.rb', line 30

def process_many(items)
  return [] if items.empty?

  items.each { |item| observer.dispatch_started(item) }
  pool.process_many(items).tap do |results|
    results.each { |result| observe_result(result) }
  end
end

#shutdown ⇒ nil

Shutdown the underlying runtime when supported.

Returns:

  • (nil) 


42
43
44
45
 
# File 'lib/mammoth/concurrent_delivery_runtime.rb', line 42

def shutdown
  pool.shutdown if pool.respond_to?(:shutdown)
  nil
end