Module: EntityStore

Included in:

Controls::EntityStore::NoCategory::Example, Substitute

Defined in:

lib/entity_store/log.rb,
lib/entity_store/substitute.rb,
lib/entity_store/controls/id.rb,
lib/entity_store/entity_store.rb,
lib/entity_store/controls/entity.rb,
lib/entity_store/controls/reader.rb,
lib/entity_store/controls/record.rb,
lib/entity_store/controls/message.rb,
lib/entity_store/controls/version.rb,
lib/entity_store/controls/category.rb,
lib/entity_store/controls/snapshot.rb,
lib/entity_store/controls/specifier.rb,
lib/entity_store/controls/projection.rb,
lib/entity_store/controls/stream_name.rb,
lib/entity_store/controls/entity_store.rb,
lib/entity_store/controls/snapshot_interval.rb

Defined Under Namespace

Modules: Build, Controls, EntityMacro, ProjectionMacro, ReaderMacro, SnapshotMacro, SpecifierMacro Classes: Log, Substitute

Constant Summary

Error =

Class.new(RuntimeError)

Class Method Summary

• .included(cls) ⇒ Object

Instance Method Summary

• #delete_cache_record(id) ⇒ Object
• #fetch(id, include: nil) ⇒ Object (also: #project)
• #get(id, include: nil, &probe_action) ⇒ Object
• #get_version(id) ⇒ Object
• #new_entity ⇒ Object
• #next_position(position) ⇒ Object
• #refresh(entity, id, current_position, &probe_action) ⇒ Object
• #stream_name(id) ⇒ Object

Class Method Details

.included(cls) ⇒ Object

# File 'lib/entity_store/entity_store.rb', line 4

def self.included(cls)
  cls.class_exec do
    include Configure
    include Dependency
    include Virtual

    include Log::Dependency
    include Messaging::Category

    substitute_class = Class.new(Substitute)

    substitute_class.send :define_method, :entity_class do
      cls.entity_class
    end

    const_set :Substitute, substitute_class

    attr_accessor :session
    attr_accessor :new_entity_probe

    dependency :cache, EntityCache

    configure :store

    attr_accessor :category
    attr_accessor :specifier
    attr_accessor :snapshot_interval

    virtual :reader_class
    virtual :projection_class
    virtual :reader_batch_size
    virtual :snapshot_class

    virtual :configure

    extend Build
    extend EntityMacro
    extend ProjectionMacro
    extend ReaderMacro
    extend SnapshotMacro
    extend SpecifierMacro
  end
end

Instance Method Details

#delete_cache_record(id) ⇒ Object

# File 'lib/entity_store/entity_store.rb', line 216

def delete_cache_record(id)
  cache.internal_store.delete(id)
end

#fetch(id, include: nil) ⇒ Object Also known as: project

# File 'lib/entity_store/entity_store.rb', line 178

def fetch(id, include: nil)
  logger.trace(tag: :fetch) { "Fetching entity (ID: #{id.inspect}, Entity Class: #{entity_class.name})" }

  res = get(id, include: include)

  if res.nil?
    res = new_entity
  end

  if res.is_a?(Array) && res[0].nil?
    res[0] = new_entity
  end

  logger.info(tag: :fetch) { "Fetch entity done (ID: #{id.inspect}, Entity Class: #{entity_class.name})" }

  res
end

#get(id, include: nil, &probe_action) ⇒ Object

# File 'lib/entity_store/entity_store.rb', line 106

def get(id, include: nil, &probe_action)
  logger.trace(tag: :get) { "Getting entity (ID: #{id.inspect}, Entity Class: #{entity_class.name})" }

  record = cache.get id

  if record
    entity = record.entity
    version = record.version
    persisted_version = record.persisted_version
    persisted_time = record.persisted_time
  else
    entity = new_entity
  end

  current_version = refresh(entity, id, version, &probe_action)

  unless current_version.nil?
    record = cache.put(
      id,
      entity,
      current_version,
      persisted_version: persisted_version,
      persisted_time: persisted_time
    )
  end

  logger.info(tag: :get) { "Get entity done (ID: #{id.inspect}, Entity Class: #{entity_class.name}, Version: #{record&.version.inspect}, Time: #{record&.time.inspect})" }
  logger.info(tags: [:data, :entity]) { entity.pretty_inspect }

  EntityCache::Record.destructure(record, include)
end

#get_version(id) ⇒ Object

# File 'lib/entity_store/entity_store.rb', line 173

def get_version(id)
  _, version = get id, include: :version
  version
end

#new_entity ⇒ Object

# File 'lib/entity_store/entity_store.rb', line 197

def new_entity
  entity = nil
  if entity_class.respond_to? :build
    entity = entity_class.build
  else
    entity = entity_class.new
  end

  unless new_entity_probe.nil?
    new_entity_probe.(entity)
  end

  entity
end

#next_position(position) ⇒ Object

# File 'lib/entity_store/entity_store.rb', line 165

def next_position(position)
  unless position.nil?
    position + 1
  else
    nil
  end
end

#refresh(entity, id, current_position, &probe_action) ⇒ Object

# File 'lib/entity_store/entity_store.rb', line 138

def refresh(entity, id, current_position, &probe_action)
  logger.trace(tag: :refresh) { "Refreshing (ID: #{id.inspect}, Entity Class: #{entity_class.name}, Current Position #{current_position.inspect})" }
  logger.trace(tags: [:data, :entity]) { entity.pretty_inspect }

  stream_name = self.stream_name(id)

  start_position = next_position(current_position)

  project = projection_class.build(entity)

  logger.trace(tag: :refresh) { "Reading (Stream Name: #{stream_name}, Position: #{current_position})" }
  reader_class.(stream_name, position: start_position, batch_size: reader_batch_size, session: session) do |event_data|
    project.(event_data)
    current_position = event_data.position

    unless probe_action.nil?
      probe_action.(event_data)
    end
  end
  logger.debug(tag: :refresh) { "Read (Stream Name: #{stream_name}, Position: #{current_position.inspect})" }

  logger.debug(tag: :refresh) { "Refreshed (ID: #{id.inspect}, Entity Class: #{entity_class.name}, Current Position: #{current_position.inspect})" }
  logger.debug(tags: [:data, :entity]) { entity.pretty_inspect }

  current_position
end

#stream_name(id) ⇒ Object

# File 'lib/entity_store/entity_store.rb', line 212

def stream_name(id)
  MessageStore::StreamName.stream_name(category, id)
end