Class: Akasha::Repository

Inherits:
Object
  • Object
show all
Defined in:
lib/akasha/repository.rb

Overview

Aggregate repository. Not meant to be used directly (see aggregate/syntax_helpers.rb) See specs for usage.

Constant Summary collapse

STREAM_NAME_SEP = 
'-'.freeze

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(store, namespace: nil) ⇒ Repository

Creates a new repository using the underlying store (e.g. MemoryEventStore).

  • namespace - optional namespace allowing for multiple applications to share the same Eventstore

    database without name conflicts



13
14
15
16
17
 
# File 'lib/akasha/repository.rb', line 13

def initialize(store, namespace: nil)
  @store = store
  @subscribers = []
  @namespace = namespace
end

Instance Attribute Details

#namespace ⇒ Object (readonly)

Returns the value of attribute namespace.



6
7
8
 
# File 'lib/akasha/repository.rb', line 6

def namespace
  @namespace
end

#store ⇒ Object (readonly)

Returns the value of attribute store.



6
7
8
 
# File 'lib/akasha/repository.rb', line 6

def store
  @store
end

Instance Method Details

#load_aggregate(klass, id) ⇒ Object

Loads an aggregate identified by id and klass from the repository. Returns an aggregate instance of class klass constructed by applying events from the corresponding stream.



22
23
24
25
26
27
28
29
30
31
32
 
# File 'lib/akasha/repository.rb', line 22

def load_aggregate(klass, id)
  agg = klass.new(id)

  start = 0
  page_size = 20
  stream(klass, id).read_events(start, page_size) do |events|
    agg.apply_events(events)
  end

  agg
end

#merge_all_by_event(into:, only:) ⇒ Object

Merges all streams into one, filtering the resulting stream so it only contains events with the specified names, using a projection.

Arguments:

`into` - name of the new stream
`only` - array of event names



61
62
63
 
# File 'lib/akasha/repository.rb', line 61

def merge_all_by_event(into:, only:)
  @store.merge_all_by_event(into: into, only: only, namespace: @namespace)
end

#save_aggregate(aggregate, concurrency: :none) ⇒ Object

Saves an aggregate to the repository, appending events to the corresponding stream.



35
36
37
38
39
40
41
 
# File 'lib/akasha/repository.rb', line 35

def save_aggregate(aggregate, concurrency: :none)
  changeset = aggregate.changeset
  events = changeset.events.map { |event| event.with_metadata(namespace: @namespace) }
  revision = aggregate.revision if concurrency == :optimistic
  stream(aggregate.class, changeset.aggregate_id).write_events(events, revision: revision)
  notify_subscribers(changeset.aggregate_id, events)
end

#subscribe(lambda = nil, &block) ⇒ Object

Subscribes to event streams passing either a lambda or a block. Example:

repo.subscribe do |aggregate_id, event|
  ... handle the event ...
end



49
50
51
52
 
# File 'lib/akasha/repository.rb', line 49

def subscribe(lambda = nil, &block)
  callable = lambda || block
  @subscribers << callable
end