Class: Hecks::Adapters::Heki

Inherits:
Object
  • Object
show all
Includes:
Journal, Snapshot
Defined in:
lib/hecks/adapters/driven/heki.rb,
lib/hecks/adapters/driven/heki/journal.rb,
lib/hecks/adapters/driven/heki/snapshot.rb,
lib/hecks/adapters/driven/heki/saga_store.rb

Overview

The file-backed store: a compressed snapshot (heki/snapshot.rb) plus an append-only journal beside it (heki/journal.rb). What stays here is the repository surface — find/all/save/delete, and the entry append/project pair the persistence port drives.

Defined Under Namespace

Modules: Journal, Snapshot Classes: Malformed, SagaStore

Constant Summary collapse

MAGIC = 
"HEKI"
HEADER_BYTES = 
8

Instance Attribute Summary collapse

Instance Method Summary collapse

Methods included from Journal

#entries

Constructor Details

#initialize(aggregate:, settings: {}, root: nil) ⇒ Heki

Returns a new instance of Heki.



28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
 
# File 'lib/hecks/adapters/driven/heki.rb', line 28

def initialize(aggregate:, settings: {}, root: nil)
  @aggregate = aggregate
  @path      = resolve_path(settings, root)
  @journal_path = "#{@path}.journal"
  @events    = []
  # THE OPTIONAL saga-persistence capability's own scoping (§2/§4)
  # — falls back to the aggregate's own name for a directly-
  # instantiated adapter (specs), same fallback shape Postgres's
  # own @domain already uses.
  @domain    = (
    if settings.key?(:domain)
      settings[:domain]
    elsif settings.key?("domain")
      settings["domain"]
    else
      aggregate.name
    end
  ).to_s

  FileUtils.mkdir_p(File.dirname(@path))
end

Instance Attribute Details

#aggregate ⇒ Object (readonly)

Returns the value of attribute aggregate.



26
27
28
 
# File 'lib/hecks/adapters/driven/heki.rb', line 26

def aggregate
  @aggregate
end

#path ⇒ Object (readonly)

Returns the value of attribute path.



26
27
28
 
# File 'lib/hecks/adapters/driven/heki.rb', line 26

def path
  @path
end

Instance Method Details

#all(order_by: nil, direction: :asc) ⇒ Object 



57
58
59
60
 
# File 'lib/hecks/adapters/driven/heki.rb', line 57

def all(order_by: nil, direction: :asc)
  records = store.sort_by { |id, _| id }.map { |id, record| instance(id, record) }
  InMemoryOrdering.ordered(records, aggregate: @aggregate, order_by: order_by, direction: direction)
end

#append(entry) ⇒ Object 



75
76
77
78
79
80
81
 
# File 'lib/hecks/adapters/driven/heki.rb', line 75

def append(entry)
  @entry_mirrors = entry.mirrors
  append_entry(entry.operation, entry.id, entry.state)
  entry
ensure
  @entry_mirrors = nil
end

#count ⇒ Object 



62
 
# File 'lib/hecks/adapters/driven/heki.rb', line 62

def count = store.size

#delete(id) ⇒ Object 



104
105
106
107
108
109
110
111
112
113
 
# File 'lib/hecks/adapters/driven/heki.rb', line 104

def delete(id)
  return false unless find(id)

  entry = Ports::Persistence::Entry.new(operation: "delete", id: id.to_s, state: nil)
  with_lock do
    append(entry)
    project(entry)
  end
  true
end

#delete_saga(process_manager:, correlation:) ⇒ Object 



127
128
129
 
# File 'lib/hecks/adapters/driven/heki.rb', line 127

def delete_saga(process_manager:, correlation:)
  saga_store.delete_saga(@domain, process_manager.to_s, correlation.to_s)
end

#each_saga(&block) ⇒ Object 



131
 
# File 'lib/hecks/adapters/driven/heki.rb', line 131

def each_saga(&block) = saga_store.each_saga(@domain, &block)

#events ⇒ Object 



117
 
# File 'lib/hecks/adapters/driven/heki.rb', line 117

def events = @events

#find(id) ⇒ Object 



50
51
52
53
54
55
 
# File 'lib/hecks/adapters/driven/heki.rb', line 50

def find(id)
  record = store[id.to_s]
  return nil unless record

  instance(id.to_s, record)
end

#project(entry) ⇒ Object

Reads fresh rather than trusting the memoized store — under with_lock, another process may have projected a snapshot since this one last read it, and mutating its stale copy would overwrite that write on disk rather than layer on top of it.



87
88
89
90
91
92
93
 
# File 'lib/hecks/adapters/driven/heki.rb', line 87

def project(entry)
  current = read
  entry.save? ? current[entry.id] = entry.state.dup : current.delete(entry.id)
  write(current)
  @store = current
  entry
end

#query(specification, args = {}, context: {}) ⇒ Object

registry: context[:registry] — Memory's own query already threads this through; Heki's own never did, which made none_in_state? (Ports::Query::InMemory) unconditionally return true (its own graceful "no registry, no way to look the target up" default) for EVERY none_in_state where-clause against a Heki-backed aggregate — silently excluding nothing, always, no matter the actual target state.



71
72
73
 
# File 'lib/hecks/adapters/driven/heki.rb', line 71

def query(specification, args = {}, context: {})
  Ports::Query::InMemory.execute(all, specification, args, registry: context[:registry])
end

#record_event(event) ⇒ Object 



115
 
# File 'lib/hecks/adapters/driven/heki.rb', line 115

def record_event(event) = @events << event

#save(instance) ⇒ Object 



95
96
97
98
99
100
101
102
 
# File 'lib/hecks/adapters/driven/heki.rb', line 95

def save(instance)
  entry = Ports::Persistence::Entry.new(operation: "save", id: instance.id.to_s, state: instance.state.dup)
  with_lock do
    append(entry)
    project(entry)
  end
  instance
end

#save_saga(process_manager:, correlation:, state:, memory:, completed_compensations: []) ⇒ Object

── the OPTIONAL saga-persistence capability (§2) — Heki's own shape (a sibling snapshot+journal file pair, SagaStore, heki/saga_store.rb) rather than a table in a store this adapter doesn't have.



123
124
125
 
# File 'lib/hecks/adapters/driven/heki.rb', line 123

def save_saga(process_manager:, correlation:, state:, memory:, completed_compensations: [])
  saga_store.save_saga(@domain, process_manager.to_s, correlation.to_s, state.to_s, memory, completed_compensations)
end