Class: CDC::Concurrent::Runtime

Inherits:
Object
  • Object
show all
Defined in:
lib/cdc/concurrent/runtime.rb

Overview

High-level concurrent runtime facade for cdc-core processors.

Instance Method Summary collapse

Constructor Details

#initialize(processor:, concurrency: 100, timeout: nil, preserve_order: true) ⇒ Runtime

Returns a new instance of Runtime.

Parameters:

  • processor (CDC::Core::Processor)
  • concurrency (Integer) (defaults to: 100)
  • timeout (Float, nil) (defaults to: nil)
  • preserve_order (Boolean) (defaults to: true) 


11
12
13
14
15
16
 
# File 'lib/cdc/concurrent/runtime.rb', line 11

def initialize(processor:, concurrency: 100, timeout: nil, preserve_order: true)
  @processor_pool = ProcessorPool.new(processor:, concurrency:, timeout:, preserve_order:)
  @transaction_pool = TransactionPool.new(processor:, concurrency:, timeout:, preserve_order:)
  @router = Router.new(processor_pool: @processor_pool, transaction_pool: @transaction_pool)
  @shutdown = false
end

Instance Method Details

#process(item) ⇒ Object

Parameters:

  • item (Object)

Returns:

  • (Object)

Raises:



20
21
22
23
24
 
# File 'lib/cdc/concurrent/runtime.rb', line 20

def process(item)
  raise ShutdownError, "runtime has been shut down" if @shutdown

  @router.process(item)
end

#process_many(events) ⇒ Array<CDC::Core::ProcessorResult>

Parameters:

  • events (Array<CDC::Core::ChangeEvent>)

Returns:

  • (Array<CDC::Core::ProcessorResult>) 


28
29
30
 
# File 'lib/cdc/concurrent/runtime.rb', line 28

def process_many(events)
  process(events)
end

#process_transaction(transaction) ⇒ CDC::Core::ProcessorResult

Parameters:

  • transaction (CDC::Core::TransactionEnvelope)

Returns:

  • (CDC::Core::ProcessorResult) 


34
35
36
 
# File 'lib/cdc/concurrent/runtime.rb', line 34

def process_transaction(transaction)
  process(transaction)
end

#shutdown ⇒ void

This method returns an undefined value.



39
40
41
42
43
44
45
 
# File 'lib/cdc/concurrent/runtime.rb', line 39

def shutdown
  return if @shutdown

  @shutdown = true
  @processor_pool.shutdown
  @transaction_pool.shutdown
end