Merge pull request #8 from trekdemo/gs-add-async-producer-method

Add #async_producer and #shutdown

lib/fake/kafka.rb

@@ -30,9 +30,32 @@ def consumer(*options)
       Consumer.new(self)
     end
 
-    def producer(*)
+    # https://github.com/zendesk/ruby-kafka/blob/v1.0.0/lib/kafka/client.rb#L248-L261
+    # rubocop:disable Lint/UnusedMethodArgument, Metric/ParameterLists, Layout/LineLength
+    def producer(
+      compression_codec: nil,
+      compression_threshold: 1,
+      ack_timeout: 5,
+      required_acks: :all,
+      max_retries: 2,
+      retry_backoff: 1,
+      max_buffer_size: 1000,
+      max_buffer_bytesize: 10_000_000,
+      idempotent: false,
+      transactional: false,
+      transactional_id: nil,
+      transactional_timeout: 60
+    )
       Producer.new(self)
     end
+    # rubocop:enable all
+
+    # https://github.com/zendesk/ruby-kafka/blob/v1.0.0/lib/kafka/client.rb#L307
+    # rubocop:disable Lint/UnusedMethodArgument, Metric/ParameterLists, Layout/LineLength
+    def async_producer(delivery_interval: 0, delivery_threshold: 0, max_queue_size: 1000, max_retries: -1, retry_backoff: 0, **options)
+      producer(**options)
+    end
+    # rubocop:enable all
 
     # Used to clean in-memory data
     # Useful between test runs

lib/fake/kafka/producer.rb

@@ -13,4 +13,9 @@ def deliver_messages
       @kafka.deliver_message(value.to_s, **options)
     end
   end
+
+  # https://github.com/zendesk/ruby-kafka/blob/v1.0.0/lib/kafka/producer.rb#L285
+  def shutdown
+    # NOOP
+  end
 end

spec/fake/kafka_spec.rb

@@ -1,9 +1,33 @@
 RSpec.describe Fake::Kafka do
-  it "has a version number" do
+  subject(:kafka) { described_class.new }
+
+  it 'has a version number' do
     expect(Fake::Kafka::VERSION).not_to be nil
   end
 
-  it "does something useful" do
-    expect(false).to eq(true)
+  describe '#producer' do
+    subject(:producer) { kafka.producer }
+
+    it { is_expected.to be_a(described_class::Producer) }
+
+    context 'when invalid options are passed' do
+      specify do
+        expect { kafka.producer(foo: 'x') }
+          .to raise_error(ArgumentError)
+      end
+    end
+  end
+
+  describe '#async_producer' do
+    subject(:producer) { kafka.async_producer }
+
+    it { is_expected.to be_a(described_class::Producer) }
+
+    context 'when invalid options are passed' do
+      specify do
+        expect { kafka.async_producer(foo: 'x') }
+          .to raise_error(ArgumentError)
+      end
+    end
   end
 end