KafkaRest

gem 'kafka_rest'

# Create a client
kafka = KafkaRest::Client.new(url: 'http://localhost:8080')

# List and update brokers
kafka.list_brokers

# List and update topics
kafka.list_topics

# Access single topic
topic = kafka.topic(name) # or kafka[name]

# Get a topic's metadata
topic.get

# List and update partitions for topic
topic.list_partitions

# Get a single topic partition by id
partition = topic.partition(id) # or topic[id]

# Produce a message to a topic
topic.produce(message)

# Messages can be produced in a number of formats
topic.produce('msg1')
topic.produce('msg1', 'msg2', 'msg3')
topic.produce(['msg1', 'msg2', 'msg3'])
topic.produce(key: 'key1', value: 'msg1')
topic.produce(partition: 0, value: 'msg1')
topic.produce({ key: 'key1', value: 'msg1'}, { partition: 0, value: 'msg2' })
topic.produce([{ key: 'key1', value: 'msg1'}, { partition: 0, value: 'msg2' }])

# Messages can also be produced from a partition
partition.produce(message)

# You can even produce messages asynchronously
partition.produce_async(message)

# To produce Avro encoded messages, parse a schema definition and pass to topic
schema = KafkaRest::Schema.parse('Foo.avsc')
topic = KafkaRest.topic('Foo', schema)

topic.produce(value: message)

# Create a consumer group
consumer = kafka.consumer('group1')

# Create an instance in the group, blocks and consumes in a loop after yielding
consumer.join do |instance|
  # Subscribe to a stream for topic
  instance.subscribe('topic1') do |stream|
  stream.on(:data) do |messages|
    # Your event-driven code
  end

  stream.on(:error) do |error|
    # Error handling
    if some_unrecoverable_exception?
      stream.shutdown! do
        # Optionally any cleanup code before stream is killed
      end
    end
  end
end

# The same consumer instance *CANNOT* be used to subscribe to multiple topics
consumer.join do |instance|
  instance.subscribe('foo') do |stream|
    stream.on(:data) { }
  end
  instance.subscribe('bar') do |stream|
    stream.on(:data) { }
  end
end