Contents

require "kafka/protocol/message"

module Kafka
  class MessageSet
    def initialize(messages)
      @messages = messages
    end

    def to_h
      hsh = {}
      
      @messages.each do |message|
        value, key = message.value, message.key
        topic, partition = message.topic, message.partition

        hsh[topic] ||= {}
        hsh[topic][partition] ||= []
        hsh[topic][partition] << Protocol::Message.new(value: value, key: key)
      end

      hsh
    end
  end
end

Version data entries

2 entries across 2 versions & 1 rubygems

Version	Path
ruby-kafka-0.1.0.pre.beta1	lib/kafka/message_set.rb
ruby-kafka-0.1.0.pre.alpha2	lib/kafka/message_set.rb