Class: Hecks::Adapters::Heki::SagaStore

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

Overview

THE OPTIONAL saga-persistence capability (§2), Heki's own shape — a SIBLING snapshot+journal file pair, built the exact same way an aggregate's own persistence already is: Snapshot/Journal (heki/snapshot.rb, heki/journal.rb) operate generically on @path/@journal_path/@entry_mirrors and never touch @aggregate, so this reuses them unchanged rather than re-deriving the same binary framing and crash-recovery replay.

Reserved file name (hecks_saga_instances.heki, matching the hecks_-prefix convention every other new saga table in this work uses) avoids colliding with any real aggregate's own storage_name. Lives in the SAME directory an aggregate's own .heki file would (File.dirname(@path), Heki's own call below) — which, since Heki's resolve_path has no per-domain component at all, is typically shared across EVERY domain booted from the same root. domain is therefore carried inside each record and filtered on read, the same reason Postgres's own hecks_saga_instances keeps an explicit domain column under schema isolation (§3).

ONE flat records hash, keyed by a composite string (Heki's own snapshot format is id-keyed, not tuple-keyed) — never exposed outside this class; each_saga yields the five real fields a caller actually wants, not the internal key shape.

Locked the same way an aggregate's own store is: with_lock (Snapshot, shared) serializes each save/delete's read-modify- write against @path's own lock file — a saga gets exactly the durability and concurrency-safety this adapter already gives its aggregates, no better, no worse.

Instance Method Summary collapse

Methods included from Journal

#entries

Constructor Details

#initialize(dir) ⇒ SagaStore

Returns a new instance of SagaStore.



41
42
43
44
45
 
# File 'lib/hecks/adapters/driven/heki/saga_store.rb', line 41

def initialize(dir)
  @path         = File.join(dir, "hecks_saga_instances.heki")
  @journal_path = "#{@path}.journal"
  @entry_mirrors = nil
end

Instance Method Details

#delete_saga(domain, process_manager, correlation) ⇒ Object 



62
63
64
65
66
67
68
69
70
71
72
 
# File 'lib/hecks/adapters/driven/heki/saga_store.rb', line 62

def delete_saga(domain, process_manager, correlation)
  key = key_for(domain, process_manager, correlation)

  with_lock do
    append_entry("delete", key, nil)
    current = replay_journal(read_snapshot)
    current.delete(key)
    write(current)
    @store = current
  end
end

#each_saga(domain) ⇒ Object 



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

def each_saga(domain)
  return enum_for(:each_saga, domain) unless block_given?

  store.each_value do |record|
    next unless record["domain"] == domain

    yield record["process_manager"], record["correlation"], record["state"],
          (record["memory"] || {}).transform_keys(&:to_sym),
          record["completed_compensations"] || []
  end
end

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



47
48
49
50
51
52
53
54
55
56
57
58
59
60
 
# File 'lib/hecks/adapters/driven/heki/saga_store.rb', line 47

def save_saga(domain, process_manager, correlation, state, memory, completed_compensations = [])
  key    = key_for(domain, process_manager, correlation)
  record = { "domain" => domain, "process_manager" => process_manager,
             "correlation" => correlation, "state" => state, "memory" => memory,
             "completed_compensations" => completed_compensations }

  with_lock do
    append_entry("save", key, record)
    current = replay_journal(read_snapshot)
    current[key] = record
    write(current)
    @store = current
  end
end