Contents

require 'kafka'

class KafkaSession
  class Producer
    def initialize(name:, brokers:)
      @client = Kafka.new(
        seed_brokers: brokers,
        client_id:    name
      )
    end

    def publish(messages:, topic:)
      producer = @client.producer

      messages.each do |message|
        producer.produce(message, topic: topic)
      end

      producer.deliver_messages
    end
  end
end

Version data entries

2 entries across 2 versions & 1 rubygems

Version	Path
kafka_session-0.4.0	lib/kafka_session/producer.rb
kafka_session-0.4.0.pre.alpha	lib/kafka_session/producer.rb