Class: Sequent::Core::AggregateRepository

Inherits:
Object
  • Object
show all
Defined in:
lib/sequent/core/aggregate_repository.rb

Overview

Repository for aggregates.

Implements the Unit-Of-Work and Identity-Map patterns to ensure each aggregate is only loaded once per transaction and that you always get the same aggregate instance back.

On commit all aggregates associated with the Unit-Of-Work are queried for uncommitted events. After persisting these events the uncommitted events are cleared from the aggregate.

The repository is keeps track of the Unit-Of-Work per thread, so can be shared between threads.

Defined Under Namespace

Classes: NonUniqueAggregateId

Constant Summary collapse

AGGREGATES_KEY =

Key used in thread local

'Sequent::Core::AggregateRepository::aggregates'.to_sym

Instance Method Summary collapse

Constructor Details

#initialize(event_store) ⇒ AggregateRepository

Returns a new instance of AggregateRepository.



25
26
27
28
 
# File 'lib/sequent/core/aggregate_repository.rb', line 25

def initialize(event_store)
  @event_store = event_store
  clear
end

Instance Method Details

#add_aggregate(aggregate) ⇒ Object

Adds the given aggregate to the repository (or unit of work).

Only when commit is called all aggregates in the unit of work are ‘processed’ and all uncammited_events are stored in the event_store



35
36
37
38
39
40
41
 
# File 'lib/sequent/core/aggregate_repository.rb', line 35

def add_aggregate(aggregate)
  if aggregates.has_key?(aggregate.id)
    raise NonUniqueAggregateId.new(aggregate, aggregates[aggregate.id]) unless aggregates[aggregate.id].equal?(aggregate)
  else
    aggregates[aggregate.id] = aggregate
  end
end

#clear ⇒ Object

Clears the Unit of Work.



79
80
81
 
# File 'lib/sequent/core/aggregate_repository.rb', line 79

def clear
  Thread.current[AGGREGATES_KEY] = {}
end

#commit(command) ⇒ Object

Gets all uncommitted_events from the ‘registered’ aggregates and stores them in the event store. The command is ‘attached’ for traceability purpose so we can see which command resulted in which events.

This is all abstracted away if you use the Sequent::Core::CommandService



70
71
72
73
74
75
76
 
# File 'lib/sequent/core/aggregate_repository.rb', line 70

def commit(command)
  all_events = []
  aggregates.each_value { |aggregate| all_events += aggregate.uncommitted_events }
  return if all_events.empty?
  aggregates.each_value { |aggregate| aggregate.clear_events }
  store_events command, all_events
end

#ensure_exists(aggregate_id, clazz) ⇒ Object

Throws exception if not exists.



44
45
46
 
# File 'lib/sequent/core/aggregate_repository.rb', line 44

def ensure_exists(aggregate_id, clazz)
  !load_aggregate(aggregate_id, clazz).nil?
end

#load_aggregate(aggregate_id, clazz) ⇒ Object

Loads aggregate by given id and class Returns the one in the current Unit Of Work otherwise loads it from history.

If we implement snapshotting this is the place.



52
53
54
55
56
57
58
59
60
61
 
# File 'lib/sequent/core/aggregate_repository.rb', line 52

def load_aggregate(aggregate_id, clazz)
  if aggregates.has_key?(aggregate_id)
    result = aggregates[aggregate_id]
    raise TypeError, "#{result.class} is not a #{clazz}" unless result.is_a?(clazz)
    result
  else
    events = @event_store.load_events(aggregate_id)
    aggregates[aggregate_id] = clazz.load_from_history(events)
  end
end