gem 'kcl-rb'

class RecordProcessor < Kcl::RecordProcessor
  def after_initialize(initialization_input)
    puts "SHARD_ID: #{initialization_input.shard_id}"
  end

  def process_records(records_input)
    puts "Current behind: #{records_input.millis_behind_latest}"
    records_input.records.each do |record|
      puts "Record: #{record}"
    end
  end

  def shutdown(shutdown_input)
    puts "Shutdown reason: #{shutdown_input.shutdown_reason}"

    if shutdown_input.shutdown_reason == Kcl::Workers::ShutdownReason::TERMINATE
      shutdown_input.record_checkpointer.update_checkpoint(nil)
    end
  end
end

class RecordProcessorFactory < Kcl::RecordProcessorFactory
  def create_processor
    RecordProcessor.new
  end
end

Kcl.configure do |config|
  config.dynamodb_endpoint = "http://localhost:4566"
  config.dynamodb_table_name = "kcl-rb-test"
  config.workers_health_table_name = "kcl-rb-test-health"
  config.kinesis_endpoint = "http://localhost:4566"
  config.kinesis_stream_name = "kcl-rb-test"
end

worker_id = 'kcl-worker'
factory = RecordProcessorFactory.new
Kcl::Worker.run(worker_id, factory)