Class: Hecks::Ports::Projection::Worker

Inherits:
Object
  • Object
show all
Defined in:
lib/hecks/ports/projection.rb

Constant Summary collapse

VALID_POLICIES =

The only two policies anything in this codebase ever passes (bin/project, every spec) — there is no third, legitimate "lenient append" policy on record anywhere. Before this, any value OTHER than the exact symbol :strict silently fell through the consistent? check below and appended onto divergent history without a word — not just a real typo like :strikt, but a caller-supplied String "strict" too (this duck-typed fine via policy.to_sym, but that was luck, not a contract: nothing here declared what a valid policy even was). Refusing loudly at construction, once, for anything outside this list turns a silent no-op into an immediate, named error — the "refuse rather than silently skip" reading of L1, since :strict really is meant to be the only enforcing contract.

%i[refresh strict].freeze

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(authoritative, projection, policy: :refresh) ⇒ Worker

Returns a new instance of Worker.



47
48
49
50
51
52
53
54
55
56
 
# File 'lib/hecks/ports/projection.rb', line 47

def initialize(authoritative, projection, policy: :refresh)
  @authoritative = authoritative
  @projection = projection
  @policy = policy.to_sym
  unless VALID_POLICIES.include?(@policy)
    raise ArgumentError,
          "unknown projection catch_up! policy #{@policy.inspect} — expected one of " \
          "#{VALID_POLICIES.map(&:inspect).join(' or ')}"
  end
end

Instance Attribute Details

#projection ⇒ Object (readonly)

Returns the value of attribute projection.



30
31
32
 
# File 'lib/hecks/ports/projection.rb', line 30

def projection
  @projection
end

Instance Method Details

#catch_up! ⇒ Object

Invoke from a separate process or scheduler. The command-side write path never calls this method.



60
61
62
63
64
65
66
67
68
69
70
71
72
 
# File 'lib/hecks/ports/projection.rb', line 60

def catch_up!
  entries = Queue.new(@authoritative).entries
  present = @projection.entries
  if @policy == :refresh
    @projection.reset!
    present = []
  end
  unless consistent?(entries, present)
    raise Runtime::WiringError, "projection history does not match its authoritative history" if @policy == :strict
  end
  entries.drop(present.length).each { |entry| @projection.append(entry); @projection.project(entry) }
  @projection
end

#checkpoint ⇒ Object 



74
 
# File 'lib/hecks/ports/projection.rb', line 74

def checkpoint = @projection.entries.length