class RubyEventStore::ROM::Memory::Relations::StreamEntries

Constants

DIRECTION_MAP
SERIALIZED_GLOBAL_STREAM_NAME

Public Instance Methods

by_event_id(event_id) click to toggle source
# File lib/ruby_event_store/rom/adapters/memory/relations/stream_entries.rb, line 53
def by_event_id(event_id)
  restrict(event_id: event_id)
end
by_event_type(types) click to toggle source
# File lib/ruby_event_store/rom/adapters/memory/relations/stream_entries.rb, line 57
def by_event_type(types)
  for_events(events.restrict(event_type: Array(types)))
end
by_stream(stream) click to toggle source
# File lib/ruby_event_store/rom/adapters/memory/relations/stream_entries.rb, line 49
def by_stream(stream)
  restrict(stream: normalize_stream_name(stream))
end
by_stream_and_event_id(stream, event_id) click to toggle source
# File lib/ruby_event_store/rom/adapters/memory/relations/stream_entries.rb, line 61
def by_stream_and_event_id(stream, event_id)
  restrict(stream: normalize_stream_name(stream), event_id: event_id).one!
end
create_changeset(tuples) click to toggle source
# File lib/ruby_event_store/rom/adapters/memory/relations/stream_entries.rb, line 28
def create_changeset(tuples)
  changeset(Changesets::CreateStreamEntries, tuples)
end
delete(tuple) click to toggle source
Calls superclass method
# File lib/ruby_event_store/rom/adapters/memory/relations/stream_entries.rb, line 45
def delete(tuple)
  super tuple.to_h
end
for_events(events) click to toggle source
# File lib/ruby_event_store/rom/adapters/memory/relations/stream_entries.rb, line 20
def for_events(events)
  restrict(event_id: events.map { |e| e[:id] })
end
insert(tuple) click to toggle source
Calls superclass method
# File lib/ruby_event_store/rom/adapters/memory/relations/stream_entries.rb, line 40
def insert(tuple)
  verify_uniquness!(tuple)
  super
end
max_position(stream) click to toggle source
# File lib/ruby_event_store/rom/adapters/memory/relations/stream_entries.rb, line 65
def max_position(stream)
  new(by_stream(stream).order(:position).dataset.reverse).project(:position).take(1).one
end
offset(num) click to toggle source
# File lib/ruby_event_store/rom/adapters/memory/relations/stream_entries.rb, line 32
def offset(num)
  num.zero? ? self : new(dataset.slice(num..-1) || [])
end
ordered(direction, stream, offset_entry_id = nil, stop_entry_id = nil) click to toggle source
# File lib/ruby_event_store/rom/adapters/memory/relations/stream_entries.rb, line 74
def ordered(direction, stream, offset_entry_id = nil, stop_entry_id = nil)
  reverse, operator_offset, operator_stop = DIRECTION_MAP[direction]

  raise ArgumentError, 'Direction must be :forward or :backward' if order.nil?

  order_columns = %i[position id]
  order_columns.delete(:position) if stream.global?

  query = by_stream(stream)
  query = query.restrict { |tuple| tuple[:id].public_send(operator_offset, offset_entry_id) } if offset_entry_id
  query = query.restrict { |tuple| tuple[:id].public_send(operator_stop, stop_entry_id) } if stop_entry_id
  query = query.order(*order_columns)
  query = new(query.dataset.reverse) if reverse

  query
end
take(num) click to toggle source
Calls superclass method
# File lib/ruby_event_store/rom/adapters/memory/relations/stream_entries.rb, line 36
def take(num)
  num.nil? ? self : super
end

Private Instance Methods

normalize_stream_name(stream) click to toggle source
# File lib/ruby_event_store/rom/adapters/memory/relations/stream_entries.rb, line 107
def normalize_stream_name(stream)
  stream.global? ? SERIALIZED_GLOBAL_STREAM_NAME : stream.name
end
verify_uniquness!(tuple) click to toggle source

Verifies uniqueness of [stream, event_id] and [stream, position]

# File lib/ruby_event_store/rom/adapters/memory/relations/stream_entries.rb, line 94
def verify_uniquness!(tuple)
  stream = tuple[:stream]
  attrs = %i[position event_id]
  attrs.delete(:position) if Stream.new(stream).global?

  attrs.each do |key|
    next if key == :position && tuple[key].nil?
    next if restrict(:stream => stream, key => tuple.fetch(key)).none?

    raise TupleUniquenessError.public_send(:"for_stream_and_#{key}", stream, tuple.fetch(key))
  end
end