Class: Philiprehberger::Batch::Processor

Inherits:
Object
  • Object
show all
Defined in:
lib/philiprehberger/batch/processor.rb

Overview

Processes collections in chunks with progress tracking and error collection.

Instance Method Summary collapse

Constructor Details

#initialize(size: 100, concurrency: 1, retries: 0, timeout_per_chunk: nil) ⇒ Processor

Returns a new instance of Processor.

Parameters:

  • size (Integer) (defaults to: 100) —

    chunk size

  • concurrency (Integer) (defaults to: 1) —

    number of concurrent workers (default: 1)

  • retries (Integer) (defaults to: 0) —

    max retries per failed item (default: 0)

  • timeout_per_chunk (Numeric, nil) (defaults to: nil) —

    optional per-chunk timeout in seconds

Raises:



13
14
15
16
17
18
19
20
21
22
23
24
25
 
# File 'lib/philiprehberger/batch/processor.rb', line 13

def initialize(size: 100, concurrency: 1, retries: 0, timeout_per_chunk: nil)
  raise Error, 'size must be positive' unless size.is_a?(Integer) && size.positive?
  raise Error, 'concurrency must be positive' unless concurrency.is_a?(Integer) && concurrency.positive?
  raise Error, 'retries must be non-negative' unless retries.is_a?(Integer) && retries >= 0
  if !timeout_per_chunk.nil? && !(timeout_per_chunk.is_a?(Numeric) && timeout_per_chunk.positive?)
    raise Error, 'timeout_per_chunk must be a positive numeric'
  end

  @size = size
  @concurrency = concurrency
  @retries = retries
  @timeout_per_chunk = timeout_per_chunk
end

Instance Method Details

#call(collection, on_progress: nil) {|chunk| ... } ⇒ Result

Process a collection in batches.

Parameters:

  • collection (Array, Enumerable) —

    items to process

  • on_progress (Proc, nil) (defaults to: nil) —

    optional callback invoked after each chunk with a progress info hash

Yields:

  • (chunk) —

    block that receives a Chunk object for processing

Returns:

  • (Result) —

    processing result

Raises:



33
34
35
36
37
38
39
40
41
42
43
44
45
 
# File 'lib/philiprehberger/batch/processor.rb', line 33

def call(collection, on_progress: nil, &block)
  raise Error, 'a processing block is required' unless block

  items = collection.to_a
  slices = items.each_slice(@size).to_a
  @on_progress = on_progress

  if @concurrency > 1 && slices.size > 1
    call_concurrent(slices, items, &block)
  else
    call_sequential(slices, items, &block)
  end
end