Module: EntityStore

Included in:
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/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 Classes: Log, Substitute

Constant Summary collapse

Error = 
Class.new(RuntimeError)

Class Method Summary collapse

Instance Method Summary collapse

Class Method Details

.included(cls) ⇒ Object 



4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
 
# 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

    virtual :category
    virtual :reader_class
    virtual :projection_class
    virtual :reader_batch_size
    virtual :snapshot_class
    virtual :snapshot_interval

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

Instance Method Details

#delete_cache_record(id) ⇒ Object 



200
201
202
 
# File 'lib/entity_store/entity_store.rb', line 200

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

#fetch(id, include: nil) ⇒ Object 



163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
 
# File 'lib/entity_store/entity_store.rb', line 163

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 



91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
 
# File 'lib/entity_store/entity_store.rb', line 91

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 



158
159
160
161
 
# File 'lib/entity_store/entity_store.rb', line 158

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

#new_entity ⇒ Object 



181
182
183
184
185
186
187
188
189
190
191
192
193
194
 
# File 'lib/entity_store/entity_store.rb', line 181

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 



150
151
152
153
154
155
156
 
# File 'lib/entity_store/entity_store.rb', line 150

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

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



123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
 
# File 'lib/entity_store/entity_store.rb', line 123

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 



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

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