Class: Mammoth::Runtimes::BatchingRuntime

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

Overview

Adds configured batch submission to a selected delivery runtime.

Application streams one work item at a time into this runtime boundary. BatchingRuntime owns accumulation and submits complete or final partial batches through the selected adapter runtime's #process_many contract.

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(runtime:, batch_size:) ⇒ BatchingRuntime

Returns a new instance of BatchingRuntime.

Parameters:

  • runtime (#process_many) —

    selected delivery runtime

  • batch_size (Integer) —

    maximum work items submitted together



15
16
17
18
19
 
# File 'lib/mammoth/runtimes/batching_runtime.rb', line 15

def initialize(runtime:, batch_size:)
  @runtime = runtime
  @batch_size = batch_size
  @buffer = []
end

Instance Attribute Details

#batch_size ⇒ Object (readonly)

Returns the value of attribute batch_size.



11
12
13
 
# File 'lib/mammoth/runtimes/batching_runtime.rb', line 11

def batch_size
  @batch_size
end

#buffer ⇒ Object (readonly)

Returns the value of attribute buffer.



11
12
13
 
# File 'lib/mammoth/runtimes/batching_runtime.rb', line 11

def buffer
  @buffer
end

#runtime ⇒ Object (readonly)

Returns the value of attribute runtime.



11
12
13
 
# File 'lib/mammoth/runtimes/batching_runtime.rb', line 11

def runtime
  @runtime
end

Instance Method Details

#flush ⇒ Array

Submit the final partial batch.

Returns:

  • (Array) —

    processor results, or an empty array when no work is buffered



43
44
45
46
47
 
# File 'lib/mammoth/runtimes/batching_runtime.rb', line 43

def flush
  return [] if buffer.empty?

  runtime.process_many(buffer.shift(buffer.size))
end

#process(work) ⇒ Array

Buffer one work item and submit a full batch when ready.

Parameters:

  • work (CDC::Core::ChangeEvent, CDC::Core::TransactionEnvelope) —

    core work item

Returns:

  • (Array) —

    processor results when a batch is submitted, otherwise an empty array



25
26
27
28
29
30
 
# File 'lib/mammoth/runtimes/batching_runtime.rb', line 25

def process(work)
  buffer << work
  return [] if buffer.size < batch_size

  flush
end

#process_many(items) ⇒ Array

Submit work immediately without changing the buffered stream.

Parameters:

  • items (Array) —

    work items

Returns:

  • (Array) —

    processor results



36
37
38
 
# File 'lib/mammoth/runtimes/batching_runtime.rb', line 36

def process_many(items)
  runtime.process_many(items)
end

#shutdown ⇒ nil

Flush pending work and shut down the selected runtime when supported.

Returns:

  • (nil) 


52
53
54
55
56
 
# File 'lib/mammoth/runtimes/batching_runtime.rb', line 52

def shutdown
  flush
  runtime.shutdown if runtime.respond_to?(:shutdown)
  nil
end