Class: Consumer::Postgres::PositionStore

Inherits:
Object
  • Object
show all
Includes:
Consumer::PositionStore
Defined in:
lib/consumer/postgres/position_store.rb,
lib/consumer/postgres/position_store/messages.rb,
lib/consumer/postgres/position_store/stream_name.rb

Defined Under Namespace

Modules: Messages, StreamName

Class Method Summary collapse

Instance Method Summary collapse

Class Method Details

.build(stream_name, session: nil) ⇒ Object 



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

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

  instance = new position_stream_name
  EventSource::Postgres::Session.configure instance, session: session
  instance.configure
  instance
end

Instance Method Details

#configure ⇒ Object 



21
22
23
24
25
26
27
28
29
 
# File 'lib/consumer/postgres/position_store.rb', line 21

def configure
  EventSource::Postgres::Get::Last.configure(
    self,
    session: session,
    attr_name: :read
  )

  Messaging::Postgres::Write.configure self, session: session
end

#get ⇒ Object 



31
32
33
34
35
36
37
38
39
 
# File 'lib/consumer/postgres/position_store.rb', line 31

def get
  event_data = read.(stream_name)

  return nil if event_data.nil?

  message = Messaging::Message::Import.(event_data, Messages::PositionUpdated)

  message.position
end

#put(position) ⇒ Object 



41
42
43
44
45
46
 
# File 'lib/consumer/postgres/position_store.rb', line 41

def put(position)
  message = Messages::PositionUpdated.new
  message.position = position

  write.(message, stream_name)
end