Sha256: 8c3fcb87912f5627b0c6229334e5b77c59dab8f078bfe1630ed0978628b00732
Contents?: true
Size: 1.15 KB
Versions: 57
Compression:
Stored size: 1.15 KB
Contents
# frozen_string_literal: true
describe Deimos::Utils::SeekListener do
describe '#start_listener' do
let(:consumer) { instance_double(Kafka::Consumer) }
let(:handler) { class_double(Deimos::Utils::MessageBankHandler) }
before(:each) do
allow(handler).to receive(:start)
allow(consumer).to receive(:subscribe)
allow_any_instance_of(Phobos::Listener).to receive(:create_kafka_consumer).and_return(consumer)
allow_any_instance_of(Kafka::Client).to receive(:last_offset_for).and_return(100)
stub_const('Deimos::Utils::SeekListener::MAX_SEEK_RETRIES', 2)
end
it 'should seek offset' do
allow(consumer).to receive(:seek)
expect(consumer).to receive(:seek).once
seek_listener = described_class.new(handler: handler, group_id: 999, topic: 'test_topic')
seek_listener.start_listener
end
it 'should retry on errors when seeking offset' do
allow(consumer).to receive(:seek).and_raise(StandardError)
expect(consumer).to receive(:seek).twice
seek_listener = described_class.new(handler: handler, group_id: 999, topic: 'test_topic')
seek_listener.start_listener
end
end
end
Version data entries
57 entries across 57 versions & 1 rubygems