class Consumer::Postgres::PositionStore

Public Class Methods

build(stream_name, session: nil, consumer_identifier: nil) click to toggle source
# File lib/consumer/postgres/position_store.rb, line 13
def self.build(stream_name, session: nil, consumer_identifier: nil)
  position_stream_name = StreamName.position_stream_name(stream_name, consumer_identifier: consumer_identifier)

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

Public Instance Methods

configure() click to toggle source
# File lib/consumer/postgres/position_store.rb, line 22
def configure
  MessageStore::Postgres::Get::Stream::Last.configure(
    self,
    session: session,
    attr_name: :read
  )

  Messaging::Postgres::Write.configure(self, session: session)
end
get() click to toggle source
# File lib/consumer/postgres/position_store.rb, line 36
def get
  message_data = read.(stream_name)

  return nil if message_data.nil?

  message = Messaging::Message::Import.(message_data, Recorded)

  message.position
end
location() click to toggle source
# File lib/consumer/postgres/position_store.rb, line 32
def location
  stream_name
end
put(position) click to toggle source
# File lib/consumer/postgres/position_store.rb, line 46
def put(position)
  message = Recorded.new
  message.position = position

  write.(message, stream_name)
end