reiseburo
/
hermann
mirror of https://github.com/reiseburo/hermann
1
0
Fork 0
Code Issues Releases Wiki Activity

yield key and offset into Consumer#consume block

This commit is contained in:
Ben Osheroff 2015-06-17 12:18:03 -07:00
parent 60bc473fdd
commit 5b8dd6feef
2 changed files with 13 additions and 4 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
README.md
View File

@ -90,8 +90,8 @@ new_topic = 'other_topic'
the_consumer = Hermann::Consumer.new(topic, brokers: "localhost:9092", partition: 1)
the_consumer.consume(new_topic) do |msg| # can change topic with optional argument to .consume
puts "Recv: #{msg}"
the_consumer.consume(new_topic) do |msg, key, offset| # can change topic with optional argument to .consume
puts "Recv: #{msg}, key: #{key}, offset: #{offset}"
end
```

13
ext/hermann/hermann_rdkafka.c
View File

@ -264,9 +264,18 @@ static void msg_consume(rd_kafka_message_t *rkmessage, HermannInstanceConfig *cf
// Yield the data to the Consumer's block
if (rb_block_given_p()) {
VALUE value = rb_str_new((char *)rkmessage->payload, rkmessage->len);
VALUE key, data, offset;
data = rb_str_new((char *)rkmessage->payload, rkmessage->len);
offset = rb_ll2inum(rkmessage->offset);
if ( rkmessage->key_len > 0 )
key = rb_str_new((char*) rkmessage->key, (int)rkmessage->key_len);
else
key = Qnil;
rd_kafka_message_destroy(rkmessage);
rb_yield(value);
rb_yield_values(3, data, key, offset);
}
else {
if (DEBUG) {