Class: TCB::EventStore::ActiveRecord

Inherits:
Object
  • Object
show all
Defined in:
lib/tcb/event_store/active_record.rb

Instance Method Summary collapse

Constructor Details

#initialize ⇒ ActiveRecord

Returns a new instance of ActiveRecord.



8
9
10
 
# File 'lib/tcb/event_store/active_record.rb', line 8

def initialize
  @mutex = Mutex.new
end

Instance Method Details

#append(stream_id:, events:, occurred_at: Time.now) ⇒ Object 



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
 
# File 'lib/tcb/event_store/active_record.rb', line 12

def append(stream_id:, events:, occurred_at: Time.now)
  @mutex.synchronize do
    next_ver = next_version(stream_id)

    envelopes = events.map.with_index(next_ver) do |event, version|
      event_id = SecureRandom.uuid

      event_record_for(stream_id).create!(
        event_id:    event_id,
        stream_id:   stream_id,
        version:     version,
        event_type:  event.class.name,
        payload:     serialize(event),
        occurred_at: occurred_at
      )

      EventStreamEnvelope.new(
        event:       event,
        event_id:    event_id,
        stream_id:   stream_id,
        version:     version,
        occurred_at: occurred_at
      )
    end

    envelopes
  end
end

#read(stream_id, from_version: nil, to_version: nil, occurred_after: nil, limit: nil, order: :asc) ⇒ Object 



41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
 
# File 'lib/tcb/event_store/active_record.rb', line 41

def read(stream_id, from_version: nil, to_version: nil, occurred_after: nil, limit: nil, order: :asc)
  scope = event_record_for(stream_id)
    .where(stream_id: stream_id)
    .order(version: order)

  scope = scope.where("version >= ?", from_version) if from_version
  scope = scope.where("version <= ?", to_version) if to_version
  scope = scope.where("occurred_at > ?", occurred_after) if occurred_after
  scope = scope.limit(limit) if limit

  scope.map do |record|
    EventStreamEnvelope.new(
      event:       deserialize(record.payload),
      event_id:    record.event_id,
      stream_id:   record.stream_id,
      version:     record.version,
      occurred_at: record.occurred_at
    )
  end
end