Class: Consumer::EventStore::PositionStore

Inherits:
Object
  • Object
show all
Includes:
PositionStore
Defined in:
lib/consumer/event_store/position_store.rb,
lib/consumer/event_store/position_store/updated.rb,
lib/consumer/event_store/position_store/stream_name.rb

Defined Under Namespace

Modules: StreamName Classes: Updated

Class Method Summary collapse

Instance Method Summary collapse

Class Method Details

.build(stream_name, session: nil, consumer_identifier: nil) ⇒ Object 



12
13
14
15
16
17
18
19
 
# File 'lib/consumer/event_store/position_store.rb', line 12

def self.build(stream_name, session: nil, consumer_identifier: nil)
  position_stream_name = StreamName.get(stream_name, consumer_identifier: consumer_identifier)

  instance = new(position_stream_name)
  MessageStore::EventStore::Session.configure(instance, session: session)
  instance.configure
  instance
end

Instance Method Details

#configure ⇒ Object 



21
22
23
24
 
# File 'lib/consumer/event_store/position_store.rb', line 21

def configure
  MessageStore::EventStore::Get::Last.configure(self, session: session)
  Messaging::EventStore::Write.configure(self, session: session)
end

#get ⇒ Object 



26
27
28
29
30
31
32
33
 
# File 'lib/consumer/event_store/position_store.rb', line 26

def get
  message_data = get_last.(stream_name)

  return nil if message_data.nil?

  message = Messaging::Message::Import.(message_data, Updated)
  message.position
end

#put(position) ⇒ Object 



35
36
37
38
39
40
 
# File 'lib/consumer/event_store/position_store.rb', line 35

def put(position)
  message = Updated.new
  message.position = position

  write.(message, stream_name)
end