Class: RubyEventStore::ROM::EventRepository

Inherits:
Object
  • Object
show all
Defined in:
lib/ruby_event_store/rom/event_repository.rb

Instance Method Summary collapse

Constructor Details

#initialize(rom:, serializer:) ⇒ EventRepository

Returns a new instance of EventRepository.



6
7
8
9
10
11
 
# File 'lib/ruby_event_store/rom/event_repository.rb', line 6

def initialize(rom:, serializer:)
  @serializer = serializer
  @events = Repositories::Events.new(rom)
  @stream_entries = Repositories::StreamEntries.new(rom)
  @unit_of_work = UnitOfWork.new(rom.gateways.fetch(:default))
end

Instance Method Details

#append_to_stream(records, stream, expected_version) ⇒ Object 



13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
 
# File 'lib/ruby_event_store/rom/event_repository.rb', line 13

def append_to_stream(records, stream, expected_version)
  serialized_records = records.map { |record| record.serialize(@serializer) }
  event_ids = records.map(&:event_id)

  handle_unique_violation do
    @unit_of_work.call do |changesets|
      changesets << @events.create_changeset(serialized_records)
      changesets << @stream_entries.create_changeset(
        event_ids,
        stream,
        @stream_entries.resolve_version(stream, expected_version),
      )
    end
  end

  self
end

#count(specification) ⇒ Object 



78
79
80
 
# File 'lib/ruby_event_store/rom/event_repository.rb', line 78

def count(specification)
  @events.count(specification)
end

#delete_stream(stream) ⇒ Object 



59
60
61
 
# File 'lib/ruby_event_store/rom/event_repository.rb', line 59

def delete_stream(stream)
  @stream_entries.delete(stream)
end

#event_in_stream?(event_id, stream) ⇒ Boolean

Returns:

  • (Boolean) 


55
56
57
 
# File 'lib/ruby_event_store/rom/event_repository.rb', line 55

def event_in_stream?(event_id, stream)
  @stream_entries.event_in_stream?(event_id, stream)
end

#global_position(event_id) ⇒ Object 



51
52
53
 
# File 'lib/ruby_event_store/rom/event_repository.rb', line 51

def global_position(event_id)
  @events.global_position(event_id)
end

#has_event?(event_id) ⇒ Boolean

Returns:

  • (Boolean) 


63
64
65
66
67
68
 
# File 'lib/ruby_event_store/rom/event_repository.rb', line 63

def has_event?(event_id)
  @events.exist?(event_id)
rescue Sequel::DatabaseError => doh
  raise doh unless /PG::InvalidTextRepresentation.*uuid/.match?(doh.message)
  false
end

#last_stream_event(stream) ⇒ Object 



70
71
72
 
# File 'lib/ruby_event_store/rom/event_repository.rb', line 70

def last_stream_event(stream)
  @events.last_stream_event(stream, @serializer)
end

#link_to_stream(event_ids, stream, expected_version) ⇒ Object 



31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
 
# File 'lib/ruby_event_store/rom/event_repository.rb', line 31

def link_to_stream(event_ids, stream, expected_version)
  validate_event_ids(event_ids)

  handle_unique_violation do
    @unit_of_work.call do |changesets|
      changesets << @stream_entries.create_changeset(
        event_ids,
        stream,
        @stream_entries.resolve_version(stream, expected_version),
      )
    end
  end

  self
end

#position_in_stream(event_id, stream) ⇒ Object 



47
48
49
 
# File 'lib/ruby_event_store/rom/event_repository.rb', line 47

def position_in_stream(event_id, stream)
  @stream_entries.position_in_stream(event_id, stream)
end

#read(specification) ⇒ Object 



74
75
76
 
# File 'lib/ruby_event_store/rom/event_repository.rb', line 74

def read(specification)
  @events.read(specification, @serializer)
end

#streams_of(event_id) ⇒ Object 



91
92
93
 
# File 'lib/ruby_event_store/rom/event_repository.rb', line 91

def streams_of(event_id)
  @stream_entries.streams_of(event_id).map { |name| Stream.new(name) }
end

#update_messages(records) ⇒ Object 



82
83
84
85
86
87
88
89
 
# File 'lib/ruby_event_store/rom/event_repository.rb', line 82

def update_messages(records)
  validate_event_ids(records.map(&:event_id))

  @unit_of_work.call do |changesets|
    serialized_records = records.map { |record| record.serialize(@serializer) }
    changesets << @events.update_changeset(serialized_records)
  end
end