Refactor cgomap setup

Author: edenhill

kafka/consumer.go

@@ -292,7 +292,6 @@ func NewConsumer(conf *ConfigMap) (*Consumer, error) {
 
 	c.handle.c = c
 	c.handle.setup()
-	c.handle.cgomap = make(map[int]cgoif)
 	c.handle.rkq = C.rd_kafka_queue_get_consumer(c.handle.rk)
 	if c.handle.rkq == nil {
 		// no cgrp (no group.id configured), revert to main queue.

kafka/handle.go

@@ -80,7 +80,7 @@ func (h *handle) String() string {
 func (h *handle) setup() {
 	h.rktCache = make(map[string]*C.rd_kafka_topic_t)
 	h.rktNameCache = make(map[*C.rd_kafka_topic_t]string)
-
+	h.cgomap = make(map[int]cgoif)
 	h.terminatedChan = make(chan string, 10)
 }
 

kafka/producer.go

@@ -241,7 +241,6 @@ func NewProducer(conf *ConfigMap) (*Producer, error) {
 	p.handle.p = p
 	p.handle.setup()
 	p.handle.rkq = C.rd_kafka_queue_get_main(p.handle.rk)
-	p.handle.cgomap = make(map[int]cgoif)
 	p.Events = make(chan Event, 1000000)
 	p.ProduceChannel = make(chan *Message, produceChannelSize)
 	p.pollerTermChan = make(chan bool)