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

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

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

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

Returns a new instance of Worker.



32
33
34
35
36
 
# File 'lib/hecks/ports/projection.rb', line 32

def initialize(authoritative, projection, policy: :refresh)
  @authoritative = authoritative
  @projection = projection
  @policy = policy.to_sym
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.



40
41
42
43
44
45
46
47
48
49
50
51
52
 
# File 'lib/hecks/ports/projection.rb', line 40

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 



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

def checkpoint = @projection.entries.length