class EventSource::Postgres::Write

Public Instance Methods

configure(session: nil) click to toggle source
# File lib/event_source/postgres/write.rb, line 8
def configure(session: nil)
  Put.configure(self, session: session)
end
write(batch, stream_name, expected_version: nil) click to toggle source
# File lib/event_source/postgres/write.rb, line 12
def write(batch, stream_name, expected_version: nil)
  logger.trace(tag: :write) { "Writing batch (Stream Name: #{stream_name}, Number of Events: #{batch.length}, Expected Version: #{expected_version.inspect})" }

  unless expected_version.nil?
    expected_version = ExpectedVersion.canonize(expected_version)
  end

  last_position = nil
  put.session.transaction do
    batch.each do |event_data|
      last_position = write_event_data(event_data, stream_name, expected_version: expected_version)

      unless expected_version.nil?
        expected_version += 1
      end
    end
  end

  logger.debug(tag: :write) { "Wrote batch (Stream Name: #{stream_name}, Number of Events: #{batch.length}, Expected Version: #{expected_version.inspect})" }

  last_position
end
write_event_data(event_data, stream_name, expected_version: nil) click to toggle source
# File lib/event_source/postgres/write.rb, line 35
def write_event_data(event_data, stream_name, expected_version: nil)
  logger.trace(tag: :write) { "Writing event data (Stream Name: #{stream_name}, Type: #{event_data.type}, Expected Version: #{expected_version.inspect})" }
  logger.trace(tags: [:data, :event_data, :write]) { event_data.pretty_inspect }

  put.(event_data, stream_name, expected_version: expected_version).tap do
    logger.debug(tag: :write) { "Wrote event data (Stream Name: #{stream_name}, Type: #{event_data.type}, Expected Version: #{expected_version.inspect})" }
    logger.debug(tags: [:data, :event_data, :write]) { event_data.pretty_inspect }
  end
end