Fix 'go vet' warnings for cgomap

Author: edenhill

kafka/consumer.go

@@ -295,7 +295,7 @@ func NewConsumer(conf *ConfigMap) (*Consumer, error) {
 
 	c.handle.c = c
 	c.handle.setup()
-	c.handle.cgomap = make(map[uintptr]cgoif)
+	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/event.go

@@ -203,7 +203,7 @@ out:
 
 				if rkmessage._private != nil {
 					// Find cgoif by id
-					cg, found := h.cgo_get((uintptr)(rkmessage._private))
+					cg, found := h.cgo_get((int)((uintptr)(rkmessage._private)))
 					if found {
 						cdr := cg.(cgo_dr)
 

kafka/handle.go

@@ -55,7 +55,7 @@ type handle struct {
 	// Maps C callbacks based on cgoid back to its Go object
 	cgo_lock   sync.Mutex
 	cgoid_next uintptr
-	cgomap     map[uintptr]cgoif
+	cgomap     map[int]cgoif
 
 	//
 	// producer
@@ -176,23 +176,23 @@ type cgo_dr struct {
 // unique id for the added entry.
 // Thread-safe.
 // FIXME: the uniquity of the id is questionable over time.
-func (h *handle) cgo_put(cg cgoif) (cgoid uintptr) {
+func (h *handle) cgo_put(cg cgoif) (cgoid int) {
 	h.cgo_lock.Lock()
 	defer h.cgo_lock.Unlock()
 
 	h.cgoid_next += 1
 	if h.cgoid_next == 0 {
 		h.cgoid_next += 1
 	}
-	cgoid = h.cgoid_next
+	cgoid = (int)(h.cgoid_next)
 	h.cgomap[cgoid] = cg
 	return cgoid
 }
 
 // cgo_get looks up cgoid in the cgo map, deletes the reference from the map
 // and returns the object, if found. Else returns nil, false.
 // Thread-safe.
-func (h *handle) cgo_get(cgoid uintptr) (cg cgoif, found bool) {
+func (h *handle) cgo_get(cgoid int) (cg cgoif, found bool) {
 	if cgoid == 0 {
 		return nil, false
 	}

kafka/producer.go

@@ -26,6 +26,13 @@ import (
 /*
 #include <stdlib.h>
 #include <librdkafka/rdkafka.h>
+
+rd_kafka_resp_err_t do_produce (rd_kafka_topic_t *rkt, int32_t partition,
+          int msg_flags,
+          void *val, size_t val_len, void *key, size_t key_len, uintptr_t cgoid) {
+  return rd_kafka_produce(rkt, partition, msg_flags, val, val_len, key, key_len,
+                          (void *)cgoid);
+}
 */
 import "C"
 
@@ -81,7 +88,7 @@ func (p *Producer) produce(msg *Message, msg_flags int, delivery_chan chan Event
 		}
 	}
 
-	var cgoid uintptr
+	var cgoid int
 
 	// Per-message state that needs to be retained through the C code:
 	//   delivery channel (if specified)
@@ -93,10 +100,11 @@ func (p *Producer) produce(msg *Message, msg_flags int, delivery_chan chan Event
 		cgoid = p.handle.cgo_put(cgo_dr{delivery_chan: delivery_chan, opaque: opaque})
 	}
 
-	r := int(C.rd_kafka_produce(c_rkt, C.int32_t(msg.TopicPartition.Partition),
+	r := int(C.do_produce(c_rkt, C.int32_t(msg.TopicPartition.Partition),
 		C.int(msg_flags)|C.RD_KAFKA_MSG_F_COPY,
 		unsafe.Pointer(valp), C.size_t(val_len),
-		unsafe.Pointer(keyp), C.size_t(key_len), unsafe.Pointer(cgoid)))
+		unsafe.Pointer(keyp), C.size_t(key_len),
+		(C.uintptr_t)(cgoid)))
 	if r == -1 {
 		if cgoid != 0 {
 			p.handle.cgo_get(cgoid)
@@ -232,7 +240,7 @@ 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[uintptr]cgoif)
+	p.handle.cgomap = make(map[int]cgoif)
 	p.Events = make(chan Event, 1000000)
 	p.ProduceChannel = make(chan *Message, produce_channel_size)
 	p.poller_term_chan = make(chan bool)