Class: Whodunit::Chronicles::Chronicler

Inherits:
CDC::Core::Processor
  • Object
show all
Defined in:
lib/whodunit/chronicles/chronicler.rb

Overview

Minimal CDC::Core processor that records change events into a ledger.

Chronicler is intentionally small. It consumes a CDC::Core::ChangeEvent, converts it into an immutable LedgerEntry, and appends that entry into the provided ledger. Scheduling, retries, backpressure, worker pools, and orchestration belong to CDC runtimes outside this gem.

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(ledger:, prepare: true, ensure_indexes: true, clock: Time) ⇒ Chronicler

Create a chronicler.

Parameters:

  • ledger (#append) —

    ledger-like object

  • prepare (Boolean) (defaults to: true) —

    whether to call ledger.prepare! during initialization

  • ensure_indexes (Boolean) (defaults to: true) —

    whether to call ledger.ensure_indexes! during initialization

  • clock (#now) (defaults to: Time) —

    clock used when creating ledger entries



25
26
27
28
29
30
 
# File 'lib/whodunit/chronicles/chronicler.rb', line 25

def initialize(ledger:, prepare: true, ensure_indexes: true, clock: Time)
  @ledger = ledger
  @clock = clock
  @ledger.prepare! if prepare && @ledger.respond_to?(:prepare!)
  @ledger.ensure_indexes! if ensure_indexes && @ledger.respond_to?(:ensure_indexes!)
end

Instance Attribute Details

#ledger ⇒ Object (readonly)

Returns ledger receiving immutable entries.

Returns:

  • (Object) —

    ledger receiving immutable entries



17
18
19
 
# File 'lib/whodunit/chronicles/chronicler.rb', line 17

def ledger
  @ledger
end

Instance Method Details

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

Process one CDC change event and append one ledger entry.

Parameters:

  • event (CDC::Core::ChangeEvent) —

    normalized CDC event

Returns:

  • (CDC::Core::ProcessorResult) —

    standardized processor result



36
37
38
39
40
41
42
43
44
45
46
47
 
# File 'lib/whodunit/chronicles/chronicler.rb', line 36

def process(event)
  entry = LedgerEntry.from_change_event(event, clock: @clock)
  partition_for(entry).append(entry)
  CDC::Core::ProcessorResult.success(event)
rescue StandardError => e
  CDC::Core::ProcessorResult.failure(
    e,
    event: event,
    processor: self.class.name,
    retryable: true
  )
end