Merge pull request #6 from oozzal/add-message-topic

add topic method to Kafka::Message

.gitignore

@@ -0,0 +1,2 @@
+/lib/
+/shard.lock

README.md

@@ -66,7 +66,7 @@ consumer = Kafka::Consumer.new({"bootstrap.servers" => "localhost:9092", "broker
 consumer.subscribe("topic_name")
 consumer.each do |message|
   # message is an instance of Kafka::Message
-  puts message.payload
+  puts "#{String.new(message.topic)} -> #{String.new(message.payload)}
 end
 consumer.close
 ```

spec/integration/produce_consume_spec.cr

@@ -21,6 +21,7 @@ describe "Producing & Consuming" do
     raise "message is nil" if message.nil?
 
     String.new(message.payload).should eq({"foo" => "bar"}.to_json)
+    String.new(message.topic).should eq("foo")
   ensure
     consumer.try(&.close)
     producer.try(&.flush)

src/kafka/message.cr

@@ -6,12 +6,10 @@ module Kafka
     @err : RdKafka::Error?
     @offset : Int64?
     @timestamp : Int64?
+    @partition = LibRdKafka::PARTITION_UNASSIGNED
+    @rkt : LibRdKafka::Topic
     getter err, offset, key, payload, partition, timestamp
 
-    def initialize(@payload : Bytes, @key : Bytes)
-      @partition = LibRdKafka::PARTITION_UNASSIGNED
-    end
-
     def initialize(msg : LibRdKafka::Message)
       if msg.err != LibRdKafka::OK
         @err = RdKafka::Error.new(msg.err)
@@ -23,6 +21,11 @@ module Kafka
       @partition = msg.partition
       @offset = msg.offset
       @timestamp = msg.timestamp
+      @rkt = msg.rkt
+    end
+
+    def topic
+      LibRdKafka.topic_name(@rkt)
     end
   end
 end