Class: CDC::Parallel::TransactionPool

Inherits:
Object
  • Object
show all
Defined in:
lib/cdc/parallel/transaction_pool.rb

Overview

Processes a TransactionEnvelope as a single ordering-preserving unit.

Instance Method Summary collapse

Constructor Details

#initialize(processor:, size: Etc.nprocessors, timeout: nil) ⇒ TransactionPool

Returns a new instance of TransactionPool.

Parameters:

  • processor (CDC::Core::Processor)
  • size (Integer) (defaults to: Etc.nprocessors)
  • timeout (Float, nil) (defaults to: nil) 


10
11
12
 
# File 'lib/cdc/parallel/transaction_pool.rb', line 10

def initialize(processor:, size: Etc.nprocessors, timeout: nil)
  @processor_pool = ProcessorPool.new(processor:, size:, timeout:)
end

Instance Method Details

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

Process all events inside a transaction envelope.

Parameters:

  • transaction (CDC::Core::TransactionEnvelope)

Returns:

  • (CDC::Core::ProcessorResult) 


18
19
20
21
22
23
24
25
 
# File 'lib/cdc/parallel/transaction_pool.rb', line 18

def process(transaction)
  results = @processor_pool.process_many(transaction.events).freeze
  failure = results.find(&:failure?)

  return CDC::Core::ProcessorResult.failure(failure.error, event: results) if failure

  CDC::Core::ProcessorResult.success(results)
end

#shutdown ⇒ void

This method returns an undefined value.

Shut down worker resources.



30
31
32
 
# File 'lib/cdc/parallel/transaction_pool.rb', line 30

def shutdown
  @processor_pool.shutdown
end